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

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/source"
)

type KubeModel struct {
	ds         source.OpenCostDataSource
	forceV1    bool
	clusterUID string
}

func NewKubeModel(clusterUID string, forceV1 bool, dataSource source.OpenCostDataSource) (*KubeModel, error) {
	if dataSource == nil {
		return nil, errors.New("OpenCostDataSource cannot be nil")
	}

	km := &KubeModel{
		ds:         dataSource,
		forceV1:    forceV1,
		clusterUID: clusterUID,
	}

	km.clusterUID = clusterUID

	log.Debugf("NewKubeModel(%s)", km.clusterUID)

	return km, nil
}

type computeFunc func(*kubemodel.KubeModelSet, time.Time, time.Time) error

// 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) {
	kms := kubemodel.NewKubeModelSet(start, end)

	// Compute every resource from one consistent state of the data source by running the compute
	// functions against a copy of the KubeModel whose data source is pinned.
	ds, release := source.PinDataSource(km.ds)
	defer release()
	pinned := *km
	pinned.ds = ds

	computeFuncs := pinned.computeFuncs(start, end)

	for _, f := range computeFuncs {
		if err := f(kms, start, end); err != nil {
			kms.Error(err)
			return kms, fmt.Errorf("error computing kubemodel for (%s, %s): %w", start.Format(time.DateTime), end.Format(time.DateTime), err)
		}
	}

	kms.Metadata.CompletedAt = time.Now().UTC()

	return kms, nil
}

// computeFuncs returns the set of compute functions to run for the window,
// in struct field order. If cluster_info is not yet reporting the
// complete_kubemodel label for any cluster in the result, the source has not
// been upgraded to emit a full kubemodel, so only the minimal set of
// resources (cluster, namespaces, resource quotas) is computed. The
// FORCE_KUBEMODEL_V1 env var skips the check entirely and always returns the
// minimal set.
func (km *KubeModel) computeFuncs(start, end time.Time) []computeFunc {
	KubeModelV1ComputeFucs := []computeFunc{
		km.computeCluster,
		km.computeNamespaces,
		km.computeResourceQuotas,
	}

	if km.forceV1 {
		return KubeModelV1ComputeFucs
	}

	results, err := km.ds.Metrics().QueryClusterKubeModelVersion(start, end).Await()
	if err != nil {
		log.Errorf("computeFuncs: querying cluster complete kubemodel: %s", err)
		return KubeModelV1ComputeFucs
	}

	// If the window contains a result which is missing the version number then the window will produce incomplete KubeModel data
	// and should not export
	for _, res := range results {
		if res.Version == "" {
			return KubeModelV1ComputeFucs
		}
	}

	return []computeFunc{
		km.computeCluster,
		km.computeNamespaces,
		km.computeResourceQuotas,
		km.computeServices,
		km.computeDeployments,
		km.computeStatefulSets,
		km.computeDaemonSets,
		km.computeJobs,
		km.computeCronJobs,
		km.computeReplicaSets,
		km.computeNodes,
		km.computePersistentVolumes,
		km.computePersistentVolumeClaims,
		km.computePods,
		km.computeContainers,
		km.computeDevices,
	}
}