|
|
@@ -1,1589 +0,0 @@
|
|
|
-package kubemodel
|
|
|
-
|
|
|
-import (
|
|
|
- "errors"
|
|
|
- "fmt"
|
|
|
- "time"
|
|
|
-
|
|
|
- "github.com/opencost/opencost/core/pkg/log"
|
|
|
- "github.com/opencost/opencost/core/pkg/model/kubemodel"
|
|
|
- "github.com/opencost/opencost/core/pkg/model/shared"
|
|
|
- "github.com/opencost/opencost/core/pkg/source"
|
|
|
-)
|
|
|
-
|
|
|
-const logTimeFmt string = "2006-01-02T15:04:05"
|
|
|
-
|
|
|
-type KubeModel struct {
|
|
|
- ds source.OpenCostDataSource
|
|
|
- clusterUID string
|
|
|
-}
|
|
|
-
|
|
|
-func NewKubeModel(clusterUID string, dataSource source.OpenCostDataSource) (*KubeModel, error) {
|
|
|
- if dataSource == nil {
|
|
|
- return nil, errors.New("OpenCostDataSource cannot be nil")
|
|
|
- }
|
|
|
-
|
|
|
- km := &KubeModel{
|
|
|
- ds: dataSource,
|
|
|
- clusterUID: clusterUID,
|
|
|
- }
|
|
|
-
|
|
|
- km.clusterUID = clusterUID
|
|
|
-
|
|
|
- log.Debugf("NewKubeModel(%s)", km.clusterUID)
|
|
|
-
|
|
|
- return km, nil
|
|
|
-}
|
|
|
-
|
|
|
-// ComputeKubeModel uses the CostModel instance to compute an KubeModelSet
|
|
|
-// for the window defined by the given start and end times. The KubeModels
|
|
|
-// returned are unaggregated (i.e. down to the container level).
|
|
|
-func (km *KubeModel) ComputeKubeModelSet(start, end time.Time) (*kubemodel.KubeModelSet, error) {
|
|
|
- // 1. Initialize new KubeModelSet for requested Window
|
|
|
- kms := kubemodel.NewKubeModelSet(start, end)
|
|
|
-
|
|
|
- // 2. Query CostModel for each set of objects
|
|
|
- var err error
|
|
|
-
|
|
|
- // 2.1 Compute Cluster
|
|
|
- err = km.computeCluster(kms, start, end)
|
|
|
- if err != nil {
|
|
|
- kms.Error(err)
|
|
|
- return kms, fmt.Errorf("error computing kubemodel.Cluster for (%s, %s): %w", start.Format(logTimeFmt), end.Format(logTimeFmt), err)
|
|
|
- }
|
|
|
-
|
|
|
- // 2.2 Compute Nodes
|
|
|
- err = km.computeNodes(kms, start, end)
|
|
|
- if err != nil {
|
|
|
- kms.Error(err)
|
|
|
- }
|
|
|
-
|
|
|
- // 2.3 Compute Namespaces
|
|
|
- err = km.computeNamespaces(kms, start, end)
|
|
|
- if err != nil {
|
|
|
- kms.Error(err)
|
|
|
- }
|
|
|
-
|
|
|
- // 2.5 Compute Pods
|
|
|
- err = km.computePods(kms, start, end)
|
|
|
- if err != nil {
|
|
|
- kms.Error(err)
|
|
|
- }
|
|
|
-
|
|
|
- // 2.6 Compute Deployments
|
|
|
- err = km.computeDeployments(kms, start, end)
|
|
|
- if err != nil {
|
|
|
- kms.Error(err)
|
|
|
- }
|
|
|
-
|
|
|
- // 2.7 Compute StatefulSets
|
|
|
- err = km.computeStatefulSets(kms, start, end)
|
|
|
- if err != nil {
|
|
|
- kms.Error(err)
|
|
|
- }
|
|
|
-
|
|
|
- // 2.8 Compute DaemonSets
|
|
|
- err = km.computeDaemonSets(kms, start, end)
|
|
|
- if err != nil {
|
|
|
- kms.Error(err)
|
|
|
- }
|
|
|
-
|
|
|
- // 2.9 Compute Jobs
|
|
|
- err = km.computeJobs(kms, start, end)
|
|
|
- if err != nil {
|
|
|
- kms.Error(err)
|
|
|
- }
|
|
|
-
|
|
|
- // 2.10 Compute CronJobs
|
|
|
- err = km.computeCronJobs(kms, start, end)
|
|
|
- if err != nil {
|
|
|
- kms.Error(err)
|
|
|
- }
|
|
|
-
|
|
|
- // 2.11 Compute ReplicaSets
|
|
|
- err = km.computeReplicaSets(kms, start, end)
|
|
|
- if err != nil {
|
|
|
- kms.Error(err)
|
|
|
- }
|
|
|
-
|
|
|
- // 2.12 Compute Containers
|
|
|
- err = km.computeContainers(kms, start, end)
|
|
|
- if err != nil {
|
|
|
- kms.Error(err)
|
|
|
- }
|
|
|
-
|
|
|
- // 2.13 Compute ResourceQuotas
|
|
|
- err = km.computeResourceQuotas(kms, start, end)
|
|
|
- if err != nil {
|
|
|
- kms.Error(err)
|
|
|
- }
|
|
|
-
|
|
|
- // 2.14 Compute Services
|
|
|
- err = km.computeServices(kms, start, end)
|
|
|
- if err != nil {
|
|
|
- kms.Error(err)
|
|
|
- }
|
|
|
-
|
|
|
- // 2.15 Compute PersistentVolumes
|
|
|
- err = km.computePersistentVolumes(kms, start, end)
|
|
|
- if err != nil {
|
|
|
- kms.Error(err)
|
|
|
- }
|
|
|
-
|
|
|
- // 2.16 Compute PersistentVolumeClaims
|
|
|
- err = km.computePersistentVolumeClaims(kms, start, end)
|
|
|
- if err != nil {
|
|
|
- kms.Error(err)
|
|
|
- }
|
|
|
-
|
|
|
- // 2.17 Compute DCGM Devices
|
|
|
- err = km.computeDCGMDevices(kms, start, end)
|
|
|
- if err != nil {
|
|
|
- kms.Error(err)
|
|
|
- }
|
|
|
-
|
|
|
- // 3. Mark KubeModelSet as completed
|
|
|
- kms.Metadata.CompletedAt = time.Now().UTC()
|
|
|
-
|
|
|
- return kms, nil
|
|
|
-}
|
|
|
-
|
|
|
-func (km *KubeModel) computeCluster(kms *kubemodel.KubeModelSet, start, end time.Time) error {
|
|
|
-
|
|
|
- grp := source.NewQueryGroup()
|
|
|
- metrics := km.ds.Metrics()
|
|
|
- clusterInfoResultFuture := source.WithGroup(grp, metrics.QueryClusterInfo(start, end))
|
|
|
- clusterUptimeResultFuture := source.WithGroup(grp, metrics.QueryClusterUptime(start, end))
|
|
|
-
|
|
|
- clusterMap := make(map[string]*kubemodel.Cluster)
|
|
|
-
|
|
|
- clusterInfoResult, _ := clusterInfoResultFuture.Await()
|
|
|
- for _, res := range clusterInfoResult {
|
|
|
- clusterMap[res.UID] = &kubemodel.Cluster{
|
|
|
- UID: res.UID,
|
|
|
- Provider: shared.ParseProvider(res.Provider),
|
|
|
- Account: res.AccountID,
|
|
|
- Name: res.Cluster,
|
|
|
- Region: res.Region,
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- clusterUptimeResult, _ := clusterUptimeResultFuture.Await()
|
|
|
- for _, res := range clusterUptimeResult {
|
|
|
- cluster, ok := clusterMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("cluster with UID '%s' has not been initialized to add uptime", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
- s, e := res.GetStartEnd(start, end, km.ds.Resolution())
|
|
|
- cluster.Start = s
|
|
|
- cluster.End = e
|
|
|
- }
|
|
|
-
|
|
|
- cluster, ok := clusterMap[km.clusterUID]
|
|
|
- if !ok {
|
|
|
- return fmt.Errorf("failed to compute cluster with UID '%s'", km.clusterUID)
|
|
|
- }
|
|
|
-
|
|
|
- kms.RegisterCluster(cluster)
|
|
|
-
|
|
|
- return nil
|
|
|
-}
|
|
|
-
|
|
|
-func (km *KubeModel) computeNodes(kms *kubemodel.KubeModelSet, start, end time.Time) error {
|
|
|
- grp := source.NewQueryGroup()
|
|
|
- metrics := km.ds.Metrics()
|
|
|
-
|
|
|
- nodeInfoResultFuture := source.WithGroup(grp, metrics.QueryNodeInfo(start, end))
|
|
|
- nodeUptimeResultFuture := source.WithGroup(grp, metrics.QueryNodeUptime(start, end))
|
|
|
- nodeLabelsResultFuture := source.WithGroup(grp, metrics.QueryNodeLabels(start, end))
|
|
|
- nodeResourceCapacitiesFuture := source.WithGroup(grp, metrics.QueryNodeResourceCapacities(start, end))
|
|
|
- nodeResourcesAllocatableFuture := source.WithGroup(grp, metrics.QueryNodeResourcesAllocatable(start, end))
|
|
|
-
|
|
|
- localStorageBytesFuture := source.WithGroup(grp, metrics.QueryKMLocalStorageBytes(start, end))
|
|
|
- localStorageUsedAvgFuture := source.WithGroup(grp, metrics.QueryKMLocalStorageUsedAvg(start, end))
|
|
|
- localStorageUsedMaxFuture := source.WithGroup(grp, metrics.QueryKMLocalStorageUsedMax(start, end))
|
|
|
-
|
|
|
- nodeMap := make(map[string]*kubemodel.Node)
|
|
|
-
|
|
|
- nodeInfoResult, _ := nodeInfoResultFuture.Await()
|
|
|
- for _, res := range nodeInfoResult {
|
|
|
- nodeMap[res.UID] = &kubemodel.Node{
|
|
|
- UID: res.UID,
|
|
|
- ProviderID: res.ProviderID,
|
|
|
- Name: res.Node,
|
|
|
- ResourceCapacities: make(kubemodel.ResourceQuantities),
|
|
|
- ResourcesAllocatable: make(kubemodel.ResourceQuantities),
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- nodeUptimeResult, _ := nodeUptimeResultFuture.Await()
|
|
|
- for _, res := range nodeUptimeResult {
|
|
|
- node, ok := nodeMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("node with UID '%s' has not been initialized to add uptime", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
- s, e := res.GetStartEnd(start, end, km.ds.Resolution())
|
|
|
- node.Start = s
|
|
|
- node.End = e
|
|
|
- }
|
|
|
-
|
|
|
- nodeResourceCapacitiesResult, _ := nodeResourceCapacitiesFuture.Await()
|
|
|
- for _, res := range nodeResourceCapacitiesResult {
|
|
|
- node, ok := nodeMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("node with UID '%s' has not been initialized to add resource capacities", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
- resource, unit, value := resourceUnitValue(res.Resource, res.Unit, res.Value)
|
|
|
- node.ResourceCapacities.Set(resource, unit, kubemodel.StatAvg, value)
|
|
|
- }
|
|
|
-
|
|
|
- nodeResourcesAllocatableResult, _ := nodeResourcesAllocatableFuture.Await()
|
|
|
- for _, res := range nodeResourcesAllocatableResult {
|
|
|
- node, ok := nodeMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("node with UID '%s' has not been initialized to add resources allocatable", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
- resource, unit, value := resourceUnitValue(res.Resource, res.Unit, res.Value)
|
|
|
- node.ResourcesAllocatable.Set(resource, unit, kubemodel.StatAvg, value)
|
|
|
- }
|
|
|
-
|
|
|
- nodeLabelsResult, _ := nodeLabelsResultFuture.Await()
|
|
|
- for _, res := range nodeLabelsResult {
|
|
|
- node, ok := nodeMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("node with UID '%s' has not been initialized to add labels", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
- node.Labels = res.Labels
|
|
|
- }
|
|
|
-
|
|
|
- localStorageBytesResult, _ := localStorageBytesFuture.Await()
|
|
|
- for _, res := range localStorageBytesResult {
|
|
|
- node, ok := nodeMap[res.UID]
|
|
|
- if ok {
|
|
|
- node.FileSystem.CapacityBytes = res.Value
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- localStorageUsedAvgResult, _ := localStorageUsedAvgFuture.Await()
|
|
|
- for _, res := range localStorageUsedAvgResult {
|
|
|
- node, ok := nodeMap[res.UID]
|
|
|
- if ok {
|
|
|
- node.FileSystem.UsageByteAvg = res.Value
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- localStorageUsedMaxResult, _ := localStorageUsedMaxFuture.Await()
|
|
|
- for _, res := range localStorageUsedMaxResult {
|
|
|
- node, ok := nodeMap[res.UID]
|
|
|
- if ok {
|
|
|
- node.FileSystem.UsageByteMax = res.Value
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- for _, node := range nodeMap {
|
|
|
- err := kms.RegisterNode(node)
|
|
|
- if err != nil {
|
|
|
- log.Warnf("Failed to register node: %s", err.Error())
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- return nil
|
|
|
-}
|
|
|
-
|
|
|
-// resourceUnitValue converts prometheus resource/unit strings from ResourceResult
|
|
|
-// into kubemodel types, applying any necessary unit conversions.
|
|
|
-func resourceUnitValue(resource, unit string, value float64) (kubemodel.Resource, kubemodel.Unit, float64) {
|
|
|
- switch resource {
|
|
|
- case "cpu":
|
|
|
- return kubemodel.ResourceCPU, kubemodel.UnitCore, value
|
|
|
- case "memory":
|
|
|
- return kubemodel.ResourceMemory, kubemodel.UnitByte, value
|
|
|
- default:
|
|
|
- return kubemodel.Resource(resource), kubemodel.Unit(unit), value
|
|
|
- }
|
|
|
-}
|
|
|
-
|
|
|
-func (km *KubeModel) computeNamespaces(kms *kubemodel.KubeModelSet, start, end time.Time) error {
|
|
|
- grp := source.NewQueryGroup()
|
|
|
- metrics := km.ds.Metrics()
|
|
|
-
|
|
|
- nsInfoResultFuture := source.WithGroup(grp, metrics.QueryNamespaceInfo(start, end))
|
|
|
- nsUptimeResultFuture := source.WithGroup(grp, metrics.QueryNamespaceUptime(start, end))
|
|
|
- nsLabelsResultFuture := source.WithGroup(grp, metrics.QueryNamespaceLabels(start, end))
|
|
|
- nsAnnosResultFuture := source.WithGroup(grp, metrics.QueryNamespaceAnnotations(start, end))
|
|
|
-
|
|
|
- nsMap := make(map[string]*kubemodel.Namespace)
|
|
|
-
|
|
|
- // Initialize namespaces from info
|
|
|
- nsInfoResult, _ := nsInfoResultFuture.Await()
|
|
|
- for _, res := range nsInfoResult {
|
|
|
- nsMap[res.UID] = &kubemodel.Namespace{
|
|
|
- UID: res.UID,
|
|
|
- Name: res.Namespace,
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- nsUptimeResult, _ := nsUptimeResultFuture.Await()
|
|
|
- for _, res := range nsUptimeResult {
|
|
|
- ns, ok := nsMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("namespace with UID '%s' has not been initialized to add uptime", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
- s, e := res.GetStartEnd(start, end, km.ds.Resolution())
|
|
|
- ns.Start = s
|
|
|
- ns.End = e
|
|
|
- }
|
|
|
-
|
|
|
- nsLabelsResult, _ := nsLabelsResultFuture.Await()
|
|
|
- for _, res := range nsLabelsResult {
|
|
|
- ns, ok := nsMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("namespace with UID '%s' has not been initialized to add labels", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
- ns.Labels = res.Labels
|
|
|
- }
|
|
|
-
|
|
|
- nsAnnosResult, _ := nsAnnosResultFuture.Await()
|
|
|
- for _, res := range nsAnnosResult {
|
|
|
- ns, ok := nsMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("namespace with UID '%s' has not been initialized to add annotations", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
- ns.Annotations = res.Annotations
|
|
|
- }
|
|
|
-
|
|
|
- for _, namespace := range nsMap {
|
|
|
- err := kms.RegisterNamespace(namespace)
|
|
|
- if err != nil {
|
|
|
- log.Warnf("Failed to register namespace: %s", err.Error())
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- return nil
|
|
|
-}
|
|
|
-
|
|
|
-func (km *KubeModel) computePods(kms *kubemodel.KubeModelSet, start, end time.Time) error {
|
|
|
- grp := source.NewQueryGroup()
|
|
|
- metrics := km.ds.Metrics()
|
|
|
-
|
|
|
- podInfoResultFuture := source.WithGroup(grp, metrics.QueryPodInfo(start, end))
|
|
|
- podUptimeResultFuture := source.WithGroup(grp, metrics.QueryPodUptime(start, end))
|
|
|
- podOwnerResultFuture := source.WithGroup(grp, metrics.QueryPodOwners(start, end))
|
|
|
- podPVCVolumesResultFuture := source.WithGroup(grp, metrics.QueryPodPVCVolumes(start, end))
|
|
|
- podLabelsResultFuture := source.WithGroup(grp, metrics.QueryPodLabels(start, end))
|
|
|
- podAnnosResultFuture := source.WithGroup(grp, metrics.QueryPodAnnotations(start, end))
|
|
|
-
|
|
|
- podNetworkEgressBytesResultFuture := source.WithGroup(grp, metrics.QueryPodNetworkEgressBytes(start, end))
|
|
|
- podNetworkIngressBytesResultFuture := source.WithGroup(grp, metrics.QueryPodNetworkIngressBytes(start, end))
|
|
|
-
|
|
|
- podMap := make(map[string]*kubemodel.Pod)
|
|
|
-
|
|
|
- podInfoResult, _ := podInfoResultFuture.Await()
|
|
|
- for _, res := range podInfoResult {
|
|
|
- podMap[res.UID] = &kubemodel.Pod{
|
|
|
- UID: res.UID,
|
|
|
- Name: res.Pod,
|
|
|
- NamespaceUID: res.NamespaceUID,
|
|
|
- NodeUID: res.NodeUID,
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- podUptimeResult, _ := podUptimeResultFuture.Await()
|
|
|
- for _, res := range podUptimeResult {
|
|
|
- pod, ok := podMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("pod with UID '%s' has not been initialized to add uptime", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
- s, e := res.GetStartEnd(start, end, km.ds.Resolution())
|
|
|
- pod.Start = s
|
|
|
- pod.End = e
|
|
|
- }
|
|
|
-
|
|
|
- podOwnersResult, _ := podOwnerResultFuture.Await()
|
|
|
- for _, res := range podOwnersResult {
|
|
|
- pod, ok := podMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("pod with UID '%s' has not been initialized to add labels", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
- pod.Owners = append(pod.Owners, kubemodel.Owner{
|
|
|
- UID: res.OwnerUID,
|
|
|
- Kind: kubemodel.ParseOwnerKind(res.OwnerKind),
|
|
|
- Controller: res.Controller,
|
|
|
- })
|
|
|
- }
|
|
|
-
|
|
|
- podPVCVolumesResult, _ := podPVCVolumesResultFuture.Await()
|
|
|
- for _, res := range podPVCVolumesResult {
|
|
|
- pod, ok := podMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("pod with UID '%s' has not been initialized to add PVC volumes", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
- pod.PVCVolumes = append(pod.PVCVolumes, kubemodel.PodPVCVolume{
|
|
|
- Name: res.PodVolumeName,
|
|
|
- PersistentVolumeClaimUID: res.PVCUID,
|
|
|
- })
|
|
|
- }
|
|
|
-
|
|
|
- podLabelsResult, _ := podLabelsResultFuture.Await()
|
|
|
- for _, res := range podLabelsResult {
|
|
|
- pod, ok := podMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("pod with UID '%s' has not been initialized to add labels", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
- pod.Labels = res.Labels
|
|
|
- }
|
|
|
-
|
|
|
- podAnnosResult, _ := podAnnosResultFuture.Await()
|
|
|
- for _, res := range podAnnosResult {
|
|
|
- pod, ok := podMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("pod with UID '%s' has not been initialized to add annotations", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
- pod.Annotations = res.Annotations
|
|
|
- }
|
|
|
-
|
|
|
- appendDetail := func(uid string, dir kubemodel.TrafficDirection, tt kubemodel.TrafficType, isNatGateway bool, endpoint string, bytes float64) {
|
|
|
- pod, ok := podMap[uid]
|
|
|
- if !ok || bytes <= 0 {
|
|
|
- return
|
|
|
- }
|
|
|
- pod.NetworkTrafficDetails = append(pod.NetworkTrafficDetails, kubemodel.NetworkTrafficDetail{
|
|
|
- PodUID: uid,
|
|
|
- TrafficDirection: dir,
|
|
|
- TrafficType: tt,
|
|
|
- IsNatGateway: isNatGateway,
|
|
|
- Endpoint: endpoint,
|
|
|
- Bytes: bytes,
|
|
|
- })
|
|
|
- }
|
|
|
-
|
|
|
- networkTrafficType := func(res *source.PodNetworkBytesResult) (kubemodel.TrafficType, bool) {
|
|
|
- if res.Internet {
|
|
|
- return kubemodel.TrafficTypeInternet, true
|
|
|
- }
|
|
|
- if !res.SameRegion {
|
|
|
- return kubemodel.TrafficTypeCrossRegion, true
|
|
|
- }
|
|
|
- if !res.SameZone {
|
|
|
- return kubemodel.TrafficTypeCrossZone, true
|
|
|
- }
|
|
|
- return "", false
|
|
|
- }
|
|
|
-
|
|
|
- podNetworkEgressResult, _ := podNetworkEgressBytesResultFuture.Await()
|
|
|
- for _, res := range podNetworkEgressResult {
|
|
|
- tt, ok := networkTrafficType(res)
|
|
|
- if !ok {
|
|
|
- continue
|
|
|
- }
|
|
|
- appendDetail(res.UID, kubemodel.TrafficDirectionEgress, tt, res.NatGateway, res.Service, res.Value)
|
|
|
- }
|
|
|
-
|
|
|
- podNetworkIngressResult, _ := podNetworkIngressBytesResultFuture.Await()
|
|
|
- for _, res := range podNetworkIngressResult {
|
|
|
- tt, ok := networkTrafficType(res)
|
|
|
- if !ok {
|
|
|
- continue
|
|
|
- }
|
|
|
- appendDetail(res.UID, kubemodel.TrafficDirectionIngress, tt, res.NatGateway, res.Service, res.Value)
|
|
|
- }
|
|
|
-
|
|
|
- for _, pod := range podMap {
|
|
|
- err := kms.RegisterPod(pod)
|
|
|
- if err != nil {
|
|
|
- log.Warnf("Failed to register pod: %s", err.Error())
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- return nil
|
|
|
-}
|
|
|
-
|
|
|
-func (km *KubeModel) computeDeployments(kms *kubemodel.KubeModelSet, start, end time.Time) error {
|
|
|
- grp := source.NewQueryGroup()
|
|
|
- metrics := km.ds.Metrics()
|
|
|
-
|
|
|
- deploymentInfoResultFuture := source.WithGroup(grp, metrics.QueryDeploymentInfo(start, end))
|
|
|
- deploymentUptimeResultFuture := source.WithGroup(grp, metrics.QueryDeploymentUptime(start, end))
|
|
|
- deploymentLabelsResultFuture := source.WithGroup(grp, metrics.QueryDeploymentLabels(start, end))
|
|
|
- deploymentAnnotationsResultFuture := source.WithGroup(grp, metrics.QueryDeploymentAnnotations(start, end))
|
|
|
- deploymentMatchLabelsResultFuture := source.WithGroup(grp, metrics.QueryDeploymentMatchLabels(start, end))
|
|
|
-
|
|
|
- deploymentMap := make(map[string]*kubemodel.Deployment)
|
|
|
-
|
|
|
- deploymentInfoResult, _ := deploymentInfoResultFuture.Await()
|
|
|
- for _, res := range deploymentInfoResult {
|
|
|
- deploymentMap[res.UID] = &kubemodel.Deployment{
|
|
|
- UID: res.UID,
|
|
|
- Name: res.Deployment,
|
|
|
- NamespaceUID: res.NamespaceUID,
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- deploymentUptimeResult, _ := deploymentUptimeResultFuture.Await()
|
|
|
- for _, res := range deploymentUptimeResult {
|
|
|
- deployment, ok := deploymentMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("deployment with UID '%s' has not been initialized to add uptime", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
- s, e := res.GetStartEnd(start, end, km.ds.Resolution())
|
|
|
- deployment.Start = s
|
|
|
- deployment.End = e
|
|
|
- }
|
|
|
-
|
|
|
- deploymentLabelsResult, _ := deploymentLabelsResultFuture.Await()
|
|
|
- for _, res := range deploymentLabelsResult {
|
|
|
- deployment, ok := deploymentMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("deployment with UID '%s' has not been initialized to add labels", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
- deployment.Labels = res.Labels
|
|
|
- }
|
|
|
-
|
|
|
- deploymentAnnotationsResult, _ := deploymentAnnotationsResultFuture.Await()
|
|
|
- for _, res := range deploymentAnnotationsResult {
|
|
|
- deployment, ok := deploymentMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("deployment with UID '%s' has not been initialized to add annotations", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
- deployment.Annotations = res.Annotations
|
|
|
- }
|
|
|
-
|
|
|
- deploymentMatchLabelsResult, _ := deploymentMatchLabelsResultFuture.Await()
|
|
|
- for _, res := range deploymentMatchLabelsResult {
|
|
|
- deployment, ok := deploymentMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("deployment with UID '%s' has not been initialized to add match labels", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
- deployment.MatchLabels = res.Labels
|
|
|
- }
|
|
|
-
|
|
|
- for _, deployment := range deploymentMap {
|
|
|
- err := kms.RegisterDeployment(deployment)
|
|
|
- if err != nil {
|
|
|
- log.Warnf("Failed to register deployment: %s", err.Error())
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- return nil
|
|
|
-}
|
|
|
-
|
|
|
-func (km *KubeModel) computeStatefulSets(kms *kubemodel.KubeModelSet, start, end time.Time) error {
|
|
|
- grp := source.NewQueryGroup()
|
|
|
- metrics := km.ds.Metrics()
|
|
|
-
|
|
|
- statefulSetInfoResultFuture := source.WithGroup(grp, metrics.QueryStatefulSetInfo(start, end))
|
|
|
- statefulSetUptimeResultFuture := source.WithGroup(grp, metrics.QueryStatefulSetUptime(start, end))
|
|
|
- statefulSetLabelsResultFuture := source.WithGroup(grp, metrics.QueryStatefulSetLabels(start, end))
|
|
|
- statefulSetAnnotationsResultFuture := source.WithGroup(grp, metrics.QueryStatefulSetAnnotations(start, end))
|
|
|
- statefulSetMatchLabelsResultFuture := source.WithGroup(grp, metrics.QueryStatefulSetMatchLabels(start, end))
|
|
|
-
|
|
|
- statefulSetMap := make(map[string]*kubemodel.StatefulSet)
|
|
|
-
|
|
|
- statefulSetInfoResult, _ := statefulSetInfoResultFuture.Await()
|
|
|
- for _, res := range statefulSetInfoResult {
|
|
|
- statefulSetMap[res.UID] = &kubemodel.StatefulSet{
|
|
|
- UID: res.UID,
|
|
|
- Name: res.StatefulSet,
|
|
|
- NamespaceUID: res.NamespaceUID,
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- statefulSetUptimeResult, _ := statefulSetUptimeResultFuture.Await()
|
|
|
- for _, res := range statefulSetUptimeResult {
|
|
|
- statefulSet, ok := statefulSetMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("statefulset with UID '%s' has not been initialized to add uptime", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
- s, e := res.GetStartEnd(start, end, km.ds.Resolution())
|
|
|
- statefulSet.Start = s
|
|
|
- statefulSet.End = e
|
|
|
- }
|
|
|
-
|
|
|
- statefulSetLabelsResult, _ := statefulSetLabelsResultFuture.Await()
|
|
|
- for _, res := range statefulSetLabelsResult {
|
|
|
- statefulSet, ok := statefulSetMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("statefulset with UID '%s' has not been initialized to add labels", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
- statefulSet.Labels = res.Labels
|
|
|
- }
|
|
|
-
|
|
|
- statefulSetAnnotationsResult, _ := statefulSetAnnotationsResultFuture.Await()
|
|
|
- for _, res := range statefulSetAnnotationsResult {
|
|
|
- statefulSet, ok := statefulSetMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("statefulset with UID '%s' has not been initialized to add annotations", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
- statefulSet.Annotations = res.Annotations
|
|
|
- }
|
|
|
-
|
|
|
- statefulSetMatchLabelsResult, _ := statefulSetMatchLabelsResultFuture.Await()
|
|
|
- for _, res := range statefulSetMatchLabelsResult {
|
|
|
- statefulSet, ok := statefulSetMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("statefulset with UID '%s' has not been initialized to add match labels", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
- statefulSet.MatchLabels = res.Labels
|
|
|
- }
|
|
|
-
|
|
|
- for _, statefulSet := range statefulSetMap {
|
|
|
- err := kms.RegisterStatefulSet(statefulSet)
|
|
|
- if err != nil {
|
|
|
- log.Warnf("Failed to register statefulset: %s", err.Error())
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- return nil
|
|
|
-}
|
|
|
-
|
|
|
-func (km *KubeModel) computeDaemonSets(kms *kubemodel.KubeModelSet, start, end time.Time) error {
|
|
|
- grp := source.NewQueryGroup()
|
|
|
- metrics := km.ds.Metrics()
|
|
|
-
|
|
|
- daemonSetInfoResultFuture := source.WithGroup(grp, metrics.QueryDaemonSetInfo(start, end))
|
|
|
- daemonSetUptimeResultFuture := source.WithGroup(grp, metrics.QueryDaemonSetUptime(start, end))
|
|
|
- daemonSetLabelsResultFuture := source.WithGroup(grp, metrics.QueryDaemonSetLabels(start, end))
|
|
|
- daemonSetAnnotationsResultFuture := source.WithGroup(grp, metrics.QueryDaemonSetAnnotations(start, end))
|
|
|
-
|
|
|
- daemonSetMap := make(map[string]*kubemodel.DaemonSet)
|
|
|
-
|
|
|
- daemonSetInfoResult, _ := daemonSetInfoResultFuture.Await()
|
|
|
- for _, res := range daemonSetInfoResult {
|
|
|
- daemonSetMap[res.UID] = &kubemodel.DaemonSet{
|
|
|
- UID: res.UID,
|
|
|
- Name: res.DaemonSet,
|
|
|
- NamespaceUID: res.NamespaceUID,
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- daemonSetUptimeResult, _ := daemonSetUptimeResultFuture.Await()
|
|
|
- for _, res := range daemonSetUptimeResult {
|
|
|
- daemonSet, ok := daemonSetMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("daemonset with UID '%s' has not been initialized to add uptime", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
- s, e := res.GetStartEnd(start, end, km.ds.Resolution())
|
|
|
- daemonSet.Start = s
|
|
|
- daemonSet.End = e
|
|
|
- }
|
|
|
-
|
|
|
- daemonSetLabelsResult, _ := daemonSetLabelsResultFuture.Await()
|
|
|
- for _, res := range daemonSetLabelsResult {
|
|
|
- daemonSet, ok := daemonSetMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("daemonset with UID '%s' has not been initialized to add labels", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
- daemonSet.Labels = res.Labels
|
|
|
- }
|
|
|
-
|
|
|
- daemonSetAnnotationsResult, _ := daemonSetAnnotationsResultFuture.Await()
|
|
|
- for _, res := range daemonSetAnnotationsResult {
|
|
|
- daemonSet, ok := daemonSetMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("daemonset with UID '%s' has not been initialized to add annotations", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
- daemonSet.Annotations = res.Annotations
|
|
|
- }
|
|
|
-
|
|
|
- for _, daemonSet := range daemonSetMap {
|
|
|
- err := kms.RegisterDaemonSet(daemonSet)
|
|
|
- if err != nil {
|
|
|
- log.Warnf("Failed to register daemonset: %s", err.Error())
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- return nil
|
|
|
-}
|
|
|
-
|
|
|
-func (km *KubeModel) computeJobs(kms *kubemodel.KubeModelSet, start, end time.Time) error {
|
|
|
- grp := source.NewQueryGroup()
|
|
|
- metrics := km.ds.Metrics()
|
|
|
-
|
|
|
- jobInfoResultFuture := source.WithGroup(grp, metrics.QueryJobInfo(start, end))
|
|
|
- jobUptimeResultFuture := source.WithGroup(grp, metrics.QueryJobUptime(start, end))
|
|
|
- jobLabelsResultFuture := source.WithGroup(grp, metrics.QueryJobLabels(start, end))
|
|
|
- jobAnnotationsResultFuture := source.WithGroup(grp, metrics.QueryJobAnnotations(start, end))
|
|
|
-
|
|
|
- jobMap := make(map[string]*kubemodel.Job)
|
|
|
-
|
|
|
- jobInfoResult, _ := jobInfoResultFuture.Await()
|
|
|
- for _, res := range jobInfoResult {
|
|
|
- jobMap[res.UID] = &kubemodel.Job{
|
|
|
- UID: res.UID,
|
|
|
- Name: res.Job,
|
|
|
- NamespaceUID: res.NamespaceUID,
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- jobUptimeResult, _ := jobUptimeResultFuture.Await()
|
|
|
- for _, res := range jobUptimeResult {
|
|
|
- job, ok := jobMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("job with UID '%s' has not been initialized to add uptime", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
- s, e := res.GetStartEnd(start, end, km.ds.Resolution())
|
|
|
- job.Start = s
|
|
|
- job.End = e
|
|
|
- }
|
|
|
-
|
|
|
- jobLabelsResult, _ := jobLabelsResultFuture.Await()
|
|
|
- for _, res := range jobLabelsResult {
|
|
|
- job, ok := jobMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("job with UID '%s' has not been initialized to add labels", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
- job.Labels = res.Labels
|
|
|
- }
|
|
|
-
|
|
|
- jobAnnotationsResult, _ := jobAnnotationsResultFuture.Await()
|
|
|
- for _, res := range jobAnnotationsResult {
|
|
|
- job, ok := jobMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("job with UID '%s' has not been initialized to add annotations", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
- job.Annotations = res.Annotations
|
|
|
- }
|
|
|
-
|
|
|
- for _, job := range jobMap {
|
|
|
- err := kms.RegisterJob(job)
|
|
|
- if err != nil {
|
|
|
- log.Warnf("Failed to register job: %s", err.Error())
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- return nil
|
|
|
-}
|
|
|
-
|
|
|
-func (km *KubeModel) computeCronJobs(kms *kubemodel.KubeModelSet, start, end time.Time) error {
|
|
|
- grp := source.NewQueryGroup()
|
|
|
- metrics := km.ds.Metrics()
|
|
|
-
|
|
|
- cronJobInfoResultFuture := source.WithGroup(grp, metrics.QueryCronJobInfo(start, end))
|
|
|
- cronJobUptimeResultFuture := source.WithGroup(grp, metrics.QueryCronJobUptime(start, end))
|
|
|
- cronJobLabelsResultFuture := source.WithGroup(grp, metrics.QueryCronJobLabels(start, end))
|
|
|
- cronJobAnnotationsResultFuture := source.WithGroup(grp, metrics.QueryCronJobAnnotations(start, end))
|
|
|
-
|
|
|
- cronJobMap := make(map[string]*kubemodel.CronJob)
|
|
|
-
|
|
|
- cronJobInfoResult, _ := cronJobInfoResultFuture.Await()
|
|
|
- for _, res := range cronJobInfoResult {
|
|
|
- cronJobMap[res.UID] = &kubemodel.CronJob{
|
|
|
- UID: res.UID,
|
|
|
- Name: res.CronJob,
|
|
|
- NamespaceUID: res.NamespaceUID,
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- cronJobUptimeResult, _ := cronJobUptimeResultFuture.Await()
|
|
|
- for _, res := range cronJobUptimeResult {
|
|
|
- cronJob, ok := cronJobMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("cronjob with UID '%s' has not been initialized to add uptime", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
- s, e := res.GetStartEnd(start, end, km.ds.Resolution())
|
|
|
- cronJob.Start = s
|
|
|
- cronJob.End = e
|
|
|
- }
|
|
|
-
|
|
|
- cronJobLabelsResult, _ := cronJobLabelsResultFuture.Await()
|
|
|
- for _, res := range cronJobLabelsResult {
|
|
|
- cronJob, ok := cronJobMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("cronjob with UID '%s' has not been initialized to add labels", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
- cronJob.Labels = res.Labels
|
|
|
- }
|
|
|
-
|
|
|
- cronJobAnnotationsResult, _ := cronJobAnnotationsResultFuture.Await()
|
|
|
- for _, res := range cronJobAnnotationsResult {
|
|
|
- cronJob, ok := cronJobMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("cronjob with UID '%s' has not been initialized to add annotations", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
- cronJob.Annotations = res.Annotations
|
|
|
- }
|
|
|
-
|
|
|
- for _, cronJob := range cronJobMap {
|
|
|
- err := kms.RegisterCronJob(cronJob)
|
|
|
- if err != nil {
|
|
|
- log.Warnf("Failed to register cronjob: %s", err.Error())
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- return nil
|
|
|
-}
|
|
|
-
|
|
|
-func (km *KubeModel) computeReplicaSets(kms *kubemodel.KubeModelSet, start, end time.Time) error {
|
|
|
- grp := source.NewQueryGroup()
|
|
|
- metrics := km.ds.Metrics()
|
|
|
-
|
|
|
- replicaSetInfoResultFuture := source.WithGroup(grp, metrics.QueryReplicaSetInfo(start, end))
|
|
|
- replicaSetUptimeResultFuture := source.WithGroup(grp, metrics.QueryReplicaSetUptime(start, end))
|
|
|
- replicaSetOwnerResultFuture := source.WithGroup(grp, metrics.QueryReplicaSetOwners(start, end))
|
|
|
- replicaSetLabelsResultFuture := source.WithGroup(grp, metrics.QueryReplicaSetLabels(start, end))
|
|
|
- replicaSetAnnotationsResultFuture := source.WithGroup(grp, metrics.QueryReplicaSetAnnotations(start, end))
|
|
|
-
|
|
|
- replicaSetMap := make(map[string]*kubemodel.ReplicaSet)
|
|
|
-
|
|
|
- replicaSetInfoResult, _ := replicaSetInfoResultFuture.Await()
|
|
|
- for _, res := range replicaSetInfoResult {
|
|
|
- replicaSetMap[res.UID] = &kubemodel.ReplicaSet{
|
|
|
- UID: res.UID,
|
|
|
- Name: res.ReplicaSet,
|
|
|
- NamespaceUID: res.NamespaceUID,
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- replicaSetUptimeResult, _ := replicaSetUptimeResultFuture.Await()
|
|
|
- for _, res := range replicaSetUptimeResult {
|
|
|
- replicaSet, ok := replicaSetMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("replicaset with UID '%s' has not been initialized to add uptime", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
- s, e := res.GetStartEnd(start, end, km.ds.Resolution())
|
|
|
- replicaSet.Start = s
|
|
|
- replicaSet.End = e
|
|
|
- }
|
|
|
-
|
|
|
- replicaSetOwnersResult, _ := replicaSetOwnerResultFuture.Await()
|
|
|
- for _, res := range replicaSetOwnersResult {
|
|
|
- replicaSet, ok := replicaSetMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("replicaset with UID '%s' has not been initialized to add owner", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
- replicaSet.Owners = append(replicaSet.Owners, kubemodel.Owner{
|
|
|
- UID: res.OwnerUID,
|
|
|
- Kind: kubemodel.ParseOwnerKind(res.OwnerKind),
|
|
|
- Controller: res.Controller,
|
|
|
- })
|
|
|
- }
|
|
|
-
|
|
|
- replicaSetLabelsResult, _ := replicaSetLabelsResultFuture.Await()
|
|
|
- for _, res := range replicaSetLabelsResult {
|
|
|
- replicaSet, ok := replicaSetMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("replicaset with UID '%s' has not been initialized to add labels", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
- replicaSet.Labels = res.Labels
|
|
|
- }
|
|
|
-
|
|
|
- replicaSetAnnotationsResult, _ := replicaSetAnnotationsResultFuture.Await()
|
|
|
- for _, res := range replicaSetAnnotationsResult {
|
|
|
- replicaSet, ok := replicaSetMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("replicaset with UID '%s' has not been initialized to add annotations", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
- replicaSet.Annotations = res.Annotations
|
|
|
- }
|
|
|
-
|
|
|
- for _, replicaSet := range replicaSetMap {
|
|
|
- err := kms.RegisterReplicaSet(replicaSet)
|
|
|
- if err != nil {
|
|
|
- log.Warnf("Failed to register replicaset: %s", err.Error())
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- return nil
|
|
|
-}
|
|
|
-
|
|
|
-func (km *KubeModel) computeContainers(kms *kubemodel.KubeModelSet, start, end time.Time) error {
|
|
|
- grp := source.NewQueryGroup()
|
|
|
- metrics := km.ds.Metrics()
|
|
|
-
|
|
|
- containerUptimeFuture := source.WithGroup(grp, metrics.QueryContainerUptime(start, end))
|
|
|
-
|
|
|
- containerResourceRequestsFuture := source.WithGroup(grp, metrics.QueryContainerResourceRequests(start, end))
|
|
|
- containerResourceLimitsFuture := source.WithGroup(grp, metrics.QueryContainerResourceLimits(start, end))
|
|
|
-
|
|
|
- cpuCoresAllocatedFuture := source.WithGroup(grp, metrics.QueryCPUCoresAllocated(start, end))
|
|
|
- cpuUsageAvgFuture := source.WithGroup(grp, metrics.QueryCPUUsageAvg(start, end))
|
|
|
- cpuUsageMaxFuture := source.WithGroup(grp, metrics.QueryCPUUsageMax(start, end))
|
|
|
-
|
|
|
- ramBytesAllocatedFuture := source.WithGroup(grp, metrics.QueryRAMBytesAllocated(start, end))
|
|
|
- ramUsageAvgFuture := source.WithGroup(grp, metrics.QueryRAMUsageAvg(start, end))
|
|
|
- ramUsageMaxFuture := source.WithGroup(grp, metrics.QueryRAMUsageMax(start, end))
|
|
|
-
|
|
|
- type containerKey struct {
|
|
|
- podUID string
|
|
|
- name string
|
|
|
- }
|
|
|
-
|
|
|
- containerMap := make(map[containerKey]*kubemodel.Container)
|
|
|
-
|
|
|
- containerUptimeResult, _ := containerUptimeFuture.Await()
|
|
|
- for _, res := range containerUptimeResult {
|
|
|
- key := containerKey{podUID: res.UID, name: res.Container}
|
|
|
- s, e := res.GetStartEnd(start, end, km.ds.Resolution())
|
|
|
- containerMap[key] = &kubemodel.Container{
|
|
|
- PodUID: res.UID,
|
|
|
- Name: res.Container,
|
|
|
- ResourceRequests: make(kubemodel.ResourceQuantities),
|
|
|
- ResourceLimits: make(kubemodel.ResourceQuantities),
|
|
|
- Start: s,
|
|
|
- End: e,
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- containerResourceRequestsResult, _ := containerResourceRequestsFuture.Await()
|
|
|
- for _, res := range containerResourceRequestsResult {
|
|
|
- key := containerKey{podUID: res.UID, name: res.Container}
|
|
|
- container, ok := containerMap[key]
|
|
|
- if !ok {
|
|
|
- log.Warnf("container %s/%s has not been initialized to add resource requests", res.UID, res.Container)
|
|
|
- continue
|
|
|
- }
|
|
|
- resource, unit, value := resourceUnitValue(res.Resource, res.Unit, res.Value)
|
|
|
- container.ResourceRequests.Set(resource, unit, kubemodel.StatAvg, value)
|
|
|
- }
|
|
|
-
|
|
|
- containerResourceLimitsResult, _ := containerResourceLimitsFuture.Await()
|
|
|
- for _, res := range containerResourceLimitsResult {
|
|
|
- key := containerKey{podUID: res.UID, name: res.Container}
|
|
|
- container, ok := containerMap[key]
|
|
|
- if !ok {
|
|
|
- log.Warnf("container %s/%s has not been initialized to add resource limits", res.UID, res.Container)
|
|
|
- continue
|
|
|
- }
|
|
|
- resource, unit, value := resourceUnitValue(res.Resource, res.Unit, res.Value)
|
|
|
- container.ResourceLimits.Set(resource, unit, kubemodel.StatAvg, value)
|
|
|
- }
|
|
|
-
|
|
|
- cpuCoresAllocatedResult, _ := cpuCoresAllocatedFuture.Await()
|
|
|
- for _, res := range cpuCoresAllocatedResult {
|
|
|
- key := containerKey{podUID: res.UID, name: res.Container}
|
|
|
- container, ok := containerMap[key]
|
|
|
- if !ok {
|
|
|
- log.Warnf("container %s/%s has not been initialized to add CPU cores allocated", res.UID, res.Container)
|
|
|
- continue
|
|
|
- }
|
|
|
- if len(res.Data) > 0 {
|
|
|
- container.CPUCoresAllocated = res.Data[0].Value
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- ramBytesAllocatedResult, _ := ramBytesAllocatedFuture.Await()
|
|
|
- for _, res := range ramBytesAllocatedResult {
|
|
|
- key := containerKey{podUID: res.UID, name: res.Container}
|
|
|
- container, ok := containerMap[key]
|
|
|
- if !ok {
|
|
|
- log.Warnf("container %s/%s has not been initialized to add RAM bytes allocated", res.UID, res.Container)
|
|
|
- continue
|
|
|
- }
|
|
|
- if len(res.Data) > 0 {
|
|
|
- container.RAMBytesAllocated = res.Data[0].Value
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- cpuUsageAvgResult, _ := cpuUsageAvgFuture.Await()
|
|
|
- for _, res := range cpuUsageAvgResult {
|
|
|
- key := containerKey{podUID: res.UID, name: res.Container}
|
|
|
- container, ok := containerMap[key]
|
|
|
- if !ok {
|
|
|
- log.Warnf("container %s/%s has not been initialized to add CPU usage avg", res.UID, res.Container)
|
|
|
- continue
|
|
|
- }
|
|
|
- if len(res.Data) > 0 {
|
|
|
- container.CPUCoreUsageAvg = res.Data[0].Value
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- cpuUsageMaxResult, _ := cpuUsageMaxFuture.Await()
|
|
|
- for _, res := range cpuUsageMaxResult {
|
|
|
- key := containerKey{podUID: res.UID, name: res.Container}
|
|
|
- container, ok := containerMap[key]
|
|
|
- if !ok {
|
|
|
- log.Warnf("container %s/%s has not been initialized to add CPU usage max", res.UID, res.Container)
|
|
|
- continue
|
|
|
- }
|
|
|
- if len(res.Data) > 0 {
|
|
|
- container.CPUCoreUsageMax = res.Data[0].Value
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- ramUsageAvgResult, _ := ramUsageAvgFuture.Await()
|
|
|
- for _, res := range ramUsageAvgResult {
|
|
|
- key := containerKey{podUID: res.UID, name: res.Container}
|
|
|
- container, ok := containerMap[key]
|
|
|
- if !ok {
|
|
|
- log.Warnf("container %s/%s has not been initialized to add RAM usage avg", res.UID, res.Container)
|
|
|
- continue
|
|
|
- }
|
|
|
- if len(res.Data) > 0 {
|
|
|
- container.RAMBytesUsageAvg = res.Data[0].Value
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- ramUsageMaxResult, _ := ramUsageMaxFuture.Await()
|
|
|
- for _, res := range ramUsageMaxResult {
|
|
|
- key := containerKey{podUID: res.UID, name: res.Container}
|
|
|
- container, ok := containerMap[key]
|
|
|
- if !ok {
|
|
|
- log.Warnf("container %s/%s has not been initialized to add RAM usage max", res.UID, res.Container)
|
|
|
- continue
|
|
|
- }
|
|
|
- if len(res.Data) > 0 {
|
|
|
- container.RAMBytesUsageMax = res.Data[0].Value
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- for _, container := range containerMap {
|
|
|
- err := kms.RegisterContainer(container)
|
|
|
- if err != nil {
|
|
|
- log.Warnf("Failed to register container: %s", err.Error())
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- return nil
|
|
|
-}
|
|
|
-
|
|
|
-func (km *KubeModel) computeResourceQuotas(kms *kubemodel.KubeModelSet, start, end time.Time) error {
|
|
|
- grp := source.NewQueryGroup()
|
|
|
- metrics := km.ds.Metrics()
|
|
|
-
|
|
|
- rqInfoResultFuture := source.WithGroup(grp, metrics.QueryResourceQuotaInfo(start, end))
|
|
|
- rqUptimeResultFuture := source.WithGroup(grp, metrics.QueryResourceQuotaUptime(start, end))
|
|
|
-
|
|
|
- // spec.hard.requests
|
|
|
- rqSpecCPURequestAverageResultFuture := source.WithGroup(grp, metrics.QueryResourceQuotaSpecCPURequestAverage(start, end))
|
|
|
- rqSpecCPURequestMaxResultFuture := source.WithGroup(grp, metrics.QueryResourceQuotaSpecCPURequestMax(start, end))
|
|
|
- rqSpecRAMRequestAverageResultFuture := source.WithGroup(grp, metrics.QueryResourceQuotaSpecRAMRequestAverage(start, end))
|
|
|
- rqSpecRAMRequestMaxResultFuture := source.WithGroup(grp, metrics.QueryResourceQuotaSpecRAMRequestMax(start, end))
|
|
|
-
|
|
|
- // spec.hard.limits
|
|
|
- rqSpecCPULimitAverageResultFuture := source.WithGroup(grp, metrics.QueryResourceQuotaSpecCPULimitAverage(start, end))
|
|
|
- rqSpecCPULimitMaxResultFuture := source.WithGroup(grp, metrics.QueryResourceQuotaSpecCPULimitMax(start, end))
|
|
|
- rqSpecRAMLimitAverageResultFuture := source.WithGroup(grp, metrics.QueryResourceQuotaSpecRAMLimitAverage(start, end))
|
|
|
- rqSpecRAMLimitMaxResultFuture := source.WithGroup(grp, metrics.QueryResourceQuotaSpecRAMLimitMax(start, end))
|
|
|
-
|
|
|
- // status.used.requests
|
|
|
- rqStatusUsedCPURequestAverageResultFuture := source.WithGroup(grp, metrics.QueryResourceQuotaStatusUsedCPURequestAverage(start, end))
|
|
|
- rqStatusUsedCPURequestMaxResultFuture := source.WithGroup(grp, metrics.QueryResourceQuotaStatusUsedCPURequestMax(start, end))
|
|
|
- rqStatusUsedRAMRequestAverageResultFuture := source.WithGroup(grp, metrics.QueryResourceQuotaStatusUsedRAMRequestAverage(start, end))
|
|
|
- rqStatusUsedRAMRequestMaxResultFuture := source.WithGroup(grp, metrics.QueryResourceQuotaStatusUsedRAMRequestMax(start, end))
|
|
|
-
|
|
|
- // status.used.limits
|
|
|
- rqStatusUsedCPULimitAverageResultFuture := source.WithGroup(grp, metrics.QueryResourceQuotaStatusUsedCPULimitAverage(start, end))
|
|
|
- rqStatusUsedCPULimitMaxResultFuture := source.WithGroup(grp, metrics.QueryResourceQuotaStatusUsedCPULimitMax(start, end))
|
|
|
- rqStatusUsedRAMLimitAverageResultFuture := source.WithGroup(grp, metrics.QueryResourceQuotaStatusUsedRAMLimitAverage(start, end))
|
|
|
- rqStatusUsedRAMLimitMaxResultFuture := source.WithGroup(grp, metrics.QueryResourceQuotaStatusUsedRAMLimitMax(start, end))
|
|
|
-
|
|
|
- rqMap := make(map[string]*kubemodel.ResourceQuota)
|
|
|
-
|
|
|
- // Initialize resource quotas from info
|
|
|
- rqInfoResult, _ := rqInfoResultFuture.Await()
|
|
|
- for _, res := range rqInfoResult {
|
|
|
- rqMap[res.UID] = &kubemodel.ResourceQuota{
|
|
|
- UID: res.UID,
|
|
|
- Name: res.ResourceQuota,
|
|
|
- NamespaceUID: res.NamespaceUID,
|
|
|
- Spec: &kubemodel.ResourceQuotaSpec{Hard: &kubemodel.ResourceQuotaSpecHard{}},
|
|
|
- Status: &kubemodel.ResourceQuotaStatus{Used: &kubemodel.ResourceQuotaStatusUsed{}},
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- rqUptimeResult, _ := rqUptimeResultFuture.Await()
|
|
|
- for _, res := range rqUptimeResult {
|
|
|
- rq, ok := rqMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("resource quota with UID '%s' has not been initialized to add uptime", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
- s, e := res.GetStartEnd(start, end, km.ds.Resolution())
|
|
|
- rq.Start = s
|
|
|
- rq.End = e
|
|
|
- }
|
|
|
-
|
|
|
- rqSpecCPURequestAverageResult, _ := rqSpecCPURequestAverageResultFuture.Await()
|
|
|
- for _, res := range rqSpecCPURequestAverageResult {
|
|
|
- rq, ok := rqMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("resource quota with UID '%s' has not been initialized to add spec CPU request average", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
-
|
|
|
- mcpu := res.Value * 1000
|
|
|
- rq.Spec.Hard.SetRequest(kubemodel.ResourceCPU, kubemodel.UnitMillicore, kubemodel.StatAvg, mcpu)
|
|
|
-
|
|
|
- }
|
|
|
-
|
|
|
- rqSpecCPURequestMaxResult, _ := rqSpecCPURequestMaxResultFuture.Await()
|
|
|
- for _, res := range rqSpecCPURequestMaxResult {
|
|
|
- rq, ok := rqMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("resource quota with UID '%s' has not been initialized to add spec CPU request max", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
-
|
|
|
- mcpu := res.Value * 1000
|
|
|
- rq.Spec.Hard.SetRequest(kubemodel.ResourceCPU, kubemodel.UnitMillicore, kubemodel.StatMax, mcpu)
|
|
|
- }
|
|
|
-
|
|
|
- rqSpecRAMRequestAverageResult, _ := rqSpecRAMRequestAverageResultFuture.Await()
|
|
|
- for _, res := range rqSpecRAMRequestAverageResult {
|
|
|
- rq, ok := rqMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("resource quota with UID '%s' has not been initialized to add spec RAM request average", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
-
|
|
|
- rq.Spec.Hard.SetRequest(kubemodel.ResourceMemory, kubemodel.UnitByte, kubemodel.StatAvg, res.Value)
|
|
|
- }
|
|
|
-
|
|
|
- rqSpecRAMRequestMaxResult, _ := rqSpecRAMRequestMaxResultFuture.Await()
|
|
|
- for _, res := range rqSpecRAMRequestMaxResult {
|
|
|
- rq, ok := rqMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("resource quota with UID '%s' has not been initialized to add spec RAM request max", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
-
|
|
|
- rq.Spec.Hard.SetRequest(kubemodel.ResourceMemory, kubemodel.UnitByte, kubemodel.StatMax, res.Value)
|
|
|
- }
|
|
|
-
|
|
|
- rqSpecCPULimitAverageResult, _ := rqSpecCPULimitAverageResultFuture.Await()
|
|
|
- for _, res := range rqSpecCPULimitAverageResult {
|
|
|
- rq, ok := rqMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("resource quota with UID '%s' has not been initialized to add spec CPU limit average", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
-
|
|
|
- mcpu := res.Value * 1000
|
|
|
- rq.Spec.Hard.SetLimit(kubemodel.ResourceCPU, kubemodel.UnitMillicore, kubemodel.StatAvg, mcpu)
|
|
|
-
|
|
|
- }
|
|
|
-
|
|
|
- rqSpecCPULimitMaxResult, _ := rqSpecCPULimitMaxResultFuture.Await()
|
|
|
- for _, res := range rqSpecCPULimitMaxResult {
|
|
|
- rq, ok := rqMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("resource quota with UID '%s' has not been initialized to add spec CPU limit max", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
-
|
|
|
- mcpu := res.Value * 1000
|
|
|
- rq.Spec.Hard.SetLimit(kubemodel.ResourceCPU, kubemodel.UnitMillicore, kubemodel.StatMax, mcpu)
|
|
|
- }
|
|
|
-
|
|
|
- rqSpecRAMLimitAverageResult, _ := rqSpecRAMLimitAverageResultFuture.Await()
|
|
|
- for _, res := range rqSpecRAMLimitAverageResult {
|
|
|
- rq, ok := rqMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("resource quota with UID '%s' has not been initialized to add spec RAM limit average", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
-
|
|
|
- rq.Spec.Hard.SetLimit(kubemodel.ResourceMemory, kubemodel.UnitByte, kubemodel.StatAvg, res.Value)
|
|
|
- }
|
|
|
-
|
|
|
- rqSpecRAMLimitMaxResult, _ := rqSpecRAMLimitMaxResultFuture.Await()
|
|
|
- for _, res := range rqSpecRAMLimitMaxResult {
|
|
|
- rq, ok := rqMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("resource quota with UID '%s' has not been initialized to add spec RAM limit max", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
-
|
|
|
- rq.Spec.Hard.SetLimit(kubemodel.ResourceMemory, kubemodel.UnitByte, kubemodel.StatMax, res.Value)
|
|
|
- }
|
|
|
-
|
|
|
- rqStatusUsedCPURequestAverageResult, _ := rqStatusUsedCPURequestAverageResultFuture.Await()
|
|
|
- for _, res := range rqStatusUsedCPURequestAverageResult {
|
|
|
- rq, ok := rqMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("resource quota with UID '%s' has not been initialized to add status CPU request average", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
-
|
|
|
- mcpu := res.Value * 1000
|
|
|
- rq.Status.Used.SetRequest(kubemodel.ResourceCPU, kubemodel.UnitMillicore, kubemodel.StatAvg, mcpu)
|
|
|
- }
|
|
|
-
|
|
|
- rqStatusUsedCPURequestMaxResult, _ := rqStatusUsedCPURequestMaxResultFuture.Await()
|
|
|
- for _, res := range rqStatusUsedCPURequestMaxResult {
|
|
|
- rq, ok := rqMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("resource quota with UID '%s' has not been initialized to add status CPU request max", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
-
|
|
|
- mcpu := res.Value * 1000
|
|
|
- rq.Status.Used.SetRequest(kubemodel.ResourceCPU, kubemodel.UnitMillicore, kubemodel.StatMax, mcpu)
|
|
|
- }
|
|
|
-
|
|
|
- rqStatusUsedRAMRequestAverageResult, _ := rqStatusUsedRAMRequestAverageResultFuture.Await()
|
|
|
- for _, res := range rqStatusUsedRAMRequestAverageResult {
|
|
|
- rq, ok := rqMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("resource quota with UID '%s' has not been initialized to add status RAM request average", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
-
|
|
|
- rq.Status.Used.SetRequest(kubemodel.ResourceMemory, kubemodel.UnitByte, kubemodel.StatAvg, res.Value)
|
|
|
- }
|
|
|
-
|
|
|
- rqStatusUsedRAMRequestMaxResult, _ := rqStatusUsedRAMRequestMaxResultFuture.Await()
|
|
|
- for _, res := range rqStatusUsedRAMRequestMaxResult {
|
|
|
- rq, ok := rqMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("resource quota with UID '%s' has not been initialized to add status RAM request max", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
-
|
|
|
- rq.Status.Used.SetRequest(kubemodel.ResourceMemory, kubemodel.UnitByte, kubemodel.StatMax, res.Value)
|
|
|
- }
|
|
|
-
|
|
|
- rqStatusUsedCPULimitAverageResult, _ := rqStatusUsedCPULimitAverageResultFuture.Await()
|
|
|
- for _, res := range rqStatusUsedCPULimitAverageResult {
|
|
|
- rq, ok := rqMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("resource quota with UID '%s' has not been initialized to add status CPU limit average", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
-
|
|
|
- mcpu := res.Value * 1000
|
|
|
- rq.Status.Used.SetLimit(kubemodel.ResourceCPU, kubemodel.UnitMillicore, kubemodel.StatAvg, mcpu)
|
|
|
- }
|
|
|
-
|
|
|
- rqStatusUsedCPULimitMaxResult, _ := rqStatusUsedCPULimitMaxResultFuture.Await()
|
|
|
- for _, res := range rqStatusUsedCPULimitMaxResult {
|
|
|
- rq, ok := rqMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("resource quota with UID '%s' has not been initialized to add status CPU limit max", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
-
|
|
|
- mcpu := res.Value * 1000
|
|
|
- rq.Status.Used.SetLimit(kubemodel.ResourceCPU, kubemodel.UnitMillicore, kubemodel.StatMax, mcpu)
|
|
|
- }
|
|
|
-
|
|
|
- rqStatusUsedRAMLimitAverageResult, _ := rqStatusUsedRAMLimitAverageResultFuture.Await()
|
|
|
- for _, res := range rqStatusUsedRAMLimitAverageResult {
|
|
|
- rq, ok := rqMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("resource quota with UID '%s' has not been initialized to add status RAM limit average", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
-
|
|
|
- rq.Status.Used.SetLimit(kubemodel.ResourceMemory, kubemodel.UnitByte, kubemodel.StatAvg, res.Value)
|
|
|
- }
|
|
|
-
|
|
|
- rqStatusUsedRAMLimitMaxResult, _ := rqStatusUsedRAMLimitMaxResultFuture.Await()
|
|
|
- for _, res := range rqStatusUsedRAMLimitMaxResult {
|
|
|
- rq, ok := rqMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("resource quota with UID '%s' has not been initialized to add status RAM limit max", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
-
|
|
|
- rq.Status.Used.SetLimit(kubemodel.ResourceMemory, kubemodel.UnitByte, kubemodel.StatMax, res.Value)
|
|
|
- }
|
|
|
-
|
|
|
- for _, resourceQuota := range rqMap {
|
|
|
- err := kms.RegisterResourceQuota(resourceQuota)
|
|
|
- if err != nil {
|
|
|
- log.Warnf("Failed to register resource quota: %s", err.Error())
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- return nil
|
|
|
-}
|
|
|
-
|
|
|
-func (km *KubeModel) computeServices(kms *kubemodel.KubeModelSet, start, end time.Time) error {
|
|
|
- grp := source.NewQueryGroup()
|
|
|
- metrics := km.ds.Metrics()
|
|
|
-
|
|
|
- serviceInfoResultFuture := source.WithGroup(grp, metrics.QueryServiceInfo(start, end))
|
|
|
- serviceUptimeResultFuture := source.WithGroup(grp, metrics.QueryServiceUptime(start, end))
|
|
|
- serviceSelectorLabelsResultFuture := source.WithGroup(grp, metrics.QueryServiceSelectorLabels(start, end))
|
|
|
-
|
|
|
- serviceMap := make(map[string]*kubemodel.Service)
|
|
|
-
|
|
|
- // Initialize services from info
|
|
|
- serviceInfoResult, _ := serviceInfoResultFuture.Await()
|
|
|
- for _, res := range serviceInfoResult {
|
|
|
- serviceMap[res.UID] = &kubemodel.Service{
|
|
|
- UID: res.UID,
|
|
|
- NamespaceUID: res.NamespaceUID,
|
|
|
- Name: res.Service,
|
|
|
- Type: kubemodel.ParseServiceType(res.ServiceType),
|
|
|
- LBIngressAddress: res.LBIngressAddress,
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- serviceUptimeResult, _ := serviceUptimeResultFuture.Await()
|
|
|
- for _, res := range serviceUptimeResult {
|
|
|
- service, ok := serviceMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("service with UID '%s' has not been initialized to add uptime", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
- s, e := res.GetStartEnd(start, end, km.ds.Resolution())
|
|
|
- service.Start = s
|
|
|
- service.End = e
|
|
|
- }
|
|
|
-
|
|
|
- serviceSelectorLabelsResult, _ := serviceSelectorLabelsResultFuture.Await()
|
|
|
- for _, res := range serviceSelectorLabelsResult {
|
|
|
- service, ok := serviceMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("service with UID '%s' has not been initialized to add selector labels", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
- service.Selector = res.Labels
|
|
|
- }
|
|
|
-
|
|
|
- for _, service := range serviceMap {
|
|
|
- err := kms.RegisterService(service)
|
|
|
- if err != nil {
|
|
|
- log.Warnf("Failed to register service: %s", err.Error())
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- return nil
|
|
|
-}
|
|
|
-
|
|
|
-func (km *KubeModel) computePersistentVolumes(kms *kubemodel.KubeModelSet, start, end time.Time) error {
|
|
|
- grp := source.NewQueryGroup()
|
|
|
- metrics := km.ds.Metrics()
|
|
|
-
|
|
|
- pvInfoResultFuture := source.WithGroup(grp, metrics.QueryKMPVInfo(start, end))
|
|
|
- pvUptimeResultFuture := source.WithGroup(grp, metrics.QueryPVUptime(start, end))
|
|
|
- pvBytesResultFuture := source.WithGroup(grp, metrics.QueryPVBytes(start, end))
|
|
|
-
|
|
|
- pvMap := make(map[string]*kubemodel.PersistentVolume)
|
|
|
-
|
|
|
- pvInfoResult, _ := pvInfoResultFuture.Await()
|
|
|
- for _, res := range pvInfoResult {
|
|
|
- pvMap[res.UID] = &kubemodel.PersistentVolume{
|
|
|
- UID: res.UID,
|
|
|
- Name: res.PersistentVolume,
|
|
|
- StorageClass: res.StorageClass,
|
|
|
- CSIVolumeHandle: res.CSIVolumeHandle,
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- pvUptimeResult, _ := pvUptimeResultFuture.Await()
|
|
|
- for _, res := range pvUptimeResult {
|
|
|
- pv, ok := pvMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("persistent volume with UID '%s' has not been initialized to add uptime", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
- s, e := res.GetStartEnd(start, end, km.ds.Resolution())
|
|
|
- pv.Start = s
|
|
|
- pv.End = e
|
|
|
- }
|
|
|
-
|
|
|
- pvBytesResult, _ := pvBytesResultFuture.Await()
|
|
|
- for _, res := range pvBytesResult {
|
|
|
- pv, ok := pvMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("persistent volume with UID '%s' has not been initialized to add bytes", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
-
|
|
|
- pv.SizeBytes = res.Value
|
|
|
-
|
|
|
- }
|
|
|
-
|
|
|
- for _, pv := range pvMap {
|
|
|
- err := kms.RegisterPersistentVolume(pv)
|
|
|
- if err != nil {
|
|
|
- log.Warnf("Failed to register persistent volume: %s", err.Error())
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- return nil
|
|
|
-}
|
|
|
-
|
|
|
-func (km *KubeModel) computePersistentVolumeClaims(kms *kubemodel.KubeModelSet, start, end time.Time) error {
|
|
|
- grp := source.NewQueryGroup()
|
|
|
- metrics := km.ds.Metrics()
|
|
|
-
|
|
|
- pvcInfoResultFuture := source.WithGroup(grp, metrics.QueryKMPVCInfo(start, end))
|
|
|
- pvcUptimeResultFuture := source.WithGroup(grp, metrics.QueryPVCUptime(start, end))
|
|
|
- pvcBytesRequestedResultFuture := source.WithGroup(grp, metrics.QueryPVCBytesRequested(start, end))
|
|
|
- pvcBytesUsedAvgResultFuture := source.WithGroup(grp, metrics.QueryPVCBytesUsedAverage(start, end))
|
|
|
- pvcBytesUsedMaxResultFuture := source.WithGroup(grp, metrics.QueryPVCBytesUsedMax(start, end))
|
|
|
-
|
|
|
- pvcMap := make(map[string]*kubemodel.PersistentVolumeClaim)
|
|
|
-
|
|
|
- pvcInfoResult, _ := pvcInfoResultFuture.Await()
|
|
|
- for _, res := range pvcInfoResult {
|
|
|
- pvcMap[res.UID] = &kubemodel.PersistentVolumeClaim{
|
|
|
- UID: res.UID,
|
|
|
- Name: res.PersistentVolumeClaim,
|
|
|
- NamespaceUID: res.NamespaceUID,
|
|
|
- PersistentVolumeUID: res.PVUID,
|
|
|
- StorageClass: res.StorageClass,
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- pvcUptimeResult, _ := pvcUptimeResultFuture.Await()
|
|
|
- for _, res := range pvcUptimeResult {
|
|
|
- pvc, ok := pvcMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("persistent volume claim with UID '%s' has not been initialized to add uptime", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
- s, e := res.GetStartEnd(start, end, km.ds.Resolution())
|
|
|
- pvc.Start = s
|
|
|
- pvc.End = e
|
|
|
- }
|
|
|
-
|
|
|
- pvcBytesRequestedResult, _ := pvcBytesRequestedResultFuture.Await()
|
|
|
- for _, res := range pvcBytesRequestedResult {
|
|
|
- pvc, ok := pvcMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("persistent volume claim with UID '%s' has not been initialized to add requested bytes", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
- if len(res.Data) > 0 {
|
|
|
- pvc.RequestedBytes = res.Data[0].Value
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- pvcBytesUsedAvgResult, _ := pvcBytesUsedAvgResultFuture.Await()
|
|
|
- for _, res := range pvcBytesUsedAvgResult {
|
|
|
- pvc, ok := pvcMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("persistent volume claim with UID '%s' has not been initialized to add bytes used average", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
- pvc.UsageBytesAvg = res.Value
|
|
|
- }
|
|
|
-
|
|
|
- pvcBytesUsedMaxResult, _ := pvcBytesUsedMaxResultFuture.Await()
|
|
|
- for _, res := range pvcBytesUsedMaxResult {
|
|
|
- pvc, ok := pvcMap[res.UID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("persistent volume claim with UID '%s' has not been initialized to add bytes used max", res.UID)
|
|
|
- continue
|
|
|
- }
|
|
|
- pvc.UsageBytesMax = res.Value
|
|
|
- }
|
|
|
-
|
|
|
- for _, pvc := range pvcMap {
|
|
|
- err := kms.RegisterPVC(pvc)
|
|
|
- if err != nil {
|
|
|
- log.Warnf("Failed to register persistent volume claim: %s", err.Error())
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- return nil
|
|
|
-}
|
|
|
-
|
|
|
-func (km *KubeModel) computeDCGMDevices(kms *kubemodel.KubeModelSet, start, end time.Time) error {
|
|
|
- grp := source.NewQueryGroup()
|
|
|
- metrics := km.ds.Metrics()
|
|
|
-
|
|
|
- dcgmInfoFuture := source.WithGroup(grp, metrics.QueryDCGMDeviceInfo(start, end))
|
|
|
- dcgmUptimeFuture := source.WithGroup(grp, metrics.QueryDCGMDeviceUptime(start, end))
|
|
|
- dcgmUsageAvgFuture := source.WithGroup(grp, metrics.QueryDCGMContainerUsageAvg(start, end))
|
|
|
- dcgmUsageMaxFuture := source.WithGroup(grp, metrics.QueryDCGMContainerUsageMax(start, end))
|
|
|
-
|
|
|
- deviceMap := make(map[string]*kubemodel.DCGMDevice)
|
|
|
-
|
|
|
- dcgmInfoResult, _ := dcgmInfoFuture.Await()
|
|
|
- for _, res := range dcgmInfoResult {
|
|
|
- if res.UUID == "" {
|
|
|
- continue
|
|
|
- }
|
|
|
- if _, ok := deviceMap[res.UUID]; ok {
|
|
|
- continue
|
|
|
- }
|
|
|
- deviceMap[res.UUID] = &kubemodel.DCGMDevice{
|
|
|
- UUID: res.UUID,
|
|
|
- Device: res.Device,
|
|
|
- ModelName: res.ModelName,
|
|
|
- PodUsages: make(map[string]kubemodel.DCGMPod),
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- dcgmUptimeResult, _ := dcgmUptimeFuture.Await()
|
|
|
- for _, res := range dcgmUptimeResult {
|
|
|
- d, ok := deviceMap[res.UUID]
|
|
|
- if !ok {
|
|
|
- log.Warnf("DCGM uptime result for unknown device UUID '%s'", res.UUID)
|
|
|
- continue
|
|
|
- }
|
|
|
- s, e := res.GetStartEnd(start, end, km.ds.Resolution())
|
|
|
- d.Start = s
|
|
|
- d.End = e
|
|
|
- }
|
|
|
-
|
|
|
- dcgmUsageAvgResult, _ := dcgmUsageAvgFuture.Await()
|
|
|
- for _, res := range dcgmUsageAvgResult {
|
|
|
- device, ok := deviceMap[res.UUID]
|
|
|
- if !ok || res.PodUID == "" || res.Container == "" {
|
|
|
- continue
|
|
|
- }
|
|
|
- pod, ok := device.PodUsages[res.PodUID]
|
|
|
- if !ok {
|
|
|
- pod = kubemodel.DCGMPod{ContainerUsages: make(map[string]kubemodel.DCGMContainer)}
|
|
|
- }
|
|
|
- c := pod.ContainerUsages[res.Container]
|
|
|
- c.UsageAvg = res.Value
|
|
|
- pod.ContainerUsages[res.Container] = c
|
|
|
- device.PodUsages[res.PodUID] = pod
|
|
|
- }
|
|
|
-
|
|
|
- dcgmUsageMaxResult, _ := dcgmUsageMaxFuture.Await()
|
|
|
- for _, res := range dcgmUsageMaxResult {
|
|
|
- device, ok := deviceMap[res.UUID]
|
|
|
- if !ok || res.PodUID == "" || res.Container == "" {
|
|
|
- continue
|
|
|
- }
|
|
|
- pod, ok := device.PodUsages[res.PodUID]
|
|
|
- if !ok {
|
|
|
- pod = kubemodel.DCGMPod{ContainerUsages: make(map[string]kubemodel.DCGMContainer)}
|
|
|
- }
|
|
|
- c := pod.ContainerUsages[res.Container]
|
|
|
- c.UsageMax = res.Value
|
|
|
- pod.ContainerUsages[res.Container] = c
|
|
|
- device.PodUsages[res.PodUID] = pod
|
|
|
- }
|
|
|
-
|
|
|
- for _, device := range deviceMap {
|
|
|
- if err := kms.RegisterDCGMDevice(device); err != nil {
|
|
|
- log.Warnf("Failed to register DCGM device: %s", err.Error())
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- return nil
|
|
|
-}
|
Sean Holcomb