cost-model/core/pkg/compute/kubemodel/node.go

package kubemodel

import (
	"time"

	"github.com/opencost/opencost/core/pkg/log"
	"github.com/opencost/opencost/core/pkg/model/kubemodel"
	"github.com/opencost/opencost/core/pkg/source"
)

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
}