| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884188518861887188818891890189118921893189418951896189718981899190019011902190319041905190619071908190919101911191219131914191519161917191819191920192119221923192419251926192719281929193019311932193319341935193619371938193919401941194219431944194519461947194819491950195119521953195419551956195719581959196019611962196319641965196619671968196919701971197219731974197519761977197819791980198119821983198419851986198719881989199019911992199319941995199619971998199920002001200220032004200520062007200820092010201120122013201420152016201720182019202020212022202320242025202620272028202920302031203220332034203520362037203820392040204120422043204420452046204720482049205020512052205320542055205620572058205920602061206220632064206520662067206820692070207120722073207420752076207720782079208020812082208320842085208620872088208920902091209220932094209520962097209820992100210121022103210421052106210721082109211021112112211321142115211621172118211921202121212221232124212521262127212821292130213121322133213421352136213721382139214021412142214321442145214621472148214921502151215221532154215521562157215821592160216121622163216421652166216721682169217021712172217321742175217621772178217921802181218221832184218521862187218821892190219121922193219421952196219721982199220022012202220322042205220622072208220922102211221222132214221522162217221822192220222122222223222422252226222722282229223022312232223322342235223622372238223922402241224222432244224522462247224822492250225122522253225422552256225722582259226022612262226322642265226622672268226922702271227222732274227522762277227822792280228122822283228422852286228722882289229022912292229322942295229622972298229923002301230223032304230523062307230823092310231123122313231423152316231723182319232023212322232323242325232623272328232923302331233223332334233523362337233823392340234123422343234423452346234723482349235023512352235323542355235623572358235923602361236223632364236523662367236823692370237123722373237423752376237723782379238023812382238323842385238623872388238923902391239223932394 |
- package costmodel
- import (
- "fmt"
- "math"
- "strconv"
- "strings"
- "time"
- "github.com/opencost/opencost/core/pkg/log"
- "github.com/opencost/opencost/core/pkg/opencost"
- "github.com/opencost/opencost/core/pkg/util/timeutil"
- "github.com/opencost/opencost/pkg/cloud/provider"
- "github.com/opencost/opencost/pkg/env"
- "github.com/opencost/opencost/pkg/prom"
- "k8s.io/apimachinery/pkg/labels"
- )
- // This is a bit of a hack to work around garbage data from cadvisor
- // Ideally you cap each pod to the max CPU on its node, but that involves a bit more complexity, as it it would need to be done when allocations joins with asset data.
- const CPU_SANITY_LIMIT = 512
- // Sanity Limit for PV usage, set to 10 PB, in bytes for now
- const KiB = 1024.0
- const MiB = 1024.0 * KiB
- const GiB = 1024.0 * MiB
- const TiB = 1024.0 * GiB
- const PiB = 1024.0 * TiB
- const PV_USAGE_SANITY_LIMIT_BYTES = 10.0 * PiB
- const (
- GpuUsageAverageMode = "AVERAGE"
- GpuUsageMaxMode = "MAX"
- GpuIsSharedMode = "SHARED"
- GpuInfoMode = "GPU_INFO"
- )
- /* Pod Helpers */
- func (cm *CostModel) buildPodMap(window opencost.Window, resolution, maxBatchSize time.Duration, podMap map[podKey]*pod, clusterStart, clusterEnd map[string]time.Time, ingestPodUID bool, podUIDKeyMap map[podKey][]podKey) error {
- // Assumes that window is positive and closed
- start, end := *window.Start(), *window.End()
- // Convert resolution duration to a query-ready string
- resStr := timeutil.DurationString(resolution)
- ctx := prom.NewNamedContext(cm.PrometheusClient, prom.AllocationContextName)
- // Query for (start, end) by (pod, namespace, cluster) over the given
- // window, using the given resolution, and if necessary in batches no
- // larger than the given maximum batch size. If working in batches, track
- // overall progress by starting with (window.start, window.start) and
- // querying in batches no larger than maxBatchSize from start-to-end,
- // folding each result set into podMap as the results come back.
- coverage := opencost.NewWindow(&start, &start)
- numQuery := 1
- for coverage.End().Before(end) {
- // Determine the (start, end) of the current batch
- batchStart := *coverage.End()
- batchEnd := coverage.End().Add(maxBatchSize)
- if batchEnd.After(end) {
- batchEnd = end
- }
- var resPods []*prom.QueryResult
- var err error
- maxTries := 3
- numTries := 0
- for resPods == nil && numTries < maxTries {
- numTries++
- // Query for the duration between start and end
- durStr := timeutil.DurationString(batchEnd.Sub(batchStart))
- if durStr == "" {
- // Negative duration, so set empty results and don't query
- resPods = []*prom.QueryResult{}
- err = nil
- break
- }
- // Submit and profile query
- var queryPods string
- // If ingesting UIDs, avg on them
- if ingestPodUID {
- queryPods = fmt.Sprintf(queryFmtPodsUID, env.GetPromClusterFilter(), env.GetPromClusterLabel(), durStr, resStr)
- } else {
- queryPods = fmt.Sprintf(queryFmtPods, env.GetPromClusterFilter(), env.GetPromClusterLabel(), durStr, resStr)
- }
- queryProfile := time.Now()
- resPods, err = ctx.QueryAtTime(queryPods, batchEnd).Await()
- if err != nil {
- log.Profile(queryProfile, fmt.Sprintf("CostModel.ComputeAllocation: pod query %d try %d failed: %s", numQuery, numTries, queryPods))
- resPods = nil
- }
- }
- if err != nil {
- return err
- }
- // queryFmtPodsUID will return both UID-containing results, and non-UID-containing results,
- // so filter out the non-containing results so we don't duplicate pods. This is due to the
- // default setup of Kubecost having replicated kube_pod_container_status_running and
- // included KSM kube_pod_container_status_running. Querying w/ UID will return both.
- if ingestPodUID {
- var resPodsUID []*prom.QueryResult
- for _, res := range resPods {
- _, err := res.GetString("uid")
- if err == nil {
- resPodsUID = append(resPodsUID, res)
- }
- }
- if len(resPodsUID) > 0 {
- resPods = resPodsUID
- } else {
- log.DedupedWarningf(5, "CostModel.ComputeAllocation: UID ingestion enabled, but query did not return any results with UID")
- }
- }
- applyPodResults(window, resolution, podMap, clusterStart, clusterEnd, resPods, ingestPodUID, podUIDKeyMap)
- coverage = coverage.ExpandEnd(batchEnd)
- numQuery++
- }
- return nil
- }
- func applyPodResults(window opencost.Window, resolution time.Duration, podMap map[podKey]*pod, clusterStart, clusterEnd map[string]time.Time, resPods []*prom.QueryResult, ingestPodUID bool, podUIDKeyMap map[podKey][]podKey) {
- for _, res := range resPods {
- if len(res.Values) == 0 {
- log.Warnf("CostModel.ComputeAllocation: empty minutes result")
- continue
- }
- cluster, err := res.GetString(env.GetPromClusterLabel())
- if err != nil {
- cluster = env.GetClusterID()
- }
- labels, err := res.GetStrings("namespace", "pod")
- if err != nil {
- log.Warnf("CostModel.ComputeAllocation: minutes query result missing field: %s", err)
- continue
- }
- namespace := labels["namespace"]
- podName := labels["pod"]
- key := newPodKey(cluster, namespace, podName)
- // If thisPod UIDs are being used to ID pods, append them to the thisPod name in
- // the podKey.
- if ingestPodUID {
- uid, err := res.GetString("uid")
- if err != nil {
- log.Warnf("CostModel.ComputeAllocation: UID ingestion enabled, but query result missing field: %s", err)
- } else {
- newKey := newPodKey(cluster, namespace, podName+" "+uid)
- podUIDKeyMap[key] = append(podUIDKeyMap[key], newKey)
- key = newKey
- }
- }
- allocStart, allocEnd := calculateStartAndEnd(res, resolution, window)
- if allocStart.IsZero() || allocEnd.IsZero() {
- continue
- }
- // Set start if unset or this datum's start time is earlier than the
- // current earliest time.
- if _, ok := clusterStart[cluster]; !ok || allocStart.Before(clusterStart[cluster]) {
- clusterStart[cluster] = allocStart
- }
- // Set end if unset or this datum's end time is later than the
- // current latest time.
- if _, ok := clusterEnd[cluster]; !ok || allocEnd.After(clusterEnd[cluster]) {
- clusterEnd[cluster] = allocEnd
- }
- if thisPod, ok := podMap[key]; ok {
- // Pod has already been recorded, so update it accordingly
- if allocStart.Before(thisPod.Start) {
- thisPod.Start = allocStart
- }
- if allocEnd.After(thisPod.End) {
- thisPod.End = allocEnd
- }
- } else {
- // pod has not been recorded yet, so insert it
- podMap[key] = &pod{
- Window: window.Clone(),
- Start: allocStart,
- End: allocEnd,
- Key: key,
- Allocations: map[string]*opencost.Allocation{},
- }
- }
- }
- }
- func applyCPUCoresAllocated(podMap map[podKey]*pod, resCPUCoresAllocated []*prom.QueryResult, podUIDKeyMap map[podKey][]podKey) {
- for _, res := range resCPUCoresAllocated {
- key, err := resultPodKey(res, env.GetPromClusterLabel(), "namespace")
- if err != nil {
- log.DedupedWarningf(10, "CostModel.ComputeAllocation: CPU allocation result missing field: %s", err)
- continue
- }
- container, err := res.GetString("container")
- if err != nil {
- log.DedupedWarningf(10, "CostModel.ComputeAllocation: CPU allocation query result missing 'container': %s", key)
- continue
- }
- var pods []*pod
- if thisPod, ok := podMap[key]; !ok {
- if uidKeys, ok := podUIDKeyMap[key]; ok {
- for _, uidKey := range uidKeys {
- thisPod, ok = podMap[uidKey]
- if ok {
- pods = append(pods, thisPod)
- }
- }
- } else {
- continue
- }
- } else {
- pods = []*pod{thisPod}
- }
- for _, thisPod := range pods {
- if _, ok := thisPod.Allocations[container]; !ok {
- thisPod.appendContainer(container)
- }
- cpuCores := res.Values[0].Value
- if cpuCores > CPU_SANITY_LIMIT {
- log.Infof("[WARNING] Very large cpu allocation, clamping to %f", res.Values[0].Value*(thisPod.Allocations[container].Minutes()/60.0))
- cpuCores = 0.0
- }
- hours := thisPod.Allocations[container].Minutes() / 60.0
- thisPod.Allocations[container].CPUCoreHours = cpuCores * hours
- node, err := res.GetString("node")
- if err != nil {
- log.Warnf("CostModel.ComputeAllocation: CPU allocation query result missing 'node': %s", key)
- continue
- }
- thisPod.Allocations[container].Properties.Node = node
- thisPod.Node = node
- }
- }
- }
- func applyCPUCoresRequested(podMap map[podKey]*pod, resCPUCoresRequested []*prom.QueryResult, podUIDKeyMap map[podKey][]podKey) {
- for _, res := range resCPUCoresRequested {
- key, err := resultPodKey(res, env.GetPromClusterLabel(), "namespace")
- if err != nil {
- log.DedupedWarningf(10, "CostModel.ComputeAllocation: CPU request result missing field: %s", err)
- continue
- }
- container, err := res.GetString("container")
- if err != nil {
- log.DedupedWarningf(10, "CostModel.ComputeAllocation: CPU request query result missing 'container': %s", key)
- continue
- }
- var pods []*pod
- if thisPod, ok := podMap[key]; !ok {
- if uidKeys, ok := podUIDKeyMap[key]; ok {
- for _, uidKey := range uidKeys {
- thisPod, ok = podMap[uidKey]
- if ok {
- pods = append(pods, thisPod)
- }
- }
- } else {
- continue
- }
- } else {
- pods = []*pod{thisPod}
- }
- for _, thisPod := range pods {
- if _, ok := thisPod.Allocations[container]; !ok {
- thisPod.appendContainer(container)
- }
- thisPod.Allocations[container].CPUCoreRequestAverage = res.Values[0].Value
- // If CPU allocation is less than requests, set CPUCoreHours to
- // request level.
- if thisPod.Allocations[container].CPUCores() < res.Values[0].Value {
- thisPod.Allocations[container].CPUCoreHours = res.Values[0].Value * (thisPod.Allocations[container].Minutes() / 60.0)
- }
- if thisPod.Allocations[container].CPUCores() > CPU_SANITY_LIMIT {
- log.Infof("[WARNING] Very large cpu allocation, clamping! to %f", res.Values[0].Value*(thisPod.Allocations[container].Minutes()/60.0))
- thisPod.Allocations[container].CPUCoreHours = res.Values[0].Value * (thisPod.Allocations[container].Minutes() / 60.0)
- }
- node, err := res.GetString("node")
- if err != nil {
- log.Warnf("CostModel.ComputeAllocation: CPU request query result missing 'node': %s", key)
- continue
- }
- thisPod.Allocations[container].Properties.Node = node
- thisPod.Node = node
- }
- }
- }
- func applyCPUCoresUsedAvg(podMap map[podKey]*pod, resCPUCoresUsedAvg []*prom.QueryResult, podUIDKeyMap map[podKey][]podKey) {
- for _, res := range resCPUCoresUsedAvg {
- key, err := resultPodKey(res, env.GetPromClusterLabel(), "namespace")
- if err != nil {
- log.DedupedWarningf(10, "CostModel.ComputeAllocation: CPU usage avg result missing field: %s", err)
- continue
- }
- container, err := res.GetString("container")
- if container == "" || err != nil {
- container, err = res.GetString("container_name")
- if err != nil {
- log.DedupedWarningf(10, "CostModel.ComputeAllocation: CPU usage avg query result missing 'container': %s", key)
- continue
- }
- }
- var pods []*pod
- if thisPod, ok := podMap[key]; !ok {
- if uidKeys, ok := podUIDKeyMap[key]; ok {
- for _, uidKey := range uidKeys {
- thisPod, ok = podMap[uidKey]
- if ok {
- pods = append(pods, thisPod)
- }
- }
- } else {
- continue
- }
- } else {
- pods = []*pod{thisPod}
- }
- for _, thisPod := range pods {
- if _, ok := thisPod.Allocations[container]; !ok {
- thisPod.appendContainer(container)
- }
- thisPod.Allocations[container].CPUCoreUsageAverage = res.Values[0].Value
- if res.Values[0].Value > CPU_SANITY_LIMIT {
- log.Infof("[WARNING] Very large cpu USAGE, dropping outlier")
- thisPod.Allocations[container].CPUCoreUsageAverage = 0.0
- }
- }
- }
- }
- func applyCPUCoresUsedMax(podMap map[podKey]*pod, resCPUCoresUsedMax []*prom.QueryResult, podUIDKeyMap map[podKey][]podKey) {
- for _, res := range resCPUCoresUsedMax {
- key, err := resultPodKey(res, env.GetPromClusterLabel(), "namespace")
- if err != nil {
- log.DedupedWarningf(10, "CostModel.ComputeAllocation: CPU usage max result missing field: %s", err)
- continue
- }
- container, err := res.GetString("container")
- if container == "" || err != nil {
- container, err = res.GetString("container_name")
- if err != nil {
- log.DedupedWarningf(10, "CostModel.ComputeAllocation: CPU usage max query result missing 'container': %s", key)
- continue
- }
- }
- var pods []*pod
- if thisPod, ok := podMap[key]; !ok {
- if uidKeys, ok := podUIDKeyMap[key]; ok {
- for _, uidKey := range uidKeys {
- thisPod, ok = podMap[uidKey]
- if ok {
- pods = append(pods, thisPod)
- }
- }
- } else {
- continue
- }
- } else {
- pods = []*pod{thisPod}
- }
- for _, thisPod := range pods {
- if _, ok := thisPod.Allocations[container]; !ok {
- thisPod.appendContainer(container)
- }
- if thisPod.Allocations[container].RawAllocationOnly == nil {
- thisPod.Allocations[container].RawAllocationOnly = &opencost.RawAllocationOnlyData{
- CPUCoreUsageMax: res.Values[0].Value,
- }
- } else {
- thisPod.Allocations[container].RawAllocationOnly.CPUCoreUsageMax = res.Values[0].Value
- }
- }
- }
- }
- func applyRAMBytesAllocated(podMap map[podKey]*pod, resRAMBytesAllocated []*prom.QueryResult, podUIDKeyMap map[podKey][]podKey) {
- for _, res := range resRAMBytesAllocated {
- key, err := resultPodKey(res, env.GetPromClusterLabel(), "namespace")
- if err != nil {
- log.DedupedWarningf(10, "CostModel.ComputeAllocation: RAM allocation result missing field: %s", err)
- continue
- }
- container, err := res.GetString("container")
- if err != nil {
- log.DedupedWarningf(10, "CostModel.ComputeAllocation: RAM allocation query result missing 'container': %s", key)
- continue
- }
- var pods []*pod
- if thisPod, ok := podMap[key]; !ok {
- if uidKeys, ok := podUIDKeyMap[key]; ok {
- for _, uidKey := range uidKeys {
- thisPod, ok = podMap[uidKey]
- if ok {
- pods = append(pods, thisPod)
- }
- }
- } else {
- continue
- }
- } else {
- pods = []*pod{thisPod}
- }
- for _, thisPod := range pods {
- if _, ok := thisPod.Allocations[container]; !ok {
- thisPod.appendContainer(container)
- }
- ramBytes := res.Values[0].Value
- hours := thisPod.Allocations[container].Minutes() / 60.0
- thisPod.Allocations[container].RAMByteHours = ramBytes * hours
- node, err := res.GetString("node")
- if err != nil {
- log.Warnf("CostModel.ComputeAllocation: RAM allocation query result missing 'node': %s", key)
- continue
- }
- thisPod.Allocations[container].Properties.Node = node
- thisPod.Node = node
- }
- }
- }
- func applyRAMBytesRequested(podMap map[podKey]*pod, resRAMBytesRequested []*prom.QueryResult, podUIDKeyMap map[podKey][]podKey) {
- for _, res := range resRAMBytesRequested {
- key, err := resultPodKey(res, env.GetPromClusterLabel(), "namespace")
- if err != nil {
- log.DedupedWarningf(10, "CostModel.ComputeAllocation: RAM request result missing field: %s", err)
- continue
- }
- container, err := res.GetString("container")
- if err != nil {
- log.DedupedWarningf(10, "CostModel.ComputeAllocation: RAM request query result missing 'container': %s", key)
- continue
- }
- var pods []*pod
- if thisPod, ok := podMap[key]; !ok {
- if uidKeys, ok := podUIDKeyMap[key]; ok {
- for _, uidKey := range uidKeys {
- thisPod, ok = podMap[uidKey]
- if ok {
- pods = append(pods, thisPod)
- }
- }
- } else {
- continue
- }
- } else {
- pods = []*pod{thisPod}
- }
- for _, pod := range pods {
- if _, ok := pod.Allocations[container]; !ok {
- pod.appendContainer(container)
- }
- pod.Allocations[container].RAMBytesRequestAverage = res.Values[0].Value
- // If RAM allocation is less than requests, set RAMByteHours to
- // request level.
- if pod.Allocations[container].RAMBytes() < res.Values[0].Value {
- pod.Allocations[container].RAMByteHours = res.Values[0].Value * (pod.Allocations[container].Minutes() / 60.0)
- }
- node, err := res.GetString("node")
- if err != nil {
- log.Warnf("CostModel.ComputeAllocation: RAM request query result missing 'node': %s", key)
- continue
- }
- pod.Allocations[container].Properties.Node = node
- pod.Node = node
- }
- }
- }
- func applyRAMBytesUsedAvg(podMap map[podKey]*pod, resRAMBytesUsedAvg []*prom.QueryResult, podUIDKeyMap map[podKey][]podKey) {
- for _, res := range resRAMBytesUsedAvg {
- key, err := resultPodKey(res, env.GetPromClusterLabel(), "namespace")
- if err != nil {
- log.DedupedWarningf(10, "CostModel.ComputeAllocation: RAM avg usage result missing field: %s", err)
- continue
- }
- container, err := res.GetString("container")
- if container == "" || err != nil {
- container, err = res.GetString("container_name")
- if err != nil {
- log.DedupedWarningf(10, "CostModel.ComputeAllocation: RAM usage avg query result missing 'container': %s", key)
- continue
- }
- }
- var pods []*pod
- if thisPod, ok := podMap[key]; !ok {
- if uidKeys, ok := podUIDKeyMap[key]; ok {
- for _, uidKey := range uidKeys {
- thisPod, ok = podMap[uidKey]
- if ok {
- pods = append(pods, thisPod)
- }
- }
- } else {
- continue
- }
- } else {
- pods = []*pod{thisPod}
- }
- for _, thisPod := range pods {
- if _, ok := thisPod.Allocations[container]; !ok {
- thisPod.appendContainer(container)
- }
- thisPod.Allocations[container].RAMBytesUsageAverage = res.Values[0].Value
- }
- }
- }
- func applyRAMBytesUsedMax(podMap map[podKey]*pod, resRAMBytesUsedMax []*prom.QueryResult, podUIDKeyMap map[podKey][]podKey) {
- for _, res := range resRAMBytesUsedMax {
- key, err := resultPodKey(res, env.GetPromClusterLabel(), "namespace")
- if err != nil {
- log.DedupedWarningf(10, "CostModel.ComputeAllocation: RAM usage max result missing field: %s", err)
- continue
- }
- container, err := res.GetString("container")
- if container == "" || err != nil {
- container, err = res.GetString("container_name")
- if err != nil {
- log.DedupedWarningf(10, "CostModel.ComputeAllocation: RAM usage max query result missing 'container': %s", key)
- continue
- }
- }
- var pods []*pod
- if thisPod, ok := podMap[key]; !ok {
- if uidKeys, ok := podUIDKeyMap[key]; ok {
- for _, uidKey := range uidKeys {
- thisPod, ok = podMap[uidKey]
- if ok {
- pods = append(pods, thisPod)
- }
- }
- } else {
- continue
- }
- } else {
- pods = []*pod{thisPod}
- }
- for _, thisPod := range pods {
- if _, ok := thisPod.Allocations[container]; !ok {
- thisPod.appendContainer(container)
- }
- if thisPod.Allocations[container].RawAllocationOnly == nil {
- thisPod.Allocations[container].RawAllocationOnly = &opencost.RawAllocationOnlyData{
- RAMBytesUsageMax: res.Values[0].Value,
- }
- } else {
- thisPod.Allocations[container].RawAllocationOnly.RAMBytesUsageMax = res.Values[0].Value
- }
- }
- }
- }
- // same func is used for both GPUUsageAvg and GPUUsageMax
- func applyGPUUsage(podMap map[podKey]*pod, resGPUUsageAvgOrMax []*prom.QueryResult, podUIDKeyMap map[podKey][]podKey, mode string) {
- // Example PromQueryResult: {container="dcgmproftester12", namespace="gpu", pod="dcgmproftester3-deployment-fc89c8dd6-ph7z5"} 0.997307
- for _, res := range resGPUUsageAvgOrMax {
- key, err := resultPodKey(res, env.GetPromClusterLabel(), "namespace")
- if err != nil {
- log.DedupedWarningf(10, "CostModel.ComputeAllocation: GPU usage avg/max result missing field: %s", err)
- continue
- }
- var pods []*pod
- if thisPod, ok := podMap[key]; !ok {
- if uidKeys, ok := podUIDKeyMap[key]; ok {
- for _, uidKey := range uidKeys {
- thisPod, ok = podMap[uidKey]
- if ok {
- pods = append(pods, thisPod)
- }
- }
- } else {
- continue
- }
- } else {
- pods = []*pod{thisPod}
- }
- for _, thisPod := range pods {
- container, err := res.GetString("container")
- if err != nil {
- log.DedupedWarningf(10, "CostModel.ComputeAllocation: GPU usage avg/max query result missing 'container': %s", key)
- continue
- }
- if _, ok := thisPod.Allocations[container]; !ok {
- thisPod.appendContainer(container)
- }
- // DCGM_FI_PROF_GR_ENGINE_ACTIVE metric is a float between 0-1.
- switch mode {
- case GpuUsageAverageMode:
- if thisPod.Allocations[container].GPUAllocation == nil {
- thisPod.Allocations[container].GPUAllocation = &opencost.GPUAllocation{GPUUsageAverage: &res.Values[0].Value}
- } else {
- thisPod.Allocations[container].GPUAllocation.GPUUsageAverage = &res.Values[0].Value
- }
- case GpuUsageMaxMode:
- if thisPod.Allocations[container].RawAllocationOnly == nil {
- thisPod.Allocations[container].RawAllocationOnly = &opencost.RawAllocationOnlyData{
- GPUUsageMax: &res.Values[0].Value,
- }
- } else {
- thisPod.Allocations[container].RawAllocationOnly.GPUUsageMax = &res.Values[0].Value
- }
- case GpuIsSharedMode:
- // if a container is using a GPU and it is shared, isGPUShared will be true
- // if a container is using GPU and it is NOT shared, isGPUShared will be false
- // if a container is NOT using a GPU, isGPUShared will be null
- if res.Metric["resource"] == "nvidia_com_gpu_shared" {
- trueVal := true
- if res.Values[0].Value == 1 {
- if thisPod.Allocations[container].GPUAllocation == nil {
- thisPod.Allocations[container].GPUAllocation = &opencost.GPUAllocation{IsGPUShared: &trueVal}
- } else {
- thisPod.Allocations[container].GPUAllocation.IsGPUShared = &trueVal
- }
- }
- } else if res.Metric["resource"] == "nvidia_com_gpu" {
- falseVal := false
- if res.Values[0].Value == 1 {
- if thisPod.Allocations[container].GPUAllocation == nil {
- thisPod.Allocations[container].GPUAllocation = &opencost.GPUAllocation{IsGPUShared: &falseVal}
- } else {
- thisPod.Allocations[container].GPUAllocation.IsGPUShared = &falseVal
- }
- }
- } else {
- continue
- }
- case GpuInfoMode:
- if thisPod.Allocations[container].GPUAllocation == nil {
- thisPod.Allocations[container].GPUAllocation = &opencost.GPUAllocation{
- GPUDevice: getSanitizedDeviceName(fmt.Sprintf("%s", res.Metric["device_name"])),
- GPUModel: fmt.Sprintf("%s", res.Metric["modelName"]),
- GPUUUID: fmt.Sprintf("%s", res.Metric["UUID"]),
- }
- } else {
- thisPod.Allocations[container].GPUAllocation.GPUDevice = getSanitizedDeviceName(fmt.Sprintf("%s", res.Metric["device"]))
- thisPod.Allocations[container].GPUAllocation.GPUModel = fmt.Sprintf("%s", res.Metric["modelName"])
- thisPod.Allocations[container].GPUAllocation.GPUUUID = fmt.Sprintf("%s", res.Metric["UUID"])
- }
- default:
- log.DedupedInfof(10, "CostModel.ComputeAllocation: Unknown mode: %s", mode)
- }
- }
- }
- }
- func applyGPUsAllocated(podMap map[podKey]*pod, resGPUsRequested []*prom.QueryResult, resGPUsAllocated []*prom.QueryResult, podUIDKeyMap map[podKey][]podKey) {
- if len(resGPUsAllocated) > 0 { // Use the new query, when it's become available in a window
- resGPUsRequested = resGPUsAllocated
- }
- for _, res := range resGPUsRequested {
- key, err := resultPodKey(res, env.GetPromClusterLabel(), "namespace")
- if err != nil {
- log.DedupedWarningf(10, "CostModel.ComputeAllocation: GPU request result missing field: %s", err)
- continue
- }
- container, err := res.GetString("container")
- if err != nil {
- log.DedupedWarningf(10, "CostModel.ComputeAllocation: GPU request query result missing 'container': %s", key)
- continue
- }
- var pods []*pod
- if thisPod, ok := podMap[key]; !ok {
- if uidKeys, ok := podUIDKeyMap[key]; ok {
- for _, uidKey := range uidKeys {
- thisPod, ok = podMap[uidKey]
- if ok {
- pods = append(pods, thisPod)
- }
- }
- } else {
- continue
- }
- } else {
- pods = []*pod{thisPod}
- }
- for _, thisPod := range pods {
- if _, ok := thisPod.Allocations[container]; !ok {
- thisPod.appendContainer(container)
- }
- hrs := thisPod.Allocations[container].Minutes() / 60.0
- thisPod.Allocations[container].GPUHours = res.Values[0].Value * hrs
- // For now, it will always be the case that Request==Allocation. If
- // you would like to use a GPU you need to request the full GPU.
- // Therefore max(usage,request) will always equal request. In the
- // future this may need to be refactored when building support for
- // GPU Time Slicing.
- if thisPod.Allocations[container].GPUAllocation == nil {
- thisPod.Allocations[container].GPUAllocation = &opencost.GPUAllocation{
- GPURequestAverage: &res.Values[0].Value,
- }
- } else {
- thisPod.Allocations[container].GPUAllocation.GPURequestAverage = &res.Values[0].Value
- }
- }
- }
- }
- func applyNetworkTotals(podMap map[podKey]*pod, resNetworkTransferBytes []*prom.QueryResult, resNetworkReceiveBytes []*prom.QueryResult, podUIDKeyMap map[podKey][]podKey) {
- for _, res := range resNetworkTransferBytes {
- podKey, err := resultPodKey(res, env.GetPromClusterLabel(), "namespace")
- if err != nil {
- log.DedupedWarningf(10, "CostModel.ComputeAllocation: Network Transfer Bytes query result missing field: %s", err)
- continue
- }
- var pods []*pod
- if thisPod, ok := podMap[podKey]; !ok {
- if uidKeys, ok := podUIDKeyMap[podKey]; ok {
- for _, uidKey := range uidKeys {
- thisPod, ok = podMap[uidKey]
- if ok {
- pods = append(pods, thisPod)
- }
- }
- } else {
- continue
- }
- } else {
- pods = []*pod{thisPod}
- }
- for _, thisPod := range pods {
- for _, alloc := range thisPod.Allocations {
- alloc.NetworkTransferBytes = res.Values[0].Value / float64(len(thisPod.Allocations)) / float64(len(pods))
- }
- }
- }
- for _, res := range resNetworkReceiveBytes {
- podKey, err := resultPodKey(res, env.GetPromClusterLabel(), "namespace")
- if err != nil {
- log.DedupedWarningf(10, "CostModel.ComputeAllocation: Network Receive Bytes query result missing field: %s", err)
- continue
- }
- var pods []*pod
- if thisPod, ok := podMap[podKey]; !ok {
- if uidKeys, ok := podUIDKeyMap[podKey]; ok {
- for _, uidKey := range uidKeys {
- thisPod, ok = podMap[uidKey]
- if ok {
- pods = append(pods, thisPod)
- }
- }
- } else {
- continue
- }
- } else {
- pods = []*pod{thisPod}
- }
- for _, thisPod := range pods {
- for _, alloc := range thisPod.Allocations {
- alloc.NetworkReceiveBytes = res.Values[0].Value / float64(len(thisPod.Allocations)) / float64(len(pods))
- }
- }
- }
- }
- func applyNetworkAllocation(podMap map[podKey]*pod, resNetworkGiB []*prom.QueryResult, resNetworkCostPerGiB []*prom.QueryResult, podUIDKeyMap map[podKey][]podKey, networkCostSubType string) {
- costPerGiBByCluster := map[string]float64{}
- for _, res := range resNetworkCostPerGiB {
- cluster, err := res.GetString(env.GetPromClusterLabel())
- if err != nil {
- cluster = env.GetClusterID()
- }
- costPerGiBByCluster[cluster] = res.Values[0].Value
- }
- for _, res := range resNetworkGiB {
- podKey, err := resultPodKey(res, env.GetPromClusterLabel(), "namespace")
- if err != nil {
- log.DedupedWarningf(10, "CostModel.ComputeAllocation: Network allocation query result missing field: %s", err)
- continue
- }
- var pods []*pod
- if thisPod, ok := podMap[podKey]; !ok {
- if uidKeys, ok := podUIDKeyMap[podKey]; ok {
- for _, uidKey := range uidKeys {
- thisPod, ok = podMap[uidKey]
- if ok {
- pods = append(pods, thisPod)
- }
- }
- } else {
- continue
- }
- } else {
- pods = []*pod{thisPod}
- }
- for _, thisPod := range pods {
- for _, alloc := range thisPod.Allocations {
- gib := res.Values[0].Value / float64(len(thisPod.Allocations))
- costPerGiB := costPerGiBByCluster[podKey.Cluster]
- currentNetworkSubCost := gib * costPerGiB / float64(len(pods))
- switch networkCostSubType {
- case networkCrossZoneCost:
- alloc.NetworkCrossZoneCost = currentNetworkSubCost
- case networkCrossRegionCost:
- alloc.NetworkCrossRegionCost = currentNetworkSubCost
- case networkInternetCost:
- alloc.NetworkInternetCost = currentNetworkSubCost
- default:
- log.Warnf("CostModel.applyNetworkAllocation: unknown network subtype passed to the function: %s", networkCostSubType)
- }
- alloc.NetworkCost += currentNetworkSubCost
- }
- }
- }
- }
- func resToNodeLabels(resNodeLabels []*prom.QueryResult) map[nodeKey]map[string]string {
- nodeLabels := map[nodeKey]map[string]string{}
- for _, res := range resNodeLabels {
- nodeKey, err := resultNodeKey(res, env.GetPromClusterLabel(), "node")
- if err != nil {
- continue
- }
- if _, ok := nodeLabels[nodeKey]; !ok {
- nodeLabels[nodeKey] = map[string]string{}
- }
- labels := res.GetLabels()
- // labels are retrieved from prometheus here so it will be in prometheus sanitized state
- // e.g. topology.kubernetes.io/zone => topology_kubernetes_io_zone
- for labelKey, labelValue := range labels {
- nodeLabels[nodeKey][labelKey] = labelValue
- }
- }
- return nodeLabels
- }
- func resToNamespaceLabels(resNamespaceLabels []*prom.QueryResult) map[namespaceKey]map[string]string {
- namespaceLabels := map[namespaceKey]map[string]string{}
- for _, res := range resNamespaceLabels {
- nsKey, err := resultNamespaceKey(res, env.GetPromClusterLabel(), "namespace")
- if err != nil {
- continue
- }
- if _, ok := namespaceLabels[nsKey]; !ok {
- namespaceLabels[nsKey] = map[string]string{}
- }
- for k, l := range res.GetLabels() {
- namespaceLabels[nsKey][k] = l
- }
- }
- return namespaceLabels
- }
- func resToPodLabels(resPodLabels []*prom.QueryResult, podUIDKeyMap map[podKey][]podKey, ingestPodUID bool) map[podKey]map[string]string {
- podLabels := map[podKey]map[string]string{}
- for _, res := range resPodLabels {
- key, err := resultPodKey(res, env.GetPromClusterLabel(), "namespace")
- if err != nil {
- continue
- }
- var keys []podKey
- if ingestPodUID {
- if uidKeys, ok := podUIDKeyMap[key]; ok {
- keys = append(keys, uidKeys...)
- }
- } else {
- keys = []podKey{key}
- }
- for _, key := range keys {
- if _, ok := podLabels[key]; !ok {
- podLabels[key] = map[string]string{}
- }
- for k, l := range res.GetLabels() {
- podLabels[key][k] = l
- }
- }
- }
- return podLabels
- }
- func resToNamespaceAnnotations(resNamespaceAnnotations []*prom.QueryResult) map[string]map[string]string {
- namespaceAnnotations := map[string]map[string]string{}
- for _, res := range resNamespaceAnnotations {
- namespace, err := res.GetString("namespace")
- if err != nil {
- continue
- }
- if _, ok := namespaceAnnotations[namespace]; !ok {
- namespaceAnnotations[namespace] = map[string]string{}
- }
- for k, l := range res.GetAnnotations() {
- namespaceAnnotations[namespace][k] = l
- }
- }
- return namespaceAnnotations
- }
- func resToPodAnnotations(resPodAnnotations []*prom.QueryResult, podUIDKeyMap map[podKey][]podKey, ingestPodUID bool) map[podKey]map[string]string {
- podAnnotations := map[podKey]map[string]string{}
- for _, res := range resPodAnnotations {
- key, err := resultPodKey(res, env.GetPromClusterLabel(), "namespace")
- if err != nil {
- continue
- }
- var keys []podKey
- if ingestPodUID {
- if uidKeys, ok := podUIDKeyMap[key]; ok {
- keys = append(keys, uidKeys...)
- }
- } else {
- keys = []podKey{key}
- }
- for _, key := range keys {
- if _, ok := podAnnotations[key]; !ok {
- podAnnotations[key] = map[string]string{}
- }
- for k, l := range res.GetAnnotations() {
- podAnnotations[key][k] = l
- }
- }
- }
- return podAnnotations
- }
- func applyLabels(podMap map[podKey]*pod, nodeLabels map[nodeKey]map[string]string, namespaceLabels map[namespaceKey]map[string]string, podLabels map[podKey]map[string]string) {
- for podKey, pod := range podMap {
- for _, alloc := range pod.Allocations {
- allocLabels := alloc.Properties.Labels
- if allocLabels == nil {
- allocLabels = make(map[string]string)
- }
- nsLabels := alloc.Properties.NamespaceLabels
- if nsLabels == nil {
- nsLabels = make(map[string]string)
- }
- // Apply node labels first, then namespace labels, then pod labels
- // so that pod labels overwrite namespace labels, which overwrite
- // node labels.
- if nodeLabels != nil {
- nodeKey := newNodeKey(pod.Key.Cluster, pod.Node)
- if labels, ok := nodeLabels[nodeKey]; ok {
- for k, v := range labels {
- allocLabels[k] = v
- }
- }
- }
- nsKey := podKey.namespaceKey
- if labels, ok := namespaceLabels[nsKey]; ok {
- for k, v := range labels {
- allocLabels[k] = v
- nsLabels[k] = v
- }
- }
- if labels, ok := podLabels[podKey]; ok {
- for k, v := range labels {
- allocLabels[k] = v
- }
- }
- alloc.Properties.Labels = allocLabels
- alloc.Properties.NamespaceLabels = nsLabels
- }
- }
- }
- func applyAnnotations(podMap map[podKey]*pod, namespaceAnnotations map[string]map[string]string, podAnnotations map[podKey]map[string]string) {
- for key, pod := range podMap {
- for _, alloc := range pod.Allocations {
- allocAnnotations := alloc.Properties.Annotations
- if allocAnnotations == nil {
- allocAnnotations = make(map[string]string)
- }
- nsAnnotations := alloc.Properties.NamespaceAnnotations
- if nsAnnotations == nil {
- nsAnnotations = make(map[string]string)
- }
- // Apply namespace annotations first, then pod annotations so that
- // pod labels overwrite namespace labels.
- if labels, ok := namespaceAnnotations[key.Namespace]; ok {
- for k, v := range labels {
- allocAnnotations[k] = v
- nsAnnotations[k] = v
- }
- }
- if labels, ok := podAnnotations[key]; ok {
- for k, v := range labels {
- allocAnnotations[k] = v
- }
- }
- alloc.Properties.Annotations = allocAnnotations
- alloc.Properties.NamespaceAnnotations = nsAnnotations
- }
- }
- }
- func resToDeploymentLabels(resDeploymentLabels []*prom.QueryResult) map[controllerKey]map[string]string {
- deploymentLabels := map[controllerKey]map[string]string{}
- for _, res := range resDeploymentLabels {
- controllerKey, err := resultDeploymentKey(res, env.GetPromClusterLabel(), "namespace", "deployment")
- if err != nil {
- continue
- }
- if _, ok := deploymentLabels[controllerKey]; !ok {
- deploymentLabels[controllerKey] = map[string]string{}
- }
- for k, l := range res.GetLabels() {
- deploymentLabels[controllerKey][k] = l
- }
- }
- // Prune duplicate deployments. That is, if the same deployment exists with
- // hyphens instead of underscores, keep the one that uses hyphens.
- for key := range deploymentLabels {
- if strings.Contains(key.Controller, "_") {
- duplicateController := strings.Replace(key.Controller, "_", "-", -1)
- duplicateKey := newControllerKey(key.Cluster, key.Namespace, key.ControllerKind, duplicateController)
- if _, ok := deploymentLabels[duplicateKey]; ok {
- delete(deploymentLabels, key)
- }
- }
- }
- return deploymentLabels
- }
- func resToStatefulSetLabels(resStatefulSetLabels []*prom.QueryResult) map[controllerKey]map[string]string {
- statefulSetLabels := map[controllerKey]map[string]string{}
- for _, res := range resStatefulSetLabels {
- controllerKey, err := resultStatefulSetKey(res, env.GetPromClusterLabel(), "namespace", "statefulSet")
- if err != nil {
- continue
- }
- if _, ok := statefulSetLabels[controllerKey]; !ok {
- statefulSetLabels[controllerKey] = map[string]string{}
- }
- for k, l := range res.GetLabels() {
- statefulSetLabels[controllerKey][k] = l
- }
- }
- // Prune duplicate stateful sets. That is, if the same stateful set exists
- // with hyphens instead of underscores, keep the one that uses hyphens.
- for key := range statefulSetLabels {
- if strings.Contains(key.Controller, "_") {
- duplicateController := strings.Replace(key.Controller, "_", "-", -1)
- duplicateKey := newControllerKey(key.Cluster, key.Namespace, key.ControllerKind, duplicateController)
- if _, ok := statefulSetLabels[duplicateKey]; ok {
- delete(statefulSetLabels, key)
- }
- }
- }
- return statefulSetLabels
- }
- func labelsToPodControllerMap(podLabels map[podKey]map[string]string, controllerLabels map[controllerKey]map[string]string) map[podKey]controllerKey {
- podControllerMap := map[podKey]controllerKey{}
- // For each controller, turn the labels into a selector and attempt to
- // match it with each set of pod labels. A match indicates that the pod
- // belongs to the controller.
- for cKey, cLabels := range controllerLabels {
- selector := labels.Set(cLabels).AsSelectorPreValidated()
- for pKey, pLabels := range podLabels {
- // If the pod is in a different cluster or namespace, there is
- // no need to compare the labels.
- if cKey.Cluster != pKey.Cluster || cKey.Namespace != pKey.Namespace {
- continue
- }
- podLabelSet := labels.Set(pLabels)
- if selector.Matches(podLabelSet) {
- if _, ok := podControllerMap[pKey]; ok {
- log.DedupedWarningf(5, "CostModel.ComputeAllocation: PodControllerMap match already exists: %s matches %s and %s", pKey, podControllerMap[pKey], cKey)
- }
- podControllerMap[pKey] = cKey
- }
- }
- }
- return podControllerMap
- }
- func resToPodDaemonSetMap(resDaemonSetLabels []*prom.QueryResult, podUIDKeyMap map[podKey][]podKey, ingestPodUID bool) map[podKey]controllerKey {
- daemonSetLabels := map[podKey]controllerKey{}
- for _, res := range resDaemonSetLabels {
- controllerKey, err := resultDaemonSetKey(res, env.GetPromClusterLabel(), "namespace", "owner_name")
- if err != nil {
- continue
- }
- pod, err := res.GetString("pod")
- if err != nil {
- log.Warnf("CostModel.ComputeAllocation: DaemonSetLabel result without pod: %s", controllerKey)
- }
- key := newPodKey(controllerKey.Cluster, controllerKey.Namespace, pod)
- var keys []podKey
- if ingestPodUID {
- if uidKeys, ok := podUIDKeyMap[key]; ok {
- keys = append(keys, uidKeys...)
- }
- } else {
- keys = []podKey{key}
- }
- for _, key := range keys {
- daemonSetLabels[key] = controllerKey
- }
- }
- return daemonSetLabels
- }
- func resToPodJobMap(resJobLabels []*prom.QueryResult, podUIDKeyMap map[podKey][]podKey, ingestPodUID bool) map[podKey]controllerKey {
- jobLabels := map[podKey]controllerKey{}
- for _, res := range resJobLabels {
- controllerKey, err := resultJobKey(res, env.GetPromClusterLabel(), "namespace", "owner_name")
- if err != nil {
- continue
- }
- // Convert the name of Jobs generated by CronJobs to the name of the
- // CronJob by stripping the timestamp off the end.
- match := isCron.FindStringSubmatch(controllerKey.Controller)
- if match != nil {
- controllerKey.Controller = match[1]
- }
- pod, err := res.GetString("pod")
- if err != nil {
- log.Warnf("CostModel.ComputeAllocation: JobLabel result without pod: %s", controllerKey)
- }
- key := newPodKey(controllerKey.Cluster, controllerKey.Namespace, pod)
- var keys []podKey
- if ingestPodUID {
- if uidKeys, ok := podUIDKeyMap[key]; ok {
- keys = append(keys, uidKeys...)
- }
- } else {
- keys = []podKey{key}
- }
- for _, key := range keys {
- jobLabels[key] = controllerKey
- }
- }
- return jobLabels
- }
- func resToPodReplicaSetMap(resPodsWithReplicaSetOwner []*prom.QueryResult, resReplicaSetsWithoutOwners []*prom.QueryResult, resReplicaSetsWithRolloutOwner []*prom.QueryResult, podUIDKeyMap map[podKey][]podKey, ingestPodUID bool) map[podKey]controllerKey {
- // Build out set of ReplicaSets that have no owners, themselves, such that
- // the ReplicaSet should be used as the owner of the Pods it controls.
- // (This should exclude, for example, ReplicaSets that are controlled by
- // Deployments, in which case the Deployment should be the pod's owner.)
- // Additionally, add to this set of ReplicaSets those ReplicaSets that
- // are owned by a Rollout
- replicaSets := map[controllerKey]struct{}{}
- // Create unowned ReplicaSet controller keys
- for _, res := range resReplicaSetsWithoutOwners {
- controllerKey, err := resultReplicaSetKey(res, env.GetPromClusterLabel(), "namespace", "replicaset")
- if err != nil {
- continue
- }
- replicaSets[controllerKey] = struct{}{}
- }
- // Create Rollout-owned ReplicaSet controller keys
- for _, res := range resReplicaSetsWithRolloutOwner {
- controllerKey, err := resultReplicaSetRolloutKey(res, env.GetPromClusterLabel(), "namespace", "replicaset")
- if err != nil {
- continue
- }
- replicaSets[controllerKey] = struct{}{}
- }
- // Create the mapping of Pods to ReplicaSets, ignoring any ReplicaSets that
- // do not appear in the set of unowned/Rollout-owned ReplicaSets above.
- podToReplicaSet := map[podKey]controllerKey{}
- for _, res := range resPodsWithReplicaSetOwner {
- // First, check if this pod is owned by an unowned ReplicaSet
- controllerKey, err := resultReplicaSetKey(res, env.GetPromClusterLabel(), "namespace", "owner_name")
- if err != nil {
- continue
- } else if _, ok := replicaSets[controllerKey]; !ok {
- // If the pod is not owned by an unowned ReplicaSet, check if
- // it's owned by a Rollout-owned ReplicaSet
- controllerKey, err = resultReplicaSetRolloutKey(res, env.GetPromClusterLabel(), "namespace", "owner_name")
- if err != nil {
- continue
- } else if _, ok := replicaSets[controllerKey]; !ok {
- continue
- }
- }
- pod, err := res.GetString("pod")
- if err != nil {
- log.Warnf("CostModel.ComputeAllocation: ReplicaSet result without pod: %s", controllerKey)
- }
- key := newPodKey(controllerKey.Cluster, controllerKey.Namespace, pod)
- var keys []podKey
- if ingestPodUID {
- if uidKeys, ok := podUIDKeyMap[key]; ok {
- keys = append(keys, uidKeys...)
- }
- } else {
- keys = []podKey{key}
- }
- for _, key := range keys {
- podToReplicaSet[key] = controllerKey
- }
- }
- return podToReplicaSet
- }
- func applyControllersToPods(podMap map[podKey]*pod, podControllerMap map[podKey]controllerKey) {
- for key, pod := range podMap {
- for _, alloc := range pod.Allocations {
- if controllerKey, ok := podControllerMap[key]; ok {
- alloc.Properties.ControllerKind = controllerKey.ControllerKind
- alloc.Properties.Controller = controllerKey.Controller
- }
- }
- }
- }
- /* Service Helpers */
- func getServiceLabels(resServiceLabels []*prom.QueryResult) map[serviceKey]map[string]string {
- serviceLabels := map[serviceKey]map[string]string{}
- for _, res := range resServiceLabels {
- serviceKey, err := resultServiceKey(res, env.GetPromClusterLabel(), "namespace", "service")
- if err != nil {
- continue
- }
- if _, ok := serviceLabels[serviceKey]; !ok {
- serviceLabels[serviceKey] = map[string]string{}
- }
- for k, l := range res.GetLabels() {
- serviceLabels[serviceKey][k] = l
- }
- }
- // Prune duplicate services. That is, if the same service exists with
- // hyphens instead of underscores, keep the one that uses hyphens.
- for key := range serviceLabels {
- if strings.Contains(key.Service, "_") {
- duplicateService := strings.Replace(key.Service, "_", "-", -1)
- duplicateKey := newServiceKey(key.Cluster, key.Namespace, duplicateService)
- if _, ok := serviceLabels[duplicateKey]; ok {
- delete(serviceLabels, key)
- }
- }
- }
- return serviceLabels
- }
- func applyServicesToPods(podMap map[podKey]*pod, podLabels map[podKey]map[string]string, allocsByService map[serviceKey][]*opencost.Allocation, serviceLabels map[serviceKey]map[string]string) {
- podServicesMap := map[podKey][]serviceKey{}
- // For each service, turn the labels into a selector and attempt to
- // match it with each set of pod labels. A match indicates that the pod
- // belongs to the service.
- for sKey, sLabels := range serviceLabels {
- selector := labels.Set(sLabels).AsSelectorPreValidated()
- for pKey, pLabels := range podLabels {
- // If the pod is in a different cluster or namespace, there is
- // no need to compare the labels.
- if sKey.Cluster != pKey.Cluster || sKey.Namespace != pKey.Namespace {
- continue
- }
- podLabelSet := labels.Set(pLabels)
- if selector.Matches(podLabelSet) {
- if _, ok := podServicesMap[pKey]; !ok {
- podServicesMap[pKey] = []serviceKey{}
- }
- podServicesMap[pKey] = append(podServicesMap[pKey], sKey)
- }
- }
- }
- // For each allocation in each pod, attempt to find and apply the list of
- // services associated with the allocation's pod.
- for key, pod := range podMap {
- for _, alloc := range pod.Allocations {
- if sKeys, ok := podServicesMap[key]; ok {
- services := []string{}
- for _, sKey := range sKeys {
- services = append(services, sKey.Service)
- allocsByService[sKey] = append(allocsByService[sKey], alloc)
- }
- alloc.Properties.Services = services
- }
- }
- }
- }
- func getLoadBalancerCosts(lbMap map[serviceKey]*lbCost, resLBCost, resLBActiveMins []*prom.QueryResult, resolution time.Duration, window opencost.Window) {
- for _, res := range resLBActiveMins {
- serviceKey, err := resultServiceKey(res, env.GetPromClusterLabel(), "namespace", "service_name")
- if err != nil || len(res.Values) == 0 {
- continue
- }
- lbStart, lbEnd := calculateStartAndEnd(res, resolution, window)
- if lbStart.IsZero() || lbEnd.IsZero() {
- log.Warnf("CostModel.ComputeAllocation: pvc %s has no running time", serviceKey)
- }
- lbMap[serviceKey] = &lbCost{
- Start: lbStart,
- End: lbEnd,
- }
- }
- for _, res := range resLBCost {
- serviceKey, err := resultServiceKey(res, env.GetPromClusterLabel(), "namespace", "service_name")
- if err != nil {
- continue
- }
- // get the ingress IP to determine if this is a private LB
- ip, err := res.GetString("ingress_ip")
- if err != nil {
- log.Warnf("error getting ingress ip for key %s: %v, skipping", serviceKey, err)
- // do not count the time that the service was being created or deleted
- // ingress IP will be empty string
- // only add cost to allocation when external IP is provisioned
- if ip == "" {
- continue
- }
- }
- // Apply cost as price-per-hour * hours
- if lb, ok := lbMap[serviceKey]; ok {
- lbPricePerHr := res.Values[0].Value
- // interpolate any missing data
- resolutionHours := resolution.Hours()
- resultHours := lb.End.Sub(lb.Start).Hours()
- scaleFactor := (resolutionHours + resultHours) / resultHours
- // after scaling, we can adjust the timings to reflect the interpolated data
- lb.End = lb.End.Add(resolution)
- lb.TotalCost += lbPricePerHr * resultHours * scaleFactor
- lb.Ip = ip
- lb.Private = privateIPCheck(ip)
- } else {
- log.DedupedWarningf(20, "CostModel: found minutes for key that does not exist: %s", serviceKey)
- }
- }
- }
- func applyLoadBalancersToPods(window opencost.Window, podMap map[podKey]*pod, lbMap map[serviceKey]*lbCost, allocsByService map[serviceKey][]*opencost.Allocation) {
- for sKey, lb := range lbMap {
- totalHours := 0.0
- allocHours := make(map[*opencost.Allocation]float64)
- allocs, ok := allocsByService[sKey]
- // if there are no allocations using the service, add its cost to the Unmounted pod for its cluster
- if !ok {
- pod := getUnmountedPodForCluster(window, podMap, sKey.Cluster)
- pod.Allocations[opencost.UnmountedSuffix].LoadBalancerCost += lb.TotalCost
- pod.Allocations[opencost.UnmountedSuffix].Properties.Services = append(pod.Allocations[opencost.UnmountedSuffix].Properties.Services, sKey.Service)
- }
- // Add portion of load balancing cost to each allocation
- // proportional to the total number of hours allocations used the load balancer
- for _, alloc := range allocs {
- // Determine the (start, end) of the relationship between the
- // given lbCost and the associated Allocation so that a precise
- // number of hours can be used to compute cumulative cost.
- s, e := alloc.Start, alloc.End
- if lb.Start.After(alloc.Start) {
- s = lb.Start
- }
- if lb.End.Before(alloc.End) {
- e = lb.End
- }
- hours := e.Sub(s).Hours()
- // A negative number of hours signifies no overlap between the windows
- if hours > 0 {
- totalHours += hours
- allocHours[alloc] = hours
- }
- }
- // Distribute cost of service once total hours is calculated
- for alloc, hours := range allocHours {
- alloc.LoadBalancerCost += lb.TotalCost * hours / totalHours
- }
- for _, alloc := range allocs {
- // reocord the hours overlapped with the allocation for the load balancer
- // if there was overlap. Otherwise, record a 0.0.
- // TODO: Do we really want to include load balancers that have 0 overlap
- // TODO: hours with the allocation?
- var hours float64 = 0.0
- if _, ok := allocHours[alloc]; ok {
- hours = allocHours[alloc]
- }
- if alloc.LoadBalancers == nil {
- alloc.LoadBalancers = opencost.LbAllocations{}
- }
- if _, found := alloc.LoadBalancers[sKey.String()]; found {
- alloc.LoadBalancers[sKey.String()].Cost += alloc.LoadBalancerCost
- alloc.LoadBalancers[sKey.String()].Hours += hours
- } else {
- alloc.LoadBalancers[sKey.String()] = &opencost.LbAllocation{
- Service: sKey.Namespace + "/" + sKey.Service,
- Cost: alloc.LoadBalancerCost,
- Private: lb.Private,
- Ip: lb.Ip,
- Hours: hours,
- }
- }
- }
- // If there was no overlap apply to Unmounted pod
- if len(allocHours) == 0 {
- pod := getUnmountedPodForCluster(window, podMap, sKey.Cluster)
- pod.Allocations[opencost.UnmountedSuffix].LoadBalancerCost += lb.TotalCost
- pod.Allocations[opencost.UnmountedSuffix].Properties.Services = append(pod.Allocations[opencost.UnmountedSuffix].Properties.Services, sKey.Service)
- }
- }
- }
- /* Node Helpers */
- func applyNodeCostPerCPUHr(nodeMap map[nodeKey]*nodePricing, resNodeCostPerCPUHr []*prom.QueryResult) {
- for _, res := range resNodeCostPerCPUHr {
- cluster, err := res.GetString(env.GetPromClusterLabel())
- if err != nil {
- cluster = env.GetClusterID()
- }
- node, err := res.GetString("node")
- if err != nil {
- log.Warnf("CostModel.ComputeAllocation: Node CPU cost query result missing field: \"%s\" for node \"%s\"", err, node)
- continue
- }
- instanceType, err := res.GetString("instance_type")
- if err != nil {
- log.Warnf("CostModel.ComputeAllocation: Node CPU cost query result missing field: \"%s\" for node \"%s\"", err, node)
- }
- providerID, err := res.GetString("provider_id")
- if err != nil {
- log.Warnf("CostModel.ComputeAllocation: Node CPU cost query result missing field: \"%s\" for node \"%s\"", err, node)
- }
- key := newNodeKey(cluster, node)
- if _, ok := nodeMap[key]; !ok {
- nodeMap[key] = &nodePricing{
- Name: node,
- NodeType: instanceType,
- ProviderID: provider.ParseID(providerID),
- }
- }
- nodeMap[key].CostPerCPUHr = res.Values[0].Value
- }
- }
- func applyNodeCostPerRAMGiBHr(nodeMap map[nodeKey]*nodePricing, resNodeCostPerRAMGiBHr []*prom.QueryResult) {
- for _, res := range resNodeCostPerRAMGiBHr {
- cluster, err := res.GetString(env.GetPromClusterLabel())
- if err != nil {
- cluster = env.GetClusterID()
- }
- node, err := res.GetString("node")
- if err != nil {
- log.Warnf("CostModel.ComputeAllocation: Node RAM cost query result missing field: \"%s\" for node \"%s\"", err, node)
- continue
- }
- instanceType, err := res.GetString("instance_type")
- if err != nil {
- log.Warnf("CostModel.ComputeAllocation: Node RAM cost query result missing field: \"%s\" for node \"%s\"", err, node)
- }
- providerID, err := res.GetString("provider_id")
- if err != nil {
- log.Warnf("CostModel.ComputeAllocation: Node RAM cost query result missing field: \"%s\" for node \"%s\"", err, node)
- }
- key := newNodeKey(cluster, node)
- if _, ok := nodeMap[key]; !ok {
- nodeMap[key] = &nodePricing{
- Name: node,
- NodeType: instanceType,
- ProviderID: provider.ParseID(providerID),
- }
- }
- nodeMap[key].CostPerRAMGiBHr = res.Values[0].Value
- }
- }
- func applyNodeCostPerGPUHr(nodeMap map[nodeKey]*nodePricing, resNodeCostPerGPUHr []*prom.QueryResult) {
- for _, res := range resNodeCostPerGPUHr {
- cluster, err := res.GetString(env.GetPromClusterLabel())
- if err != nil {
- cluster = env.GetClusterID()
- }
- node, err := res.GetString("node")
- if err != nil {
- log.Warnf("CostModel.ComputeAllocation: Node GPU cost query result missing field: \"%s\" for node \"%s\"", err, node)
- continue
- }
- instanceType, err := res.GetString("instance_type")
- if err != nil {
- log.Warnf("CostModel.ComputeAllocation: Node GPU cost query result missing field: \"%s\" for node \"%s\"", err, node)
- }
- providerID, err := res.GetString("provider_id")
- if err != nil {
- log.Warnf("CostModel.ComputeAllocation: Node GPU cost query result missing field: \"%s\" for node \"%s\"", err, node)
- }
- key := newNodeKey(cluster, node)
- if _, ok := nodeMap[key]; !ok {
- nodeMap[key] = &nodePricing{
- Name: node,
- NodeType: instanceType,
- ProviderID: provider.ParseID(providerID),
- }
- }
- nodeMap[key].CostPerGPUHr = res.Values[0].Value
- }
- }
- func applyNodeSpot(nodeMap map[nodeKey]*nodePricing, resNodeIsSpot []*prom.QueryResult) {
- for _, res := range resNodeIsSpot {
- cluster, err := res.GetString(env.GetPromClusterLabel())
- if err != nil {
- cluster = env.GetClusterID()
- }
- node, err := res.GetString("node")
- if err != nil {
- log.Warnf("CostModel.ComputeAllocation: Node spot query result missing field: %s", err)
- continue
- }
- key := newNodeKey(cluster, node)
- if _, ok := nodeMap[key]; !ok {
- log.Warnf("CostModel.ComputeAllocation: Node spot query result for missing node: %s", key)
- continue
- }
- nodeMap[key].Preemptible = res.Values[0].Value > 0
- }
- }
- func applyNodeDiscount(nodeMap map[nodeKey]*nodePricing, cm *CostModel) {
- if cm == nil {
- return
- }
- c, err := cm.Provider.GetConfig()
- if err != nil {
- log.Errorf("CostModel.ComputeAllocation: applyNodeDiscount: %s", err)
- return
- }
- discount, err := ParsePercentString(c.Discount)
- if err != nil {
- log.Errorf("CostModel.ComputeAllocation: applyNodeDiscount: %s", err)
- return
- }
- negotiatedDiscount, err := ParsePercentString(c.NegotiatedDiscount)
- if err != nil {
- log.Errorf("CostModel.ComputeAllocation: applyNodeDiscount: %s", err)
- return
- }
- for _, node := range nodeMap {
- // TODO GKE Reserved Instances into account
- node.Discount = cm.Provider.CombinedDiscountForNode(node.NodeType, node.Preemptible, discount, negotiatedDiscount)
- node.CostPerCPUHr *= (1.0 - node.Discount)
- node.CostPerRAMGiBHr *= (1.0 - node.Discount)
- }
- }
- func (cm *CostModel) applyNodesToPod(podMap map[podKey]*pod, nodeMap map[nodeKey]*nodePricing) {
- for _, pod := range podMap {
- for _, alloc := range pod.Allocations {
- cluster := alloc.Properties.Cluster
- nodeName := alloc.Properties.Node
- thisNodeKey := newNodeKey(cluster, nodeName)
- node := cm.getNodePricing(nodeMap, thisNodeKey)
- alloc.Properties.ProviderID = node.ProviderID
- alloc.CPUCost = alloc.CPUCoreHours * node.CostPerCPUHr
- alloc.RAMCost = (alloc.RAMByteHours / 1024 / 1024 / 1024) * node.CostPerRAMGiBHr
- alloc.GPUCost = alloc.GPUHours * node.CostPerGPUHr
- }
- }
- }
- // getCustomNodePricing converts the CostModel's configured custom pricing
- // values into a nodePricing instance.
- func (cm *CostModel) getCustomNodePricing(spot bool, providerID string) *nodePricing {
- customPricingConfig, err := cm.Provider.GetConfig()
- if err != nil {
- return nil
- }
- cpuCostStr := customPricingConfig.CPU
- gpuCostStr := customPricingConfig.GPU
- ramCostStr := customPricingConfig.RAM
- if spot {
- cpuCostStr = customPricingConfig.SpotCPU
- gpuCostStr = customPricingConfig.SpotGPU
- ramCostStr = customPricingConfig.SpotRAM
- }
- node := &nodePricing{
- Source: "custom",
- ProviderID: providerID,
- }
- costPerCPUHr, err := strconv.ParseFloat(cpuCostStr, 64)
- if err != nil {
- log.Warnf("CostModel: custom pricing has illegal CPU cost: %s", cpuCostStr)
- }
- node.CostPerCPUHr = costPerCPUHr
- costPerGPUHr, err := strconv.ParseFloat(gpuCostStr, 64)
- if err != nil {
- log.Warnf("CostModel: custom pricing has illegal GPU cost: %s", gpuCostStr)
- }
- node.CostPerGPUHr = costPerGPUHr
- costPerRAMHr, err := strconv.ParseFloat(ramCostStr, 64)
- if err != nil {
- log.Warnf("CostModel: custom pricing has illegal RAM cost: %s", ramCostStr)
- }
- node.CostPerRAMGiBHr = costPerRAMHr
- return node
- }
- // getNodePricing determines node pricing, given a key and a mapping from keys
- // to their nodePricing instances, as well as the custom pricing configuration
- // inherent to the CostModel instance. If custom pricing is set, use that. If
- // not, use the pricing defined by the given key. If that doesn't exist, fall
- // back on custom pricing as a default.
- func (cm *CostModel) getNodePricing(nodeMap map[nodeKey]*nodePricing, nodeKey nodeKey) *nodePricing {
- // Find the relevant nodePricing, if it exists. If not, substitute the
- // custom nodePricing as a default.
- node, ok := nodeMap[nodeKey]
- if !ok || node == nil {
- if nodeKey.Node != "" {
- log.DedupedWarningf(5, "CostModel: failed to find node for %s", nodeKey)
- }
- // since the node pricing data is not found, and this won't change for the duration of the allocation
- // build process, we can update the node map with the defaults to prevent future failed lookups
- nodeMap[nodeKey] = cm.getCustomNodePricing(false, "")
- return nodeMap[nodeKey]
- }
- // If custom pricing is enabled and can be retrieved, override detected
- // node pricing with the custom values.
- customPricingConfig, err := cm.Provider.GetConfig()
- if err != nil {
- log.Warnf("CostModel: failed to load custom pricing: %s", err)
- }
- if provider.CustomPricesEnabled(cm.Provider) && customPricingConfig != nil {
- return cm.getCustomNodePricing(node.Preemptible, node.ProviderID)
- }
- node.Source = "prometheus"
- // If any of the values are NaN or zero, replace them with the custom
- // values as default.
- // TODO:CLEANUP can't we parse these custom prices once? why do we store
- // them as strings like this?
- if node.CostPerCPUHr == 0 || math.IsNaN(node.CostPerCPUHr) {
- log.Warnf("CostModel: node pricing has illegal CostPerCPUHr; replacing with custom pricing: %s", nodeKey)
- cpuCostStr := customPricingConfig.CPU
- if node.Preemptible {
- cpuCostStr = customPricingConfig.SpotCPU
- }
- costPerCPUHr, err := strconv.ParseFloat(cpuCostStr, 64)
- if err != nil {
- log.Warnf("CostModel: custom pricing has illegal CPU cost: %s", cpuCostStr)
- }
- node.CostPerCPUHr = costPerCPUHr
- node.Source += "/customCPU"
- }
- if math.IsNaN(node.CostPerGPUHr) {
- log.Warnf("CostModel: node pricing has illegal CostPerGPUHr; replacing with custom pricing: %s", nodeKey)
- gpuCostStr := customPricingConfig.GPU
- if node.Preemptible {
- gpuCostStr = customPricingConfig.SpotGPU
- }
- costPerGPUHr, err := strconv.ParseFloat(gpuCostStr, 64)
- if err != nil {
- log.Warnf("CostModel: custom pricing has illegal GPU cost: %s", gpuCostStr)
- }
- node.CostPerGPUHr = costPerGPUHr
- node.Source += "/customGPU"
- }
- if node.CostPerRAMGiBHr == 0 || math.IsNaN(node.CostPerRAMGiBHr) {
- log.Warnf("CostModel: node pricing has illegal CostPerRAMHr; replacing with custom pricing: %s", nodeKey)
- ramCostStr := customPricingConfig.RAM
- if node.Preemptible {
- ramCostStr = customPricingConfig.SpotRAM
- }
- costPerRAMHr, err := strconv.ParseFloat(ramCostStr, 64)
- if err != nil {
- log.Warnf("CostModel: custom pricing has illegal RAM cost: %s", ramCostStr)
- }
- node.CostPerRAMGiBHr = costPerRAMHr
- node.Source += "/customRAM"
- }
- // Double check each for NaNs, as there is a chance that our custom pricing
- // config could, itself, contain NaNs...
- if math.IsNaN(node.CostPerCPUHr) || math.IsInf(node.CostPerCPUHr, 0) {
- log.Warnf("CostModel: %s: node pricing has illegal CPU value: %v (setting to 0.0)", nodeKey, node.CostPerCPUHr)
- node.CostPerCPUHr = 0.0
- }
- if math.IsNaN(node.CostPerGPUHr) || math.IsInf(node.CostPerGPUHr, 0) {
- log.Warnf("CostModel: %s: node pricing has illegal RAM value: %v (setting to 0.0)", nodeKey, node.CostPerGPUHr)
- node.CostPerGPUHr = 0.0
- }
- if math.IsNaN(node.CostPerRAMGiBHr) || math.IsInf(node.CostPerRAMGiBHr, 0) {
- log.Warnf("CostModel: %s: node pricing has illegal RAM value: %v (setting to 0.0)", nodeKey, node.CostPerRAMGiBHr)
- node.CostPerRAMGiBHr = 0.0
- }
- return node
- }
- /* PV/PVC Helpers */
- func buildPVMap(resolution time.Duration, pvMap map[pvKey]*pv, resPVCostPerGiBHour, resPVActiveMins, resPVMeta []*prom.QueryResult, window opencost.Window) {
- for _, result := range resPVActiveMins {
- key, err := resultPVKey(result, env.GetPromClusterLabel(), "persistentvolume")
- if err != nil {
- log.Warnf("CostModel.ComputeAllocation: pv bytes query result missing field: %s", err)
- continue
- }
- pvStart, pvEnd := calculateStartAndEnd(result, resolution, window)
- if pvStart.IsZero() || pvEnd.IsZero() {
- log.Warnf("CostModel.ComputeAllocation: pv %s has no running time", key)
- }
- pvMap[key] = &pv{
- Cluster: key.Cluster,
- Name: key.PersistentVolume,
- Start: pvStart,
- End: pvEnd,
- }
- }
- for _, result := range resPVCostPerGiBHour {
- key, err := resultPVKey(result, env.GetPromClusterLabel(), "volumename")
- if err != nil {
- log.Warnf("CostModel.ComputeAllocation: thisPV bytes query result missing field: %s", err)
- continue
- }
- if _, ok := pvMap[key]; !ok {
- pvMap[key] = &pv{
- Cluster: key.Cluster,
- Name: key.PersistentVolume,
- }
- }
- pvMap[key].CostPerGiBHour = result.Values[0].Value
- }
- for _, result := range resPVMeta {
- key, err := resultPVKey(result, env.GetPromClusterLabel(), "persistentvolume")
- if err != nil {
- log.Warnf("error getting key for PV: %v", err)
- continue
- }
- // only add metadata for disks that exist in the other metrics
- if _, ok := pvMap[key]; ok {
- provId, err := result.GetString("provider_id")
- if err != nil {
- log.Warnf("error getting provider id for PV %v: %v", key, err)
- continue
- }
- pvMap[key].ProviderID = provId
- }
- }
- }
- func applyPVBytes(pvMap map[pvKey]*pv, resPVBytes []*prom.QueryResult) {
- for _, res := range resPVBytes {
- key, err := resultPVKey(res, env.GetPromClusterLabel(), "persistentvolume")
- if err != nil {
- log.Warnf("CostModel.ComputeAllocation: pv bytes query result missing field: %s", err)
- continue
- }
- if _, ok := pvMap[key]; !ok {
- log.Warnf("CostModel.ComputeAllocation: pv bytes result for missing pv: %s", key)
- continue
- }
- pvBytesUsed := res.Values[0].Value
- if pvBytesUsed < PV_USAGE_SANITY_LIMIT_BYTES {
- pvMap[key].Bytes = pvBytesUsed
- } else {
- pvMap[key].Bytes = 0
- log.Warnf("PV usage exceeds sanity limit, clamping to zero")
- }
- }
- }
- func buildPVCMap(resolution time.Duration, pvcMap map[pvcKey]*pvc, pvMap map[pvKey]*pv, resPVCInfo []*prom.QueryResult, window opencost.Window) {
- for _, res := range resPVCInfo {
- cluster, err := res.GetString(env.GetPromClusterLabel())
- if err != nil {
- cluster = env.GetClusterID()
- }
- values, err := res.GetStrings("persistentvolumeclaim", "storageclass", "volumename", "namespace")
- if err != nil {
- log.DedupedWarningf(10, "CostModel.ComputeAllocation: pvc info query result missing field: %s", err)
- continue
- }
- namespace := values["namespace"]
- name := values["persistentvolumeclaim"]
- volume := values["volumename"]
- storageClass := values["storageclass"]
- pvKey := newPVKey(cluster, volume)
- pvcKey := newPVCKey(cluster, namespace, name)
- pvcStart, pvcEnd := calculateStartAndEnd(res, resolution, window)
- if pvcStart.IsZero() || pvcEnd.IsZero() {
- log.Warnf("CostModel.ComputeAllocation: pvc %s has no running time", pvcKey)
- }
- if _, ok := pvMap[pvKey]; !ok {
- continue
- }
- pvMap[pvKey].StorageClass = storageClass
- if _, ok := pvcMap[pvcKey]; !ok {
- pvcMap[pvcKey] = &pvc{}
- }
- pvcMap[pvcKey].Name = name
- pvcMap[pvcKey].Namespace = namespace
- pvcMap[pvcKey].Cluster = cluster
- pvcMap[pvcKey].Volume = pvMap[pvKey]
- pvcMap[pvcKey].Start = pvcStart
- pvcMap[pvcKey].End = pvcEnd
- }
- }
- func applyPVCBytesRequested(pvcMap map[pvcKey]*pvc, resPVCBytesRequested []*prom.QueryResult) {
- for _, res := range resPVCBytesRequested {
- key, err := resultPVCKey(res, env.GetPromClusterLabel(), "namespace", "persistentvolumeclaim")
- if err != nil {
- continue
- }
- if _, ok := pvcMap[key]; !ok {
- continue
- }
- pvcMap[key].Bytes = res.Values[0].Value
- }
- }
- func buildPodPVCMap(podPVCMap map[podKey][]*pvc, pvMap map[pvKey]*pv, pvcMap map[pvcKey]*pvc, podMap map[podKey]*pod, resPodPVCAllocation []*prom.QueryResult, podUIDKeyMap map[podKey][]podKey, ingestPodUID bool) {
- for _, res := range resPodPVCAllocation {
- cluster, err := res.GetString(env.GetPromClusterLabel())
- if err != nil {
- cluster = env.GetClusterID()
- }
- values, err := res.GetStrings("persistentvolume", "persistentvolumeclaim", "pod", "namespace")
- if err != nil {
- log.DedupedWarningf(5, "CostModel.ComputeAllocation: pvc allocation query result missing field: %s", err)
- continue
- }
- namespace := values["namespace"]
- pod := values["pod"]
- name := values["persistentvolumeclaim"]
- volume := values["persistentvolume"]
- key := newPodKey(cluster, namespace, pod)
- pvKey := newPVKey(cluster, volume)
- pvcKey := newPVCKey(cluster, namespace, name)
- var keys []podKey
- if ingestPodUID {
- if uidKeys, ok := podUIDKeyMap[key]; ok {
- keys = append(keys, uidKeys...)
- }
- } else {
- keys = []podKey{key}
- }
- for _, key := range keys {
- if _, ok := pvMap[pvKey]; !ok {
- log.DedupedWarningf(5, "CostModel.ComputeAllocation: pv missing for pvc allocation query result: %s", pvKey)
- continue
- }
- if _, ok := podPVCMap[key]; !ok {
- podPVCMap[key] = []*pvc{}
- }
- pvc, ok := pvcMap[pvcKey]
- if !ok {
- log.DedupedWarningf(5, "CostModel.ComputeAllocation: pvc missing for pvc allocation query: %s", pvcKey)
- continue
- }
- if pod, ok := podMap[key]; !ok || len(pod.Allocations) <= 0 {
- log.DedupedWarningf(10, "CostModel.ComputeAllocation: pvc %s for missing pod %s", pvcKey, key)
- continue
- }
- pvc.Mounted = true
- podPVCMap[key] = append(podPVCMap[key], pvc)
- }
- }
- }
- func applyPVCsToPods(window opencost.Window, podMap map[podKey]*pod, podPVCMap map[podKey][]*pvc, pvcMap map[pvcKey]*pvc) {
- // Because PVCs can be shared among pods, the respective pv cost
- // needs to be evenly distributed to those pods based on time
- // running, as well as the amount of time the pvc was shared.
- // Build a relation between every pvc to the pods that mount it
- // and a window representing the interval during which they
- // were associated.
- pvcPodWindowMap := make(map[pvcKey]map[podKey]opencost.Window)
- for thisPodKey, thisPod := range podMap {
- if pvcs, ok := podPVCMap[thisPodKey]; ok {
- for _, thisPVC := range pvcs {
- // Determine the (start, end) of the relationship between the
- // given pvc and the associated Allocation so that a precise
- // number of hours can be used to compute cumulative cost.
- s, e := thisPod.Start, thisPod.End
- if thisPVC.Start.After(thisPod.Start) {
- s = thisPVC.Start
- }
- if thisPVC.End.Before(thisPod.End) {
- e = thisPVC.End
- }
- thisPVCKey := thisPVC.key()
- if pvcPodWindowMap[thisPVCKey] == nil {
- pvcPodWindowMap[thisPVCKey] = make(map[podKey]opencost.Window)
- }
- pvcPodWindowMap[thisPVCKey][thisPodKey] = opencost.NewWindow(&s, &e)
- }
- }
- }
- for thisPVCKey, podWindowMap := range pvcPodWindowMap {
- // Build out a pv price coefficient for each pod with a pvc. Each
- // pvc-pod relation needs a coefficient which modifies the pv cost
- // such that pv costs can be shared between all pods using that pvc.
- // Get single-point intervals from alloc-pvc relation windows.
- intervals := getIntervalPointsFromWindows(podWindowMap)
- pvc, ok := pvcMap[thisPVCKey]
- if !ok {
- log.Warnf("Allocation: Compute: applyPVCsToPods: missing pvc with key %s", thisPVCKey)
- continue
- }
- if pvc == nil {
- log.Warnf("Allocation: Compute: applyPVCsToPods: nil pvc with key %s", thisPVCKey)
- continue
- }
- // Determine coefficients for each pvc-pod relation.
- sharedPVCCostCoefficients, err := getPVCCostCoefficients(intervals, pvc)
- if err != nil {
- log.Warnf("Allocation: Compute: applyPVCsToPods: getPVCCostCoefficients: %s", err)
- continue
- }
- // Distribute pvc costs to Allocations
- for thisPodKey, coeffComponents := range sharedPVCCostCoefficients {
- pod, ok2 := podMap[thisPodKey]
- // If pod does not exist or the pod does not have any allocations
- // get unmounted pod for cluster
- if !ok2 || len(pod.Allocations) == 0 {
- // Get namespace unmounted pod, as pvc will have a namespace
- pod = getUnmountedPodForNamespace(window, podMap, pvc.Cluster, pvc.Namespace)
- }
- for _, alloc := range pod.Allocations {
- s, e := pod.Start, pod.End
- minutes := e.Sub(s).Minutes()
- hrs := minutes / 60.0
- gib := pvc.Bytes / 1024 / 1024 / 1024
- cost := pvc.Volume.CostPerGiBHour * gib * hrs
- byteHours := pvc.Bytes * hrs
- coef := getCoefficientFromComponents(coeffComponents)
- // Apply the size and cost of the pv to the allocation, each
- // weighted by count (i.e. the number of containers in the pod)
- // record the amount of total PVBytes Hours attributable to a given pv
- if alloc.PVs == nil {
- alloc.PVs = opencost.PVAllocations{}
- }
- pvKey := opencost.PVKey{
- Cluster: pvc.Volume.Cluster,
- Name: pvc.Volume.Name,
- }
- // Both Cost and byteHours should be multiplied by the coef and divided by count
- // so that if all allocations with a given pv key are summed the result of those
- // would be equal to the values of the original pv
- count := float64(len(pod.Allocations))
- alloc.PVs[pvKey] = &opencost.PVAllocation{
- ByteHours: byteHours * coef / count,
- Cost: cost * coef / count,
- ProviderID: pvc.Volume.ProviderID,
- }
- }
- }
- }
- }
- func applyUnmountedPVs(window opencost.Window, podMap map[podKey]*pod, pvMap map[pvKey]*pv, pvcMap map[pvcKey]*pvc) {
- for _, pv := range pvMap {
- mounted := false
- for _, pvc := range pvcMap {
- if pvc.Volume == nil {
- continue
- }
- if pvc.Volume == pv {
- mounted = true
- break
- }
- }
- if !mounted {
- // a pv without a pvc will not have a namespace, so get the cluster unmounted pod
- pod := getUnmountedPodForCluster(window, podMap, pv.Cluster)
- // Calculate pv Cost
- // Unmounted pv should have correct keyso it can still reconcile
- thisPVKey := opencost.PVKey{
- Cluster: pv.Cluster,
- Name: pv.Name,
- }
- gib := pv.Bytes / 1024 / 1024 / 1024
- hrs := pv.minutes() / 60.0
- cost := pv.CostPerGiBHour * gib * hrs
- unmountedPVs := opencost.PVAllocations{
- thisPVKey: {
- ByteHours: pv.Bytes * hrs,
- Cost: cost,
- },
- }
- pod.Allocations[opencost.UnmountedSuffix].PVs = pod.Allocations[opencost.UnmountedSuffix].PVs.Add(unmountedPVs)
- }
- }
- }
- func applyUnmountedPVCs(window opencost.Window, podMap map[podKey]*pod, pvcMap map[pvcKey]*pvc) {
- for _, pvc := range pvcMap {
- if !pvc.Mounted && pvc.Volume != nil {
- // Get namespace unmounted pod, as pvc will have a namespace
- pod := getUnmountedPodForNamespace(window, podMap, pvc.Cluster, pvc.Namespace)
- // Calculate pv Cost
- // Unmounted pv should have correct key so it can still reconcile
- thisPVKey := opencost.PVKey{
- Cluster: pvc.Volume.Cluster,
- Name: pvc.Volume.Name,
- }
- // Use the Volume Bytes here because pvc bytes could be different,
- // however the pv bytes are what are going to determine cost
- gib := pvc.Volume.Bytes / 1024 / 1024 / 1024
- hrs := pvc.Volume.minutes() / 60.0
- cost := pvc.Volume.CostPerGiBHour * gib * hrs
- unmountedPVs := opencost.PVAllocations{
- thisPVKey: {
- ByteHours: pvc.Volume.Bytes * hrs,
- Cost: cost,
- },
- }
- pod.Allocations[opencost.UnmountedSuffix].PVs = pod.Allocations[opencost.UnmountedSuffix].PVs.Add(unmountedPVs)
- }
- }
- }
- /* Helper Helpers */
- // getUnmountedPodForCluster retrieve the unmounted pod for a cluster and create it if it does not exist
- func getUnmountedPodForCluster(window opencost.Window, podMap map[podKey]*pod, cluster string) *pod {
- container := opencost.UnmountedSuffix
- podName := opencost.UnmountedSuffix
- namespace := opencost.UnmountedSuffix
- node := ""
- thisPodKey := getUnmountedPodKey(cluster)
- // Initialize pod and container if they do not already exist
- thisPod, ok := podMap[thisPodKey]
- if !ok {
- thisPod = &pod{
- Window: window.Clone(),
- Start: *window.Start(),
- End: *window.End(),
- Key: thisPodKey,
- Allocations: map[string]*opencost.Allocation{},
- }
- thisPod.appendContainer(container)
- thisPod.Allocations[container].Properties.Cluster = cluster
- thisPod.Allocations[container].Properties.Node = node
- thisPod.Allocations[container].Properties.Namespace = namespace
- thisPod.Allocations[container].Properties.Pod = podName
- thisPod.Allocations[container].Properties.Container = container
- thisPod.Node = node
- podMap[thisPodKey] = thisPod
- }
- return thisPod
- }
- // getUnmountedPodForNamespace is as getUnmountedPodForCluster, but keys allocation property pod/namespace field off namespace
- // This creates or adds allocations to an unmounted pod in the specified namespace, rather than in __unmounted__
- func getUnmountedPodForNamespace(window opencost.Window, podMap map[podKey]*pod, cluster string, namespace string) *pod {
- container := opencost.UnmountedSuffix
- podName := fmt.Sprintf("%s-unmounted-pvcs", namespace)
- node := ""
- thisPodKey := newPodKey(cluster, namespace, podName)
- // Initialize pod and container if they do not already exist
- thisPod, ok := podMap[thisPodKey]
- if !ok {
- thisPod = &pod{
- Window: window.Clone(),
- Start: *window.Start(),
- End: *window.End(),
- Key: thisPodKey,
- Allocations: map[string]*opencost.Allocation{},
- }
- thisPod.appendContainer(container)
- thisPod.Allocations[container].Properties.Cluster = cluster
- thisPod.Allocations[container].Properties.Node = node
- thisPod.Allocations[container].Properties.Namespace = namespace
- thisPod.Allocations[container].Properties.Pod = podName
- thisPod.Allocations[container].Properties.Container = container
- thisPod.Node = node
- podMap[thisPodKey] = thisPod
- }
- return thisPod
- }
- func calculateStartAndEnd(result *prom.QueryResult, resolution time.Duration, window opencost.Window) (time.Time, time.Time) {
- // Start and end for a range vector are pulled from the timestamps of the
- // first and final values in the range.
- s := time.Unix(int64(result.Values[0].Timestamp), 0).UTC()
- e := time.Unix(int64(result.Values[len(result.Values)-1].Timestamp), 0).UTC()
- // As of Prometheus v3, we have had to reintroduce the "offsetting" of the
- // start time.
- //
- // E.g. avg(node_total_hourly_cost{}) by (node, provider_id)[1h:5m] with
- // time=01:00:00 will return, for a node running the entire time, 12
- // timestamps where the first is 00:05:00 and the last is 01:00:00.
- if IsPrometheusVersionGTE3() {
- s = s.Add(-resolution)
- }
- // The only corner-case here is what to do if you only get one timestamp.
- // This dilemma still requires the use of the resolution, and can be
- // clamped using the window. In this case, we want to honor the existence
- // of the pod by giving "one resolution" worth of duration, half on each
- // side of the given timestamp.
- if s.Equal(e) {
- s = s.Add(-1 * resolution / time.Duration(2))
- e = e.Add(resolution / time.Duration(2))
- }
- if s.Before(*window.Start()) {
- s = *window.Start()
- }
- if e.After(*window.End()) {
- e = *window.End()
- }
- return s, e
- }
- func getSanitizedDeviceName(deviceName string) string {
- if strings.Contains(deviceName, "nvidia") {
- return "nvidia"
- }
- return deviceName
- }
|