metricsquerier.go 137 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884188518861887188818891890189118921893189418951896189718981899190019011902190319041905190619071908190919101911191219131914191519161917191819191920192119221923192419251926192719281929193019311932193319341935193619371938193919401941194219431944194519461947194819491950195119521953195419551956195719581959196019611962196319641965196619671968196919701971197219731974197519761977197819791980198119821983198419851986198719881989199019911992199319941995199619971998199920002001200220032004200520062007200820092010201120122013201420152016201720182019202020212022202320242025202620272028202920302031203220332034203520362037203820392040204120422043204420452046204720482049205020512052205320542055205620572058205920602061206220632064206520662067206820692070207120722073207420752076207720782079208020812082208320842085208620872088208920902091209220932094209520962097209820992100210121022103210421052106210721082109211021112112211321142115211621172118211921202121212221232124212521262127212821292130213121322133213421352136213721382139214021412142214321442145214621472148214921502151215221532154215521562157215821592160216121622163216421652166216721682169217021712172217321742175217621772178217921802181218221832184218521862187218821892190219121922193219421952196219721982199220022012202220322042205220622072208220922102211221222132214221522162217221822192220222122222223222422252226222722282229223022312232223322342235223622372238223922402241224222432244224522462247224822492250225122522253225422552256225722582259226022612262226322642265226622672268226922702271227222732274227522762277227822792280228122822283228422852286228722882289229022912292229322942295229622972298229923002301230223032304230523062307230823092310231123122313231423152316231723182319232023212322232323242325232623272328232923302331233223332334233523362337233823392340234123422343234423452346234723482349235023512352235323542355235623572358235923602361236223632364236523662367236823692370237123722373237423752376237723782379238023812382238323842385238623872388238923902391239223932394239523962397239823992400240124022403240424052406240724082409241024112412241324142415241624172418241924202421242224232424242524262427242824292430243124322433243424352436243724382439244024412442244324442445244624472448244924502451245224532454245524562457245824592460246124622463246424652466246724682469247024712472247324742475247624772478247924802481248224832484248524862487248824892490249124922493249424952496249724982499250025012502250325042505250625072508250925102511251225132514251525162517251825192520252125222523252425252526252725282529253025312532253325342535253625372538253925402541254225432544254525462547254825492550255125522553255425552556255725582559256025612562256325642565256625672568256925702571257225732574257525762577257825792580258125822583258425852586258725882589259025912592259325942595259625972598259926002601260226032604260526062607260826092610261126122613261426152616261726182619262026212622262326242625262626272628262926302631263226332634263526362637263826392640264126422643264426452646264726482649265026512652265326542655265626572658265926602661266226632664266526662667266826692670267126722673267426752676267726782679268026812682268326842685268626872688268926902691269226932694269526962697269826992700270127022703270427052706270727082709271027112712271327142715271627172718271927202721272227232724272527262727272827292730273127322733273427352736273727382739274027412742274327442745274627472748274927502751275227532754275527562757275827592760276127622763276427652766276727682769277027712772277327742775277627772778277927802781278227832784278527862787278827892790279127922793279427952796279727982799280028012802280328042805280628072808280928102811281228132814281528162817281828192820282128222823282428252826282728282829283028312832283328342835283628372838283928402841284228432844284528462847284828492850285128522853285428552856285728582859286028612862286328642865286628672868286928702871287228732874287528762877287828792880288128822883288428852886288728882889
  1. package prom
  2. import (
  3. "fmt"
  4. "time"
  5. "github.com/opencost/opencost/core/pkg/log"
  6. "github.com/opencost/opencost/core/pkg/source"
  7. "github.com/opencost/opencost/core/pkg/util/timeutil"
  8. prometheus "github.com/prometheus/client_golang/api"
  9. )
  10. //--------------------------------------------------------------------------
  11. // PrometheusMetricsQuerier
  12. //--------------------------------------------------------------------------
  13. // PrometheusMetricsQueryLogFormat is the log format used to log metric queries before being sent to the prometheus
  14. // instance
  15. const PrometheusMetricsQueryLogFormat = `[PrometheusMetricsQuerier][%s][At Time: %d]: %s`
  16. // PrometheusMetricsQuerier is the implementation of the data source's MetricsQuerier interface for Prometheus.
  17. type PrometheusMetricsQuerier struct {
  18. promConfig *OpenCostPrometheusConfig
  19. promClient prometheus.Client
  20. promContexts *ContextFactory
  21. }
  22. func newPrometheusMetricsQuerier(
  23. promConfig *OpenCostPrometheusConfig,
  24. promClient prometheus.Client,
  25. promContexts *ContextFactory,
  26. ) *PrometheusMetricsQuerier {
  27. return &PrometheusMetricsQuerier{
  28. promConfig: promConfig,
  29. promClient: promClient,
  30. promContexts: promContexts,
  31. }
  32. }
  33. func (pds *PrometheusMetricsQuerier) QueryPVPricePerGiBHour(start, end time.Time) *source.Future[source.PVPricePerGiBHourResult] {
  34. const queryName = "QueryPVPricePerGiBHour"
  35. const pvCostQuery = `avg(avg_over_time(pv_hourly_cost{%s}[%s])) by (%s, persistentvolume, volumename, uid, provider_id)`
  36. cfg := pds.promConfig
  37. durStr := timeutil.DurationString(end.Sub(start))
  38. if durStr == "" {
  39. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  40. }
  41. queryPVCost := fmt.Sprintf(pvCostQuery, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  42. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryPVCost)
  43. ctx := pds.promContexts.NewNamedContext(ClusterContextName)
  44. return source.NewFuture(source.DecodePVPricePerGiBHourResult, ctx.QueryAtTime(queryPVCost, end))
  45. }
  46. func (pds *PrometheusMetricsQuerier) QueryPVUsedAverage(start, end time.Time) *source.Future[source.PVUsedAvgResult] {
  47. const queryName = "QueryPVUsedAverage"
  48. const pvUsedAverageQuery = `avg(avg_over_time(kubelet_volume_stats_used_bytes{%s}[%s])) by (%s, persistentvolumeclaim, namespace, uid)`
  49. cfg := pds.promConfig
  50. durStr := timeutil.DurationString(end.Sub(start))
  51. if durStr == "" {
  52. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  53. }
  54. queryPVUsedAvg := fmt.Sprintf(pvUsedAverageQuery, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  55. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryPVUsedAvg)
  56. ctx := pds.promContexts.NewNamedContext(ClusterContextName)
  57. return source.NewFuture(source.DecodePVUsedAvgResult, ctx.QueryAtTime(queryPVUsedAvg, end))
  58. }
  59. func (pds *PrometheusMetricsQuerier) QueryPVUsedMax(start, end time.Time) *source.Future[source.PVUsedMaxResult] {
  60. const queryName = "QueryPVUsedMax"
  61. const pvUsedMaxQuery = `max(max_over_time(kubelet_volume_stats_used_bytes{%s}[%s])) by (%s, persistentvolumeclaim, namespace, uid)`
  62. cfg := pds.promConfig
  63. durStr := timeutil.DurationString(end.Sub(start))
  64. if durStr == "" {
  65. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  66. }
  67. queryPVUsedMax := fmt.Sprintf(pvUsedMaxQuery, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  68. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryPVUsedMax)
  69. ctx := pds.promContexts.NewNamedContext(ClusterContextName)
  70. return source.NewFuture(source.DecodePVUsedMaxResult, ctx.QueryAtTime(queryPVUsedMax, end))
  71. }
  72. func (pds *PrometheusMetricsQuerier) QueryPVCUptime(start, end time.Time) *source.Future[source.UptimeResult] {
  73. const queryName = "QueryPVCUptime"
  74. const queryFmtPVCUptime = `avg(kube_persistentvolumeclaim_info{%s}) by (%s, uid)[%s:%dm]`
  75. cfg := pds.promConfig
  76. minsPerResolution := cfg.DataResolutionMinutes
  77. durStr := pds.durationStringFor(start, end, minsPerResolution, false)
  78. if durStr == "" {
  79. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  80. }
  81. queryPVCUptime := fmt.Sprintf(queryFmtPVCUptime, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution)
  82. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryPVCUptime)
  83. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  84. return source.NewFuture(source.DecodeUptimeResult, ctx.QueryAtTime(queryPVCUptime, end))
  85. }
  86. func (pds *PrometheusMetricsQuerier) QueryPVCInfo(start, end time.Time) *source.Future[source.PVCInfoResult] {
  87. const queryName = "QueryPVCInfo"
  88. const queryFmtPVCInfo = `avg(kube_persistentvolumeclaim_info{volumename != "", %s}) by (persistentvolumeclaim, storageclass, volumename, namespace, uid, %s)[%s:%dm]`
  89. cfg := pds.promConfig
  90. minsPerResolution := cfg.DataResolutionMinutes
  91. durStr := pds.durationStringFor(start, end, minsPerResolution, false)
  92. if durStr == "" {
  93. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  94. }
  95. queryPVCInfo := fmt.Sprintf(queryFmtPVCInfo, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution)
  96. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryPVCInfo)
  97. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  98. return source.NewFuture(source.DecodePVCInfoResult, ctx.QueryAtTime(queryPVCInfo, end))
  99. }
  100. func (pds *PrometheusMetricsQuerier) QueryPVActiveMinutes(start, end time.Time) *source.Future[source.PVActiveMinutesResult] {
  101. const queryName = "QueryPVActiveMinutes"
  102. const pvActiveMinsQuery = `avg(kube_persistentvolume_capacity_bytes{%s}) by (%s, persistentvolume, uid)[%s:%dm]`
  103. cfg := pds.promConfig
  104. minsPerResolution := cfg.DataResolutionMinutes
  105. durStr := pds.durationStringFor(start, end, minsPerResolution, false)
  106. if durStr == "" {
  107. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  108. }
  109. queryPVActiveMins := fmt.Sprintf(pvActiveMinsQuery, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution)
  110. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryPVActiveMins)
  111. ctx := pds.promContexts.NewNamedContext(ClusterContextName)
  112. return source.NewFuture(source.DecodePVActiveMinutesResult, ctx.QueryAtTime(queryPVActiveMins, end))
  113. }
  114. func (pds *PrometheusMetricsQuerier) QueryLocalStorageCost(start, end time.Time) *source.Future[source.LocalStorageCostResult] {
  115. const queryName = "QueryLocalStorageCost"
  116. const localStorageCostQuery = `sum_over_time(sum(container_fs_limit_bytes{device=~"/dev/(nvme|sda).*", id="/", %s}) by (instance, device, %s)[%s:%dm]) / 1024 / 1024 / 1024 * %f * %f`
  117. cfg := pds.promConfig
  118. minsPerResolution := cfg.DataResolutionMinutes
  119. durStr := pds.durationStringFor(start, end, minsPerResolution, false)
  120. if durStr == "" {
  121. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  122. }
  123. // hourlyToCumulative is a scaling factor that, when multiplied by an
  124. // hourly value, converts it to a cumulative value; i.e. [$/hr] *
  125. // [min/res]*[hr/min] = [$/res]
  126. hourlyToCumulative := float64(minsPerResolution) * (1.0 / 60.0)
  127. costPerGBHr := 0.04 / 730.0
  128. queryLocalStorageCost := fmt.Sprintf(localStorageCostQuery, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution, hourlyToCumulative, costPerGBHr)
  129. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryLocalStorageCost)
  130. ctx := pds.promContexts.NewNamedContext(ClusterContextName)
  131. return source.NewFuture(source.DecodeLocalStorageCostResult, ctx.QueryAtTime(queryLocalStorageCost, end))
  132. }
  133. func (pds *PrometheusMetricsQuerier) QueryLocalStorageUsedCost(start, end time.Time) *source.Future[source.LocalStorageUsedCostResult] {
  134. const queryName = "QueryLocalStorageUsedCost"
  135. const localStorageUsedCostQuery = `sum_over_time(sum(container_fs_usage_bytes{device=~"/dev/(nvme|sda).*", id="/", %s}) by (instance, device, %s)[%s:%dm]) / 1024 / 1024 / 1024 * %f * %f`
  136. cfg := pds.promConfig
  137. minsPerResolution := cfg.DataResolutionMinutes
  138. durStr := pds.durationStringFor(start, end, minsPerResolution, false)
  139. if durStr == "" {
  140. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  141. }
  142. // hourlyToCumulative is a scaling factor that, when multiplied by an
  143. // hourly value, converts it to a cumulative value; i.e. [$/hr] *
  144. // [min/res]*[hr/min] = [$/res]
  145. hourlyToCumulative := float64(minsPerResolution) * (1.0 / 60.0)
  146. costPerGBHr := 0.04 / 730.0
  147. queryLocalStorageUsedCost := fmt.Sprintf(localStorageUsedCostQuery, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution, hourlyToCumulative, costPerGBHr)
  148. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryLocalStorageUsedCost)
  149. ctx := pds.promContexts.NewNamedContext(ClusterContextName)
  150. return source.NewFuture(source.DecodeLocalStorageUsedCostResult, ctx.QueryAtTime(queryLocalStorageUsedCost, end))
  151. }
  152. func (pds *PrometheusMetricsQuerier) QueryLocalStorageUsedAvg(start, end time.Time) *source.Future[source.LocalStorageUsedAvgResult] {
  153. const queryName = "QueryLocalStorageUsedAvg"
  154. const localStorageUsedAvgQuery = `avg(sum(avg_over_time(container_fs_usage_bytes{device=~"/dev/(nvme|sda).*", id="/", %s}[%s])) by (instance, device, %s, job)) by (instance, device, %s)`
  155. cfg := pds.promConfig
  156. durStr := timeutil.DurationString(end.Sub(start))
  157. if durStr == "" {
  158. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  159. }
  160. queryLocalStorageUsedAvg := fmt.Sprintf(localStorageUsedAvgQuery, cfg.ClusterFilter, durStr, cfg.ClusterLabel, cfg.ClusterLabel)
  161. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryLocalStorageUsedAvg)
  162. ctx := pds.promContexts.NewNamedContext(ClusterContextName)
  163. return source.NewFuture(source.DecodeLocalStorageUsedAvgResult, ctx.QueryAtTime(queryLocalStorageUsedAvg, end))
  164. }
  165. func (pds *PrometheusMetricsQuerier) QueryLocalStorageUsedMax(start, end time.Time) *source.Future[source.LocalStorageUsedMaxResult] {
  166. const queryName = "QueryLocalStorageUsedMax"
  167. const localStorageUsedMaxQuery = `max(sum(max_over_time(container_fs_usage_bytes{device=~"/dev/(nvme|sda).*", id="/", %s}[%s])) by (instance, device, %s, job)) by (instance, device, %s)`
  168. cfg := pds.promConfig
  169. durStr := timeutil.DurationString(end.Sub(start))
  170. if durStr == "" {
  171. panic("failed to parse duration string passed to QueryLocalStorageUsedMax")
  172. }
  173. queryLocalStorageUsedMax := fmt.Sprintf(localStorageUsedMaxQuery, cfg.ClusterFilter, durStr, cfg.ClusterLabel, cfg.ClusterLabel)
  174. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryLocalStorageUsedMax)
  175. ctx := pds.promContexts.NewNamedContext(ClusterContextName)
  176. return source.NewFuture(source.DecodeLocalStorageUsedMaxResult, ctx.QueryAtTime(queryLocalStorageUsedMax, end))
  177. }
  178. func (pds *PrometheusMetricsQuerier) QueryLocalStorageBytes(start, end time.Time) *source.Future[source.LocalStorageBytesResult] {
  179. const queryName = "QueryLocalStorageBytes"
  180. const localStorageBytesQuery = `avg_over_time(sum(container_fs_limit_bytes{device=~"/dev/(nvme|sda).*", id="/", %s}) by (instance, device, %s)[%s:%dm])`
  181. cfg := pds.promConfig
  182. minsPerResolution := cfg.DataResolutionMinutes
  183. durStr := pds.durationStringFor(start, end, minsPerResolution, false)
  184. if durStr == "" {
  185. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  186. }
  187. queryLocalStorageBytes := fmt.Sprintf(localStorageBytesQuery, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution)
  188. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryLocalStorageBytes)
  189. ctx := pds.promContexts.NewNamedContext(ClusterContextName)
  190. return source.NewFuture(source.DecodeLocalStorageBytesResult, ctx.QueryAtTime(queryLocalStorageBytes, end))
  191. }
  192. func (pds *PrometheusMetricsQuerier) QueryLocalStorageActiveMinutes(start, end time.Time) *source.Future[source.LocalStorageActiveMinutesResult] {
  193. const queryName = "QueryLocalStorageActiveMinutes"
  194. const localStorageActiveMinutesQuery = `count(node_total_hourly_cost{%s}) by (%s, node, uid, instance, provider_id)[%s:%dm]`
  195. cfg := pds.promConfig
  196. minsPerResolution := cfg.DataResolutionMinutes
  197. durStr := pds.durationStringFor(start, end, minsPerResolution, false)
  198. if durStr == "" {
  199. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  200. }
  201. queryLocalStorageActiveMins := fmt.Sprintf(localStorageActiveMinutesQuery, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution)
  202. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryLocalStorageActiveMins)
  203. ctx := pds.promContexts.NewNamedContext(ClusterContextName)
  204. return source.NewFuture(source.DecodeLocalStorageActiveMinutesResult, ctx.QueryAtTime(queryLocalStorageActiveMins, end))
  205. }
  206. func (pds *PrometheusMetricsQuerier) QueryNodeInfo(start, end time.Time) *source.Future[source.NodeInfoResult] {
  207. const queryName = "QueryNodeInfo"
  208. const queryFmtNodeInfo = `avg(avg_over_time(node_info{%s}[%s])) by (%s, node, uid, provider_id, instance_type)`
  209. cfg := pds.promConfig
  210. durStr := timeutil.DurationString(end.Sub(start))
  211. if durStr == "" {
  212. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  213. }
  214. queryNodeInfo := fmt.Sprintf(queryFmtNodeInfo, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  215. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNodeInfo)
  216. ctx := pds.promContexts.NewNamedContext(ClusterContextName)
  217. return source.NewFuture(source.DecodeNodeInfoResult, ctx.QueryAtTime(queryNodeInfo, end))
  218. }
  219. func (pds *PrometheusMetricsQuerier) QueryNodeUptime(start, end time.Time) *source.Future[source.UptimeResult] {
  220. const queryName = "QueryNodeUptime"
  221. const queryFmtNodeUptime = `avg(node_info{%s}) by (%s, uid)[%s:%dm]`
  222. cfg := pds.promConfig
  223. minsPerResolution := cfg.DataResolutionMinutes
  224. durStr := pds.durationStringFor(start, end, minsPerResolution, false)
  225. if durStr == "" {
  226. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  227. }
  228. queryNodeUptime := fmt.Sprintf(queryFmtNodeUptime, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution)
  229. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNodeUptime)
  230. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  231. return source.NewFuture(source.DecodeUptimeResult, ctx.QueryAtTime(queryNodeUptime, end))
  232. }
  233. func (pds *PrometheusMetricsQuerier) QueryNodeCPUCoresCapacity(start, end time.Time) *source.Future[source.NodeCPUCoresCapacityResult] {
  234. const queryName = "QueryNodeCPUCoresCapacity"
  235. const nodeCPUCoresCapacityQuery = `avg(avg_over_time(kube_node_status_capacity_cpu_cores{%s}[%s])) by (%s, node, uid)`
  236. cfg := pds.promConfig
  237. durStr := timeutil.DurationString(end.Sub(start))
  238. if durStr == "" {
  239. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  240. }
  241. queryNodeCPUCoresCapacity := fmt.Sprintf(nodeCPUCoresCapacityQuery, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  242. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNodeCPUCoresCapacity)
  243. ctx := pds.promContexts.NewNamedContext(ClusterContextName)
  244. return source.NewFuture(source.DecodeNodeCPUCoresCapacityResult, ctx.QueryAtTime(queryNodeCPUCoresCapacity, end))
  245. }
  246. func (pds *PrometheusMetricsQuerier) QueryNodeCPUCoresAllocatable(start, end time.Time) *source.Future[source.NodeCPUCoresAllocatableResult] {
  247. const queryName = "QueryNodeCPUCoresAllocatable"
  248. const nodeCPUCoresAllocatableQuery = `avg(avg_over_time(kube_node_status_allocatable_cpu_cores{%s}[%s])) by (%s, node, uid)`
  249. // `avg(avg_over_time(container_cpu_allocation{container!="", container!="POD", node!="", %s}[%s])) by (container, pod, namespace, node, %s)`
  250. cfg := pds.promConfig
  251. durStr := timeutil.DurationString(end.Sub(start))
  252. if durStr == "" {
  253. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  254. }
  255. queryNodeCPUCoresAllocatable := fmt.Sprintf(nodeCPUCoresAllocatableQuery, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  256. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNodeCPUCoresAllocatable)
  257. ctx := pds.promContexts.NewNamedContext(ClusterContextName)
  258. return source.NewFuture(source.DecodeNodeCPUCoresAllocatableResult, ctx.QueryAtTime(queryNodeCPUCoresAllocatable, end))
  259. }
  260. func (pds *PrometheusMetricsQuerier) QueryNodeRAMBytesCapacity(start, end time.Time) *source.Future[source.NodeRAMBytesCapacityResult] {
  261. const queryName = "QueryNodeRAMBytesCapacity"
  262. const nodeRAMBytesCapacityQuery = `avg(avg_over_time(kube_node_status_capacity_memory_bytes{%s}[%s])) by (%s, node, uid)`
  263. cfg := pds.promConfig
  264. durStr := timeutil.DurationString(end.Sub(start))
  265. if durStr == "" {
  266. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  267. }
  268. queryNodeRAMBytesCapacity := fmt.Sprintf(nodeRAMBytesCapacityQuery, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  269. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNodeRAMBytesCapacity)
  270. ctx := pds.promContexts.NewNamedContext(ClusterContextName)
  271. return source.NewFuture(source.DecodeNodeRAMBytesCapacityResult, ctx.QueryAtTime(queryNodeRAMBytesCapacity, end))
  272. }
  273. func (pds *PrometheusMetricsQuerier) QueryNodeRAMBytesAllocatable(start, end time.Time) *source.Future[source.NodeRAMBytesAllocatableResult] {
  274. const queryName = "QueryNodeRAMBytesAllocatable"
  275. const nodeRAMBytesAllocatableQuery = `avg(avg_over_time(kube_node_status_allocatable_memory_bytes{%s}[%s])) by (%s, node, uid)`
  276. cfg := pds.promConfig
  277. durStr := timeutil.DurationString(end.Sub(start))
  278. if durStr == "" {
  279. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  280. }
  281. queryNodeRAMBytesAllocatable := fmt.Sprintf(nodeRAMBytesAllocatableQuery, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  282. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNodeRAMBytesAllocatable)
  283. ctx := pds.promContexts.NewNamedContext(ClusterContextName)
  284. return source.NewFuture(source.DecodeNodeRAMBytesAllocatableResult, ctx.QueryAtTime(queryNodeRAMBytesAllocatable, end))
  285. }
  286. func (pds *PrometheusMetricsQuerier) QueryNodeGPUCount(start, end time.Time) *source.Future[source.NodeGPUCountResult] {
  287. const queryName = "QueryNodeGPUCount"
  288. const nodeGPUCountQuery = `avg(avg_over_time(node_gpu_count{%s}[%s])) by (%s, node, uid, provider_id)`
  289. cfg := pds.promConfig
  290. durStr := timeutil.DurationString(end.Sub(start))
  291. if durStr == "" {
  292. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  293. }
  294. queryNodeGPUCount := fmt.Sprintf(nodeGPUCountQuery, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  295. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNodeGPUCount)
  296. ctx := pds.promContexts.NewNamedContext(ClusterContextName)
  297. return source.NewFuture(source.DecodeNodeGPUCountResult, ctx.QueryAtTime(queryNodeGPUCount, end))
  298. }
  299. func (pds *PrometheusMetricsQuerier) QueryNodeLabels(start, end time.Time) *source.Future[source.NodeLabelsResult] {
  300. const queryName = "QueryNodeLabels"
  301. const labelsQuery = `avg_over_time(kube_node_labels{%s}[%s])`
  302. cfg := pds.promConfig
  303. durStr := timeutil.DurationString(end.Sub(start))
  304. if durStr == "" {
  305. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  306. }
  307. queryLabels := fmt.Sprintf(labelsQuery, cfg.ClusterFilter, durStr)
  308. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryLabels)
  309. ctx := pds.promContexts.NewNamedContext(ClusterContextName)
  310. return source.NewFuture(source.DecodeNodeLabelsResult, ctx.QueryAtTime(queryLabels, end))
  311. }
  312. func (pds *PrometheusMetricsQuerier) QueryNodeActiveMinutes(start, end time.Time) *source.Future[source.NodeActiveMinutesResult] {
  313. const queryName = "QueryNodeActiveMinutes"
  314. const activeMinsQuery = `avg(node_total_hourly_cost{%s}) by (node, uid, %s, provider_id)[%s:%dm]`
  315. cfg := pds.promConfig
  316. minsPerResolution := cfg.DataResolutionMinutes
  317. durStr := pds.durationStringFor(start, end, minsPerResolution, false)
  318. if durStr == "" {
  319. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  320. }
  321. queryActiveMins := fmt.Sprintf(activeMinsQuery, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution)
  322. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryActiveMins)
  323. ctx := pds.promContexts.NewNamedContext(ClusterContextName)
  324. return source.NewFuture(source.DecodeNodeActiveMinutesResult, ctx.QueryAtTime(queryActiveMins, end))
  325. }
  326. func (pds *PrometheusMetricsQuerier) QueryNodeCPUModeTotal(start, end time.Time) *source.Future[source.NodeCPUModeTotalResult] {
  327. const queryName = "QueryNodeCPUModeTotal"
  328. const nodeCPUModeTotalQuery = `sum(rate(node_cpu_seconds_total{%s}[%s:%dm])) by (kubernetes_node, uid, %s, mode)`
  329. cfg := pds.promConfig
  330. minsPerResolution := cfg.DataResolutionMinutes
  331. durStr := pds.durationStringFor(start, end, minsPerResolution, true)
  332. if durStr == "" {
  333. panic("failed to parse duration string passed to QueryNodeCPUModeTotal")
  334. }
  335. queryCPUModeTotal := fmt.Sprintf(nodeCPUModeTotalQuery, cfg.ClusterFilter, durStr, minsPerResolution, cfg.ClusterLabel)
  336. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryCPUModeTotal)
  337. ctx := pds.promContexts.NewNamedContext(ClusterContextName)
  338. return source.NewFuture(source.DecodeNodeCPUModeTotalResult, ctx.QueryAtTime(queryCPUModeTotal, end))
  339. }
  340. func (pds *PrometheusMetricsQuerier) QueryNodeRAMSystemPercent(start, end time.Time) *source.Future[source.NodeRAMSystemPercentResult] {
  341. const queryName = "QueryNodeRAMSystemPercent"
  342. const nodeRAMSystemPctQuery = `sum(sum_over_time(container_memory_working_set_bytes{container_name!="POD",container_name!="",namespace="kube-system", %s}[%s:%dm])) by (instance, uid, %s) / avg(label_replace(sum(sum_over_time(kube_node_status_capacity_memory_bytes{%s}[%s:%dm])) by (node, uid, %s), "instance", "$1", "node", "(.*)")) by (instance, uid, %s)`
  343. cfg := pds.promConfig
  344. minsPerResolution := cfg.DataResolutionMinutes
  345. durStr := pds.durationStringFor(start, end, minsPerResolution, false)
  346. if durStr == "" {
  347. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  348. }
  349. queryRAMSystemPct := fmt.Sprintf(nodeRAMSystemPctQuery, cfg.ClusterFilter, durStr, minsPerResolution, cfg.ClusterLabel, cfg.ClusterFilter, durStr, minsPerResolution, cfg.ClusterLabel, cfg.ClusterLabel)
  350. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryRAMSystemPct)
  351. ctx := pds.promContexts.NewNamedContext(ClusterContextName)
  352. return source.NewFuture(source.DecodeNodeRAMSystemPercentResult, ctx.QueryAtTime(queryRAMSystemPct, end))
  353. }
  354. func (pds *PrometheusMetricsQuerier) QueryNodeRAMUserPercent(start, end time.Time) *source.Future[source.NodeRAMUserPercentResult] {
  355. const queryName = "QueryNodeRAMUserPercent"
  356. const nodeRAMUserPctQuery = `sum(sum_over_time(container_memory_working_set_bytes{container_name!="POD",container_name!="",namespace!="kube-system", %s}[%s:%dm])) by (instance, uid, %s) / avg(label_replace(sum(sum_over_time(kube_node_status_capacity_memory_bytes{%s}[%s:%dm])) by (node, uid, %s), "instance", "$1", "node", "(.*)")) by (instance, uid, %s)`
  357. cfg := pds.promConfig
  358. minsPerResolution := cfg.DataResolutionMinutes
  359. durStr := pds.durationStringFor(start, end, minsPerResolution, false)
  360. if durStr == "" {
  361. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  362. }
  363. queryRAMUserPct := fmt.Sprintf(nodeRAMUserPctQuery, cfg.ClusterFilter, durStr, minsPerResolution, cfg.ClusterLabel, cfg.ClusterFilter, durStr, minsPerResolution, cfg.ClusterLabel, cfg.ClusterLabel)
  364. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryRAMUserPct)
  365. ctx := pds.promContexts.NewNamedContext(ClusterContextName)
  366. return source.NewFuture(source.DecodeNodeRAMUserPercentResult, ctx.QueryAtTime(queryRAMUserPct, end))
  367. }
  368. func (pds *PrometheusMetricsQuerier) QueryNodeResourceCapacities(start, end time.Time) *source.Future[source.ResourceResult] {
  369. const queryName = "QueryNodeResourceCapacities"
  370. const queryFmtNodeResourceCapacities = `avg(avg_over_time(kube_node_status_capacity{%s}[%s])) by (%s, node, uid, resource, unit)`
  371. cfg := pds.promConfig
  372. durStr := timeutil.DurationString(end.Sub(start))
  373. if durStr == "" {
  374. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  375. }
  376. queryNodeResourceCapacities := fmt.Sprintf(queryFmtNodeResourceCapacities, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  377. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNodeResourceCapacities)
  378. ctx := pds.promContexts.NewNamedContext(ClusterContextName)
  379. return source.NewFuture(source.DecodeResourceResult, ctx.QueryAtTime(queryNodeResourceCapacities, end))
  380. }
  381. func (pds *PrometheusMetricsQuerier) QueryNodeResourcesAllocatable(start, end time.Time) *source.Future[source.ResourceResult] {
  382. const queryName = "QueryNodeResourcesAllocatable"
  383. const queryFmtNodeResourcesAllocatable = `avg(avg_over_time(kube_node_status_allocatable{%s}[%s])) by (%s, node, uid, resource, unit)`
  384. cfg := pds.promConfig
  385. durStr := timeutil.DurationString(end.Sub(start))
  386. if durStr == "" {
  387. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  388. }
  389. queryNodeResourcesAllocatable := fmt.Sprintf(queryFmtNodeResourcesAllocatable, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  390. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNodeResourcesAllocatable)
  391. ctx := pds.promContexts.NewNamedContext(ClusterContextName)
  392. return source.NewFuture(source.DecodeResourceResult, ctx.QueryAtTime(queryNodeResourcesAllocatable, end))
  393. }
  394. func (pds *PrometheusMetricsQuerier) QueryLBPricePerHr(start, end time.Time) *source.Future[source.LBPricePerHrResult] {
  395. const queryName = "QueryLBPricePerHr"
  396. const queryFmtLBCostPerHr = `avg(avg_over_time(kubecost_load_balancer_cost{%s}[%s])) by (namespace, service_name, ingress_ip, uid, %s)`
  397. cfg := pds.promConfig
  398. durStr := timeutil.DurationString(end.Sub(start))
  399. if durStr == "" {
  400. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  401. }
  402. queryLBCostPerHr := fmt.Sprintf(queryFmtLBCostPerHr, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  403. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryLBCostPerHr)
  404. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  405. return source.NewFuture(source.DecodeLBPricePerHrResult, ctx.QueryAtTime(queryLBCostPerHr, end))
  406. }
  407. func (pds *PrometheusMetricsQuerier) QueryLBActiveMinutes(start, end time.Time) *source.Future[source.LBActiveMinutesResult] {
  408. const queryName = "QueryLBActiveMinutes"
  409. const lbActiveMinutesQuery = `avg(kubecost_load_balancer_cost{%s}) by (namespace, service_name, uid, %s, ingress_ip)[%s:%dm]`
  410. cfg := pds.promConfig
  411. minsPerResolution := cfg.DataResolutionMinutes
  412. durStr := pds.durationStringFor(start, end, minsPerResolution, false)
  413. if durStr == "" {
  414. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  415. }
  416. queryLBActiveMins := fmt.Sprintf(lbActiveMinutesQuery, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution)
  417. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryLBActiveMins)
  418. ctx := pds.promContexts.NewNamedContext(ClusterContextName)
  419. return source.NewFuture(source.DecodeLBActiveMinutesResult, ctx.QueryAtTime(queryLBActiveMins, end))
  420. }
  421. func (pds *PrometheusMetricsQuerier) QueryClusterInfo(start, end time.Time) *source.Future[source.ClusterInfoResult] {
  422. const queryName = "QueryClusterInfo"
  423. const queryFmtClusterInfo = `avg(avg_over_time(cluster_info{%s}[%s])) by (%s, uid, provider, account_id, provisioner_name, region)`
  424. cfg := pds.promConfig
  425. durStr := timeutil.DurationString(end.Sub(start))
  426. if durStr == "" {
  427. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  428. }
  429. queryClusterInfo := fmt.Sprintf(queryFmtClusterInfo, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  430. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryClusterInfo)
  431. ctx := pds.promContexts.NewNamedContext(ClusterContextName)
  432. return source.NewFuture(source.DecodeClusterInfoResult, ctx.QueryAtTime(queryClusterInfo, end))
  433. }
  434. func (pds *PrometheusMetricsQuerier) QueryClusterUptime(start, end time.Time) *source.Future[source.UptimeResult] {
  435. const queryName = "QueryClusterUptime"
  436. const queryFmtClusterUptime = `avg(cluster_info{%s}) by (%s, uid)[%s:%dm]`
  437. cfg := pds.promConfig
  438. minsPerResolution := cfg.DataResolutionMinutes
  439. durStr := pds.durationStringFor(start, end, minsPerResolution, false)
  440. if durStr == "" {
  441. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  442. }
  443. queryClusterUptime := fmt.Sprintf(queryFmtClusterUptime, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution)
  444. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryFmtClusterUptime)
  445. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  446. return source.NewFuture(source.DecodeUptimeResult, ctx.QueryAtTime(queryClusterUptime, end))
  447. }
  448. func (pds *PrometheusMetricsQuerier) QueryClusterManagementDuration(start, end time.Time) *source.Future[source.ClusterManagementDurationResult] {
  449. const queryName = "QueryClusterManagementDuration"
  450. const clusterManagementDurationQuery = `avg(kubecost_cluster_management_cost{%s}) by (%s, provisioner_name)[%s:%dm]`
  451. cfg := pds.promConfig
  452. minsPerResolution := cfg.DataResolutionMinutes
  453. durStr := pds.durationStringFor(start, end, minsPerResolution, false)
  454. if durStr == "" {
  455. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  456. }
  457. queryClusterManagementDuration := fmt.Sprintf(clusterManagementDurationQuery, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution)
  458. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryClusterManagementDuration)
  459. ctx := pds.promContexts.NewNamedContext(ClusterContextName)
  460. return source.NewFuture(source.DecodeClusterManagementDurationResult, ctx.QueryAtTime(queryClusterManagementDuration, end))
  461. }
  462. func (pds *PrometheusMetricsQuerier) QueryClusterManagementPricePerHr(start, end time.Time) *source.Future[source.ClusterManagementPricePerHrResult] {
  463. const queryName = "QueryClusterManagementPricePerHr"
  464. const clusterManagementCostQuery = `avg(avg_over_time(kubecost_cluster_management_cost{%s}[%s])) by (%s, provisioner_name)`
  465. cfg := pds.promConfig
  466. durStr := timeutil.DurationString(end.Sub(start))
  467. if durStr == "" {
  468. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  469. }
  470. queryClusterManagementCost := fmt.Sprintf(clusterManagementCostQuery, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  471. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryClusterManagementCost)
  472. ctx := pds.promContexts.NewNamedContext(ClusterContextName)
  473. return source.NewFuture(source.DecodeClusterManagementPricePerHrResult, ctx.QueryAtTime(queryClusterManagementCost, end))
  474. }
  475. // AllocationMetricQuerier
  476. func (pds *PrometheusMetricsQuerier) QueryPods(start, end time.Time) *source.Future[source.PodsResult] {
  477. const queryName = "QueryPods"
  478. const queryFmtPods = `avg(kube_pod_container_status_running{%s} != 0) by (pod, namespace, uid, %s)[%s:%dm]`
  479. cfg := pds.promConfig
  480. minsPerResolution := cfg.DataResolutionMinutes
  481. durStr := pds.durationStringFor(start, end, minsPerResolution, false)
  482. if durStr == "" {
  483. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  484. }
  485. queryPods := fmt.Sprintf(queryFmtPods, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution)
  486. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryPods)
  487. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  488. return source.NewFuture(source.DecodePodsResult, ctx.QueryAtTime(queryPods, end))
  489. }
  490. func (pds *PrometheusMetricsQuerier) QueryPodsUID(start, end time.Time) *source.Future[source.PodsResult] {
  491. const queryName = "QueryPodsUID"
  492. const queryFmtPodsUID = `avg(kube_pod_container_status_running{%s} != 0) by (pod, namespace, uid, %s)[%s:%dm]`
  493. cfg := pds.promConfig
  494. minsPerResolution := cfg.DataResolutionMinutes
  495. durStr := pds.durationStringFor(start, end, minsPerResolution, false)
  496. if durStr == "" {
  497. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  498. }
  499. queryPodsUID := fmt.Sprintf(queryFmtPodsUID, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution)
  500. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryPodsUID)
  501. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  502. return source.NewFuture(source.DecodePodsResult, ctx.QueryAtTime(queryPodsUID, end))
  503. }
  504. func (pds *PrometheusMetricsQuerier) QueryPodInfo(start, end time.Time) *source.Future[source.PodInfoResult] {
  505. const queryName = "QueryPodInfo"
  506. const queryFmtPodInfo = `avg(avg_over_time(pod_info{%s}[%s])) by (%s, pod, uid, namespace_uid, node_uid)`
  507. cfg := pds.promConfig
  508. durStr := timeutil.DurationString(end.Sub(start))
  509. if durStr == "" {
  510. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  511. }
  512. queryPodInfo := fmt.Sprintf(queryFmtPodInfo, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  513. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryPodInfo)
  514. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  515. return source.NewFuture(source.DecodePodInfoResult, ctx.QueryAtTime(queryPodInfo, end))
  516. }
  517. func (pds *PrometheusMetricsQuerier) QueryPodUptime(start, end time.Time) *source.Future[source.UptimeResult] {
  518. const queryName = "QueryPodUptime"
  519. const queryFmtPodUptime = `avg(pod_info{%s}) by (%s, uid)[%s:%dm]`
  520. cfg := pds.promConfig
  521. minsPerResolution := cfg.DataResolutionMinutes
  522. durStr := pds.durationStringFor(start, end, minsPerResolution, false)
  523. if durStr == "" {
  524. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  525. }
  526. queryPodUptime := fmt.Sprintf(queryFmtPodUptime, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution)
  527. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryPodUptime)
  528. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  529. return source.NewFuture(source.DecodeUptimeResult, ctx.QueryAtTime(queryPodUptime, end))
  530. }
  531. func (pds *PrometheusMetricsQuerier) QueryPodOwners(start, end time.Time) *source.Future[source.OwnerResult] {
  532. const queryName = "QueryPodOwners"
  533. const queryFmtPodOwners = `avg(avg_over_time(kube_pod_owner{%s}[%s])) by (%s, uid, owner_uid, owner_kind)`
  534. cfg := pds.promConfig
  535. durStr := timeutil.DurationString(end.Sub(start))
  536. if durStr == "" {
  537. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  538. }
  539. queryPodOwners := fmt.Sprintf(queryFmtPodOwners, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  540. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryPodOwners)
  541. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  542. return source.NewFuture(source.DecodeOwnerResult, ctx.QueryAtTime(queryPodOwners, end))
  543. }
  544. func (pds *PrometheusMetricsQuerier) QueryPodPVCVolumes(start, end time.Time) *source.Future[source.PodPVCVolumeResult] {
  545. const queryName = "QueryPodPVCVolumes"
  546. const queryFmtPodPVCVolumes = `avg(avg_over_time(pod_pvc_volume{%s}[%s])) by (%s, uid, persistentvolumeclaim_uid, pod_volume_name)`
  547. cfg := pds.promConfig
  548. durStr := timeutil.DurationString(end.Sub(start))
  549. if durStr == "" {
  550. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  551. }
  552. queryPodPVCVolumes := fmt.Sprintf(queryFmtPodPVCVolumes, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  553. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryPodPVCVolumes)
  554. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  555. return source.NewFuture(source.DecodePodPVCVolumeResult, ctx.QueryAtTime(queryPodPVCVolumes, end))
  556. }
  557. func (pds *PrometheusMetricsQuerier) QueryPodNetworkEgressBytes(start, end time.Time) *source.Future[source.PodNetworkBytesResult] {
  558. const queryName = "QueryPodNetworkEgressBytes"
  559. const queryFmt = `sum(increase(kubecost_pod_network_egress_bytes_total{uid!="", %s}[%s:%dm])) by (uid, service, internet, same_region, same_zone, nat_gateway, %s)`
  560. cfg := pds.promConfig
  561. minsPerResolution := cfg.DataResolutionMinutes
  562. durStr := pds.durationStringFor(start, end, minsPerResolution, true)
  563. if durStr == "" {
  564. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  565. }
  566. q := fmt.Sprintf(queryFmt, cfg.ClusterFilter, durStr, minsPerResolution, cfg.ClusterLabel)
  567. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), q)
  568. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  569. return source.NewFuture(source.DecodePodNetworkBytesResult, ctx.QueryAtTime(q, end))
  570. }
  571. func (pds *PrometheusMetricsQuerier) QueryPodNetworkIngressBytes(start, end time.Time) *source.Future[source.PodNetworkBytesResult] {
  572. const queryName = "QueryPodNetworkIngressBytes"
  573. const queryFmt = `sum(increase(kubecost_pod_network_ingress_bytes_total{uid!="", %s}[%s:%dm])) by (uid, service, internet, same_region, same_zone, nat_gateway, %s)`
  574. cfg := pds.promConfig
  575. minsPerResolution := cfg.DataResolutionMinutes
  576. durStr := pds.durationStringFor(start, end, minsPerResolution, true)
  577. if durStr == "" {
  578. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  579. }
  580. q := fmt.Sprintf(queryFmt, cfg.ClusterFilter, durStr, minsPerResolution, cfg.ClusterLabel)
  581. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), q)
  582. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  583. return source.NewFuture(source.DecodePodNetworkBytesResult, ctx.QueryAtTime(q, end))
  584. }
  585. func (pds *PrometheusMetricsQuerier) QueryContainerUptime(start, end time.Time) *source.Future[source.ContainerUptimeResult] {
  586. const queryName = "QueryContainerUptime"
  587. const queryFmtContainerUptime = `avg(kube_pod_container_status_running{container!="", %s} != 0) by (container, uid, %s)[%s:%dm]`
  588. cfg := pds.promConfig
  589. minsPerResolution := cfg.DataResolutionMinutes
  590. durStr := pds.durationStringFor(start, end, minsPerResolution, false)
  591. if durStr == "" {
  592. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  593. }
  594. queryContainerUptime := fmt.Sprintf(queryFmtContainerUptime, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution)
  595. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryContainerUptime)
  596. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  597. return source.NewFuture(source.DecodeContainerUptimeResult, ctx.QueryAtTime(queryContainerUptime, end))
  598. }
  599. func (pds *PrometheusMetricsQuerier) QueryContainerResourceRequests(start, end time.Time) *source.Future[source.ContainerResourceResult] {
  600. const queryName = "QueryContainerResourceRequests"
  601. const queryFmtContainerResourceRequests = `avg(avg_over_time(kube_pod_container_resource_requests{container!="", container!="POD", node!="", %s}[%s])) by (container, uid, resource, unit, %s)`
  602. cfg := pds.promConfig
  603. durStr := timeutil.DurationString(end.Sub(start))
  604. if durStr == "" {
  605. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  606. }
  607. queryContainerResourceRequests := fmt.Sprintf(queryFmtContainerResourceRequests, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  608. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryContainerResourceRequests)
  609. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  610. return source.NewFuture(source.DecodeContainerResourceResult, ctx.QueryAtTime(queryContainerResourceRequests, end))
  611. }
  612. func (pds *PrometheusMetricsQuerier) QueryContainerResourceLimits(start, end time.Time) *source.Future[source.ContainerResourceResult] {
  613. const queryName = "QueryContainerResourceLimits"
  614. const queryFmtContainerResourceLimits = `avg(avg_over_time(kube_pod_container_resource_limits{container!="", container!="POD", node!="", %s}[%s])) by (container, uid, resource, unit, %s)`
  615. cfg := pds.promConfig
  616. durStr := timeutil.DurationString(end.Sub(start))
  617. if durStr == "" {
  618. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  619. }
  620. queryContainerResourceLimits := fmt.Sprintf(queryFmtContainerResourceLimits, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  621. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryContainerResourceLimits)
  622. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  623. return source.NewFuture(source.DecodeContainerResourceResult, ctx.QueryAtTime(queryContainerResourceLimits, end))
  624. }
  625. func (pds *PrometheusMetricsQuerier) QueryRAMBytesAllocated(start, end time.Time) *source.Future[source.RAMBytesAllocatedResult] {
  626. const queryName = "QueryRAMBytesAllocated"
  627. const queryFmtRAMBytesAllocated = `avg(avg_over_time(container_memory_allocation_bytes{container!="", container!="POD", node!="", %s}[%s])) by (container, pod, namespace, node, uid, %s, provider_id)`
  628. cfg := pds.promConfig
  629. durStr := timeutil.DurationString(end.Sub(start))
  630. if durStr == "" {
  631. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  632. }
  633. queryRAMBytesAllocated := fmt.Sprintf(queryFmtRAMBytesAllocated, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  634. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryRAMBytesAllocated)
  635. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  636. return source.NewFuture(source.DecodeRAMBytesAllocatedResult, ctx.QueryAtTime(queryRAMBytesAllocated, end))
  637. }
  638. func (pds *PrometheusMetricsQuerier) QueryRAMRequests(start, end time.Time) *source.Future[source.RAMRequestsResult] {
  639. const queryName = "QueryRAMRequests"
  640. const queryFmtRAMRequests = `avg(avg_over_time(kube_pod_container_resource_requests{resource="memory", unit="byte", container!="", container!="POD", node!="", %s}[%s])) by (container, pod, namespace, node, uid, %s)`
  641. cfg := pds.promConfig
  642. durStr := timeutil.DurationString(end.Sub(start))
  643. if durStr == "" {
  644. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  645. }
  646. queryRAMRequests := fmt.Sprintf(queryFmtRAMRequests, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  647. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryRAMRequests)
  648. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  649. return source.NewFuture(source.DecodeRAMRequestsResult, ctx.QueryAtTime(queryRAMRequests, end))
  650. }
  651. func (pds *PrometheusMetricsQuerier) QueryRAMLimits(start, end time.Time) *source.Future[source.RAMLimitsResult] {
  652. const queryName = "QueryRAMLimits"
  653. const queryFmtRAMLimits = `avg(avg_over_time(kube_pod_container_resource_limits{resource="memory", unit="byte", container!="", container!="POD", node!="", %s}[%s])) by (container, pod, namespace, node, %s)`
  654. cfg := pds.promConfig
  655. durStr := timeutil.DurationString(end.Sub(start))
  656. if durStr == "" {
  657. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  658. }
  659. queryRAMLimits := fmt.Sprintf(queryFmtRAMLimits, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  660. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryRAMLimits)
  661. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  662. return source.NewFuture(source.DecodeRAMLimitsResult, ctx.QueryAtTime(queryRAMLimits, end))
  663. }
  664. func (pds *PrometheusMetricsQuerier) QueryRAMUsageAvg(start, end time.Time) *source.Future[source.RAMUsageAvgResult] {
  665. const queryName = "QueryRAMUsageAvg"
  666. const queryFmtRAMUsageAvg = `avg(avg_over_time(container_memory_working_set_bytes{container!="", container_name!="POD", container!="POD", %s}[%s])) by (container_name, container, pod_name, pod, namespace, node, instance, uid, %s)`
  667. cfg := pds.promConfig
  668. durStr := timeutil.DurationString(end.Sub(start))
  669. if durStr == "" {
  670. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  671. }
  672. queryRAMUsageAvg := fmt.Sprintf(queryFmtRAMUsageAvg, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  673. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryRAMUsageAvg)
  674. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  675. return source.NewFuture(source.DecodeRAMUsageAvgResult, ctx.QueryAtTime(queryRAMUsageAvg, end))
  676. }
  677. func (pds *PrometheusMetricsQuerier) QueryRAMUsageMax(start, end time.Time) *source.Future[source.RAMUsageMaxResult] {
  678. const queryName = "QueryRAMUsageMax"
  679. const queryFmtRAMUsageMax = `max(max_over_time(container_memory_working_set_bytes{container!="", container_name!="POD", container!="POD", %s}[%s])) by (container_name, container, pod_name, pod, namespace, node, instance, uid, %s)`
  680. cfg := pds.promConfig
  681. durStr := timeutil.DurationString(end.Sub(start))
  682. if durStr == "" {
  683. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  684. }
  685. queryRAMUsageMax := fmt.Sprintf(queryFmtRAMUsageMax, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  686. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryRAMUsageMax)
  687. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  688. return source.NewFuture(source.DecodeRAMUsageMaxResult, ctx.QueryAtTime(queryRAMUsageMax, end))
  689. }
  690. func (pds *PrometheusMetricsQuerier) QueryCPUCoresAllocated(start, end time.Time) *source.Future[source.CPUCoresAllocatedResult] {
  691. const queryName = "QueryCPUCoresAllocated"
  692. const queryFmtCPUCoresAllocated = `avg(avg_over_time(container_cpu_allocation{container!="", container!="POD", node!="", %s}[%s])) by (container, pod, namespace, node, uid, %s)`
  693. cfg := pds.promConfig
  694. durStr := timeutil.DurationString(end.Sub(start))
  695. if durStr == "" {
  696. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  697. }
  698. queryCPUCoresAllocated := fmt.Sprintf(queryFmtCPUCoresAllocated, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  699. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryCPUCoresAllocated)
  700. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  701. return source.NewFuture(source.DecodeCPUCoresAllocatedResult, ctx.QueryAtTime(queryCPUCoresAllocated, end))
  702. }
  703. func (pds *PrometheusMetricsQuerier) QueryCPURequests(start, end time.Time) *source.Future[source.CPURequestsResult] {
  704. const queryName = "QueryCPURequests"
  705. const queryFmtCPURequests = `avg(avg_over_time(kube_pod_container_resource_requests{resource="cpu", unit="core", container!="", container!="POD", node!="", %s}[%s])) by (container, pod, namespace, node, uid, %s)`
  706. cfg := pds.promConfig
  707. durStr := timeutil.DurationString(end.Sub(start))
  708. if durStr == "" {
  709. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  710. }
  711. queryCPURequests := fmt.Sprintf(queryFmtCPURequests, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  712. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryCPURequests)
  713. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  714. return source.NewFuture(source.DecodeCPURequestsResult, ctx.QueryAtTime(queryCPURequests, end))
  715. }
  716. func (pds *PrometheusMetricsQuerier) QueryCPULimits(start, end time.Time) *source.Future[source.CPULimitsResult] {
  717. const queryName = "QueryCPULimits"
  718. const queryFmtCPULimits = `avg(avg_over_time(kube_pod_container_resource_limits{resource="cpu", unit="core", container!="", container!="POD", node!="", %s}[%s])) by (container, pod, namespace, node, %s)`
  719. cfg := pds.promConfig
  720. durStr := timeutil.DurationString(end.Sub(start))
  721. if durStr == "" {
  722. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  723. }
  724. queryCPULimits := fmt.Sprintf(queryFmtCPULimits, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  725. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryCPULimits)
  726. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  727. return source.NewFuture(source.DecodeCPULimitsResult, ctx.QueryAtTime(queryCPULimits, end))
  728. }
  729. func (pds *PrometheusMetricsQuerier) QueryCPUUsageAvg(start, end time.Time) *source.Future[source.CPUUsageAvgResult] {
  730. const queryName = "QueryCPUUsageAvg"
  731. const queryFmtCPUUsageAvg = `avg(rate(container_cpu_usage_seconds_total{container!="", container_name!="POD", container!="POD", %s}[%s])) by (container_name, container, pod_name, pod, namespace, node, instance, uid, %s)`
  732. cfg := pds.promConfig
  733. durStr := timeutil.DurationString(end.Sub(start))
  734. if durStr == "" {
  735. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  736. }
  737. queryCPUUsageAvg := fmt.Sprintf(queryFmtCPUUsageAvg, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  738. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryCPUUsageAvg)
  739. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  740. return source.NewFuture(source.DecodeCPUUsageAvgResult, ctx.QueryAtTime(queryCPUUsageAvg, end))
  741. }
  742. func (pds *PrometheusMetricsQuerier) QueryCPUUsageMax(start, end time.Time) *source.Future[source.CPUUsageMaxResult] {
  743. const queryName = "QueryCPUUsageMax"
  744. // Because we use container_cpu_usage_seconds_total to calculate CPU usage
  745. // at any given "instant" of time, we need to use an irate or rate. To then
  746. // calculate a max (or any aggregation) we have to perform an aggregation
  747. // query on top of an instant-by-instant maximum. Prometheus supports this
  748. // type of query with a "subquery" [1], however it is reportedly expensive
  749. // to make such a query. By default, Kubecost's Prometheus config includes
  750. // a recording rule that keeps track of the instant-by-instant irate for CPU
  751. // usage. The metric in this query is created by that recording rule.
  752. //
  753. // [1] https://prometheus.io/blog/2019/01/28/subquery-support/
  754. //
  755. // If changing the name of the recording rule, make sure to update the
  756. // corresponding diagnostic query to avoid confusion.
  757. const queryFmtCPUUsageMaxRecordingRule = `max(max_over_time(kubecost_container_cpu_usage_irate{%s}[%s])) by (container_name, container, pod_name, pod, namespace, node, instance, uid, %s)`
  758. // This is the subquery equivalent of the above recording rule query. It is
  759. // more expensive, but does not require the recording rule. It should be
  760. // used as a fallback query if the recording rule data does not exist.
  761. //
  762. // The parameter after the colon [:<thisone>] in the subquery affects the
  763. // resolution of the subquery.
  764. // The parameter after the metric ...{}[<thisone>] should be set to 2x
  765. // the resolution, to make sure the irate always has two points to query
  766. // in case the Prom scrape duration has been reduced to be equal to the
  767. // query resolution.
  768. const queryFmtCPUUsageMaxSubquery = `max(max_over_time(irate(container_cpu_usage_seconds_total{container!="POD", container!="", %s}[%dm])[%s:%dm])) by (container, pod_name, pod, namespace, node, instance, uid, %s)`
  769. cfg := pds.promConfig
  770. durStr := timeutil.DurationString(end.Sub(start))
  771. if durStr == "" {
  772. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  773. }
  774. queryCPUUsageMaxRecordingRule := fmt.Sprintf(queryFmtCPUUsageMaxRecordingRule, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  775. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryCPUUsageMaxRecordingRule)
  776. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  777. resCPUUsageMaxRR := ctx.QueryAtTime(queryCPUUsageMaxRecordingRule, end)
  778. resCPUUsageMax, _ := resCPUUsageMaxRR.Await()
  779. if len(resCPUUsageMax) > 0 {
  780. return source.NewFutureFrom(source.DecodeAll(resCPUUsageMax, source.DecodeCPUUsageMaxResult))
  781. }
  782. minsPerResolution := cfg.DataResolutionMinutes
  783. durStr = pds.durationStringFor(start, end, minsPerResolution, false)
  784. if durStr == "" {
  785. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  786. }
  787. queryCPUUsageMaxSubquery := fmt.Sprintf(queryFmtCPUUsageMaxSubquery, cfg.ClusterFilter, 2*minsPerResolution, durStr, minsPerResolution, cfg.ClusterLabel)
  788. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryCPUUsageMaxSubquery)
  789. return source.NewFuture(source.DecodeCPUUsageMaxResult, ctx.QueryAtTime(queryCPUUsageMaxSubquery, end))
  790. }
  791. func (pds *PrometheusMetricsQuerier) QueryGPUsRequested(start, end time.Time) *source.Future[source.GPUsRequestedResult] {
  792. const queryName = "QueryGPUsRequested"
  793. const queryFmtGPUsRequested = `avg(avg_over_time(kube_pod_container_resource_requests{resource="nvidia_com_gpu", container!="",container!="POD", node!="", %s}[%s])) by (container, pod, namespace, node, uid, %s)`
  794. cfg := pds.promConfig
  795. durStr := timeutil.DurationString(end.Sub(start))
  796. if durStr == "" {
  797. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  798. }
  799. queryGPUsRequested := fmt.Sprintf(queryFmtGPUsRequested, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  800. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryGPUsRequested)
  801. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  802. return source.NewFuture(source.DecodeGPUsRequestedResult, ctx.QueryAtTime(queryGPUsRequested, end))
  803. }
  804. func (pds *PrometheusMetricsQuerier) QueryGPUsUsageAvg(start, end time.Time) *source.Future[source.GPUsUsageAvgResult] {
  805. const queryName = "QueryGPUsUsageAvg"
  806. const queryFmtGPUsUsageAvg = `avg(avg_over_time(DCGM_FI_PROF_GR_ENGINE_ACTIVE{container!=""}[%s])) by (container, pod, namespace, pod_uid, %s)`
  807. cfg := pds.promConfig
  808. durStr := timeutil.DurationString(end.Sub(start))
  809. if durStr == "" {
  810. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  811. }
  812. queryGPUsUsageAvg := fmt.Sprintf(queryFmtGPUsUsageAvg, durStr, cfg.ClusterLabel)
  813. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryGPUsUsageAvg)
  814. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  815. return source.NewFuture(source.DecodeGPUsUsageAvgResult, ctx.QueryAtTime(queryGPUsUsageAvg, end))
  816. }
  817. func (pds *PrometheusMetricsQuerier) QueryGPUsUsageMax(start, end time.Time) *source.Future[source.GPUsUsageMaxResult] {
  818. const queryName = "QueryGPUsUsageMax"
  819. const queryFmtGPUsUsageMax = `max(max_over_time(DCGM_FI_PROF_GR_ENGINE_ACTIVE{container!=""}[%s])) by (container, pod, namespace, pod_uid, %s)`
  820. cfg := pds.promConfig
  821. durStr := timeutil.DurationString(end.Sub(start))
  822. if durStr == "" {
  823. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  824. }
  825. queryGPUsUsageMax := fmt.Sprintf(queryFmtGPUsUsageMax, durStr, cfg.ClusterLabel)
  826. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryGPUsUsageMax)
  827. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  828. return source.NewFuture(source.DecodeGPUsUsageMaxResult, ctx.QueryAtTime(queryGPUsUsageMax, end))
  829. }
  830. func (pds *PrometheusMetricsQuerier) QueryGPUsAllocated(start, end time.Time) *source.Future[source.GPUsAllocatedResult] {
  831. const queryName = "QueryGPUsAllocated"
  832. const queryFmtGPUsAllocated = `avg(avg_over_time(container_gpu_allocation{container!="", container!="POD", node!="", %s}[%s])) by (container, pod, namespace, node, uid, %s)`
  833. cfg := pds.promConfig
  834. durStr := timeutil.DurationString(end.Sub(start))
  835. if durStr == "" {
  836. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  837. }
  838. queryGPUsAllocated := fmt.Sprintf(queryFmtGPUsAllocated, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  839. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryGPUsAllocated)
  840. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  841. return source.NewFuture(source.DecodeGPUsAllocatedResult, ctx.QueryAtTime(queryGPUsAllocated, end))
  842. }
  843. func (pds *PrometheusMetricsQuerier) QueryIsGPUShared(start, end time.Time) *source.Future[source.IsGPUSharedResult] {
  844. const queryName = "QueryIsGPUShared"
  845. const queryFmtIsGPUShared = `avg(avg_over_time(kube_pod_container_resource_requests{container!="", node != "", pod != "", container!= "", unit = "integer", %s}[%s])) by (container, pod, namespace, node, resource, uid, %s)`
  846. cfg := pds.promConfig
  847. durStr := timeutil.DurationString(end.Sub(start))
  848. if durStr == "" {
  849. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  850. }
  851. queryIsGPUShared := fmt.Sprintf(queryFmtIsGPUShared, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  852. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryIsGPUShared)
  853. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  854. return source.NewFuture(source.DecodeIsGPUSharedResult, ctx.QueryAtTime(queryIsGPUShared, end))
  855. }
  856. func (pds *PrometheusMetricsQuerier) QueryDCGMDeviceInfo(start, end time.Time) *source.Future[source.DCGMDeviceInfoResult] {
  857. const queryName = "QueryDCGMDeviceInfo"
  858. const queryFmtDCGMDeviceInfo = `avg(avg_over_time(DCGM_FI_DEV_DEC_UTIL{%s}[%s])) by (UUID, device, modelName, Hostname, %s)`
  859. cfg := pds.promConfig
  860. durStr := timeutil.DurationString(end.Sub(start))
  861. if durStr == "" {
  862. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  863. }
  864. queryDCGMDeviceInfo := fmt.Sprintf(queryFmtDCGMDeviceInfo, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  865. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryDCGMDeviceInfo)
  866. ctx := pds.promContexts.NewNamedContext(ComputeCostDataContextName)
  867. return source.NewFuture(source.DecodeDCGMDeviceInfoResult, ctx.QueryAtTime(queryDCGMDeviceInfo, end))
  868. }
  869. func (pds *PrometheusMetricsQuerier) QueryDCGMDeviceUptime(start, end time.Time) *source.Future[source.DCGMDeviceUptimeResult] {
  870. const queryName = "QueryDCGMDeviceUptime"
  871. const queryFmtDCGMDeviceUptime = `avg(DCGM_FI_DEV_DEC_UTIL{%s}) by (UUID, %s)[%s:%dm]`
  872. cfg := pds.promConfig
  873. minsPerResolution := cfg.DataResolutionMinutes
  874. durStr := pds.durationStringFor(start, end, minsPerResolution, false)
  875. if durStr == "" {
  876. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  877. }
  878. queryDCGMDeviceUptime := fmt.Sprintf(queryFmtDCGMDeviceUptime, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution)
  879. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryDCGMDeviceUptime)
  880. ctx := pds.promContexts.NewNamedContext(ComputeCostDataContextName)
  881. return source.NewFuture(source.DecodeDCGMDeviceUptimeResult, ctx.QueryAtTime(queryDCGMDeviceUptime, end))
  882. }
  883. func (pds *PrometheusMetricsQuerier) QueryDCGMContainerUsageAvg(start, end time.Time) *source.Future[source.DCGMDeviceContainerUsageResult] {
  884. const queryName = "QueryDCGMContainerUsageAvg"
  885. const queryFmtDCGMContainerUsageAvg = `avg(avg_over_time(DCGM_FI_PROF_GR_ENGINE_ACTIVE{container!="", %s}[%s])) by (UUID, pod_uid, container, %s)`
  886. cfg := pds.promConfig
  887. durStr := timeutil.DurationString(end.Sub(start))
  888. if durStr == "" {
  889. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  890. }
  891. queryDCGMContainerUsageAvg := fmt.Sprintf(queryFmtDCGMContainerUsageAvg, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  892. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryDCGMContainerUsageAvg)
  893. ctx := pds.promContexts.NewNamedContext(ComputeCostDataContextName)
  894. return source.NewFuture(source.DecodeDCGMDeviceContainerUsageResult, ctx.QueryAtTime(queryDCGMContainerUsageAvg, end))
  895. }
  896. func (pds *PrometheusMetricsQuerier) QueryDCGMContainerUsageMax(start, end time.Time) *source.Future[source.DCGMDeviceContainerUsageResult] {
  897. const queryName = "QueryDCGMContainerUsageMax"
  898. const queryFmtDCGMContainerUsageMax = `max(max_over_time(DCGM_FI_PROF_GR_ENGINE_ACTIVE{container!="", %s}[%s])) by (UUID, pod_uid, container, %s)`
  899. cfg := pds.promConfig
  900. durStr := timeutil.DurationString(end.Sub(start))
  901. if durStr == "" {
  902. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  903. }
  904. queryDCGMContainerUsageMax := fmt.Sprintf(queryFmtDCGMContainerUsageMax, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  905. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryDCGMContainerUsageMax)
  906. ctx := pds.promContexts.NewNamedContext(ComputeCostDataContextName)
  907. return source.NewFuture(source.DecodeDCGMDeviceContainerUsageResult, ctx.QueryAtTime(queryDCGMContainerUsageMax, end))
  908. }
  909. func (pds *PrometheusMetricsQuerier) QueryGPUInfo(start, end time.Time) *source.Future[source.GPUInfoResult] {
  910. const queryName = "QueryGPUInfo"
  911. const queryFmtGetGPUInfo = `avg(avg_over_time(DCGM_FI_DEV_DEC_UTIL{container!="",%s}[%s])) by (container, pod, namespace, device, modelName, UUID, pod_uid, %s)`
  912. cfg := pds.promConfig
  913. durStr := timeutil.DurationString(end.Sub(start))
  914. if durStr == "" {
  915. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  916. }
  917. queryGetGPUInfo := fmt.Sprintf(queryFmtGetGPUInfo, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  918. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryGetGPUInfo)
  919. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  920. return source.NewFuture(source.DecodeGPUInfoResult, ctx.QueryAtTime(queryGetGPUInfo, end))
  921. }
  922. func (pds *PrometheusMetricsQuerier) QueryNodeCPUPricePerHr(start, end time.Time) *source.Future[source.NodeCPUPricePerHrResult] {
  923. const queryName = "QueryNodeCPUPricePerHr"
  924. const queryFmtNodeCostPerCPUHr = `avg(avg_over_time(node_cpu_hourly_cost{%s}[%s])) by (node, uid, %s, instance_type, provider_id)`
  925. cfg := pds.promConfig
  926. durStr := timeutil.DurationString(end.Sub(start))
  927. if durStr == "" {
  928. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  929. }
  930. queryNodeCostPerCPUHr := fmt.Sprintf(queryFmtNodeCostPerCPUHr, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  931. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNodeCostPerCPUHr)
  932. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  933. return source.NewFuture(source.DecodeNodeCPUPricePerHrResult, ctx.QueryAtTime(queryNodeCostPerCPUHr, end))
  934. }
  935. func (pds *PrometheusMetricsQuerier) QueryNodeRAMPricePerGiBHr(start, end time.Time) *source.Future[source.NodeRAMPricePerGiBHrResult] {
  936. const queryName = "QueryNodeRAMPricePerGiBHr"
  937. const queryFmtNodeCostPerRAMGiBHr = `avg(avg_over_time(node_ram_hourly_cost{%s}[%s])) by (node, uid, %s, instance_type, provider_id)`
  938. cfg := pds.promConfig
  939. durStr := timeutil.DurationString(end.Sub(start))
  940. if durStr == "" {
  941. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  942. }
  943. queryNodeCostPerRAMGiBHr := fmt.Sprintf(queryFmtNodeCostPerRAMGiBHr, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  944. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNodeCostPerRAMGiBHr)
  945. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  946. return source.NewFuture(source.DecodeNodeRAMPricePerGiBHrResult, ctx.QueryAtTime(queryNodeCostPerRAMGiBHr, end))
  947. }
  948. func (pds *PrometheusMetricsQuerier) QueryNodeGPUPricePerHr(start, end time.Time) *source.Future[source.NodeGPUPricePerHrResult] {
  949. const queryName = "QueryNodeGPUPricePerHr"
  950. const queryFmtNodeCostPerGPUHr = `avg(avg_over_time(node_gpu_hourly_cost{%s}[%s])) by (node, uid, %s, instance_type, provider_id)`
  951. cfg := pds.promConfig
  952. durStr := timeutil.DurationString(end.Sub(start))
  953. if durStr == "" {
  954. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  955. }
  956. queryNodeCostPerGPUHr := fmt.Sprintf(queryFmtNodeCostPerGPUHr, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  957. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNodeCostPerGPUHr)
  958. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  959. return source.NewFuture(source.DecodeNodeGPUPricePerHrResult, ctx.QueryAtTime(queryNodeCostPerGPUHr, end))
  960. }
  961. func (pds *PrometheusMetricsQuerier) QueryNodeIsSpot(start, end time.Time) *source.Future[source.NodeIsSpotResult] {
  962. const queryName = "QueryNodeIsSpot"
  963. const queryFmtNodeIsSpot = `avg_over_time(kubecost_node_is_spot{%s}[%s])`
  964. cfg := pds.promConfig
  965. durStr := timeutil.DurationString(end.Sub(start))
  966. if durStr == "" {
  967. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  968. }
  969. queryNodeIsSpot := fmt.Sprintf(queryFmtNodeIsSpot, cfg.ClusterFilter, durStr)
  970. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNodeIsSpot)
  971. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  972. return source.NewFuture(source.DecodeNodeIsSpotResult, ctx.QueryAtTime(queryNodeIsSpot, end))
  973. }
  974. func (pds *PrometheusMetricsQuerier) QueryPodPVCAllocation(start, end time.Time) *source.Future[source.PodPVCAllocationResult] {
  975. const queryName = "QueryPodPVCAllocation"
  976. const queryFmtPodPVCAllocation = `avg(avg_over_time(pod_pvc_allocation{%s}[%s])) by (persistentvolume, persistentvolumeclaim, pod, namespace, uid, %s)`
  977. cfg := pds.promConfig
  978. durStr := timeutil.DurationString(end.Sub(start))
  979. if durStr == "" {
  980. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  981. }
  982. queryPodPVCAllocation := fmt.Sprintf(queryFmtPodPVCAllocation, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  983. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryPodPVCAllocation)
  984. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  985. return source.NewFuture(source.DecodePodPVCAllocationResult, ctx.QueryAtTime(queryPodPVCAllocation, end))
  986. }
  987. func (pds *PrometheusMetricsQuerier) QueryPVCBytesRequested(start, end time.Time) *source.Future[source.PVCBytesRequestedResult] {
  988. const queryName = "QueryPVCBytesRequested"
  989. const queryFmtPVCBytesRequested = `avg(avg_over_time(kube_persistentvolumeclaim_resource_requests_storage_bytes{%s}[%s])) by (persistentvolumeclaim, namespace, uid, %s)`
  990. cfg := pds.promConfig
  991. durStr := timeutil.DurationString(end.Sub(start))
  992. if durStr == "" {
  993. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  994. }
  995. queryPVCBytesRequested := fmt.Sprintf(queryFmtPVCBytesRequested, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  996. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryPVCBytesRequested)
  997. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  998. return source.NewFuture(source.DecodePVCBytesRequestedResult, ctx.QueryAtTime(queryPVCBytesRequested, end))
  999. }
  1000. func (pds *PrometheusMetricsQuerier) QueryPVBytes(start, end time.Time) *source.Future[source.PVBytesResult] {
  1001. const queryName = "QueryPVBytes"
  1002. const queryFmtPVBytes = `avg(avg_over_time(kube_persistentvolume_capacity_bytes{%s}[%s])) by (persistentvolume, uid, %s)`
  1003. cfg := pds.promConfig
  1004. durStr := timeutil.DurationString(end.Sub(start))
  1005. if durStr == "" {
  1006. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1007. }
  1008. queryPVBytes := fmt.Sprintf(queryFmtPVBytes, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  1009. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryPVBytes)
  1010. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  1011. return source.NewFuture(source.DecodePVBytesResult, ctx.QueryAtTime(queryPVBytes, end))
  1012. }
  1013. func (pds *PrometheusMetricsQuerier) QueryPVInfo(start, end time.Time) *source.Future[source.PVInfoResult] {
  1014. const queryName = "QueryPVInfo"
  1015. const queryFmtPVMeta = `avg(avg_over_time(kubecost_pv_info{%s}[%s])) by (%s, storageclass, persistentvolume, uid, provider_id)`
  1016. cfg := pds.promConfig
  1017. durStr := timeutil.DurationString(end.Sub(start))
  1018. if durStr == "" {
  1019. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1020. }
  1021. queryPVMeta := fmt.Sprintf(queryFmtPVMeta, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  1022. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryPVMeta)
  1023. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  1024. return source.NewFuture(source.DecodePVInfoResult, ctx.QueryAtTime(queryPVMeta, end))
  1025. }
  1026. func (pds *PrometheusMetricsQuerier) QueryPVUptime(start, end time.Time) *source.Future[source.UptimeResult] {
  1027. const queryName = "QueryPVUptime"
  1028. const queryFmtPVUptime = `avg(kubecost_pv_info{%s}) by (%s, uid)[%s:%dm]`
  1029. cfg := pds.promConfig
  1030. minsPerResolution := cfg.DataResolutionMinutes
  1031. durStr := pds.durationStringFor(start, end, minsPerResolution, false)
  1032. if durStr == "" {
  1033. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1034. }
  1035. queryPVUptime := fmt.Sprintf(queryFmtPVUptime, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution)
  1036. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryPVUptime)
  1037. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  1038. return source.NewFuture(source.DecodeUptimeResult, ctx.QueryAtTime(queryPVUptime, end))
  1039. }
  1040. func (pds *PrometheusMetricsQuerier) QueryNetZoneGiB(start, end time.Time) *source.Future[source.NetZoneGiBResult] {
  1041. const queryName = "QueryNetZoneGiB"
  1042. const queryFmtNetZoneGiB = `sum(increase(kubecost_pod_network_egress_bytes_total{internet="false", same_zone="false", same_region="true", %s}[%s:%dm])) by (pod_name, namespace, uid, %s) / 1024 / 1024 / 1024`
  1043. cfg := pds.promConfig
  1044. minsPerResolution := cfg.DataResolutionMinutes
  1045. durStr := pds.durationStringFor(start, end, minsPerResolution, true)
  1046. if durStr == "" {
  1047. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1048. }
  1049. queryNetZoneGiB := fmt.Sprintf(queryFmtNetZoneGiB, cfg.ClusterFilter, durStr, minsPerResolution, cfg.ClusterLabel)
  1050. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNetZoneGiB)
  1051. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  1052. return source.NewFuture(source.DecodeNetZoneGiBResult, ctx.QueryAtTime(queryNetZoneGiB, end))
  1053. }
  1054. func (pds *PrometheusMetricsQuerier) QueryNetZonePricePerGiB(start, end time.Time) *source.Future[source.NetZonePricePerGiBResult] {
  1055. const queryName = "QueryNetZonePricePerGiB"
  1056. const queryFmtNetZoneCostPerGiB = `avg(avg_over_time(kubecost_network_zone_egress_cost{%s}[%s])) by (%s)`
  1057. cfg := pds.promConfig
  1058. durStr := timeutil.DurationString(end.Sub(start))
  1059. if durStr == "" {
  1060. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1061. }
  1062. queryNetZoneCostPerGiB := fmt.Sprintf(queryFmtNetZoneCostPerGiB, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  1063. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNetZoneCostPerGiB)
  1064. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  1065. return source.NewFuture(source.DecodeNetZonePricePerGiBResult, ctx.QueryAtTime(queryNetZoneCostPerGiB, end))
  1066. }
  1067. func (pds *PrometheusMetricsQuerier) QueryNetRegionGiB(start, end time.Time) *source.Future[source.NetRegionGiBResult] {
  1068. const queryName = "QueryNetRegionGiB"
  1069. const queryFmtNetRegionGiB = `sum(increase(kubecost_pod_network_egress_bytes_total{internet="false", same_zone="false", same_region="false", %s}[%s:%dm])) by (pod_name, namespace, uid, %s) / 1024 / 1024 / 1024`
  1070. cfg := pds.promConfig
  1071. minsPerResolution := cfg.DataResolutionMinutes
  1072. durStr := pds.durationStringFor(start, end, minsPerResolution, true)
  1073. if durStr == "" {
  1074. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1075. }
  1076. queryNetRegionGiB := fmt.Sprintf(queryFmtNetRegionGiB, cfg.ClusterFilter, durStr, minsPerResolution, cfg.ClusterLabel)
  1077. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNetRegionGiB)
  1078. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  1079. return source.NewFuture(source.DecodeNetRegionGiBResult, ctx.QueryAtTime(queryNetRegionGiB, end))
  1080. }
  1081. func (pds *PrometheusMetricsQuerier) QueryNetRegionPricePerGiB(start, end time.Time) *source.Future[source.NetRegionPricePerGiBResult] {
  1082. const queryName = "QueryNetRegionPricePerGiB"
  1083. const queryFmtNetRegionCostPerGiB = `avg(avg_over_time(kubecost_network_region_egress_cost{%s}[%s])) by (%s)`
  1084. cfg := pds.promConfig
  1085. durStr := timeutil.DurationString(end.Sub(start))
  1086. if durStr == "" {
  1087. panic("failed to parse duration string passed to QueryNetRegionPricePerGiB")
  1088. }
  1089. queryNetRegionCostPerGiB := fmt.Sprintf(queryFmtNetRegionCostPerGiB, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  1090. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNetRegionCostPerGiB)
  1091. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  1092. return source.NewFuture(source.DecodeNetRegionPricePerGiBResult, ctx.QueryAtTime(queryNetRegionCostPerGiB, end))
  1093. }
  1094. func (pds *PrometheusMetricsQuerier) QueryNetInternetGiB(start, end time.Time) *source.Future[source.NetInternetGiBResult] {
  1095. const queryName = "QueryNetInternetGiB"
  1096. const queryFmtNetInternetGiB = `sum(increase(kubecost_pod_network_egress_bytes_total{internet="true", %s}[%s:%dm])) by (pod_name, namespace, uid, %s) / 1024 / 1024 / 1024`
  1097. cfg := pds.promConfig
  1098. minsPerResolution := cfg.DataResolutionMinutes
  1099. durStr := pds.durationStringFor(start, end, minsPerResolution, true)
  1100. if durStr == "" {
  1101. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1102. }
  1103. queryNetInternetGiB := fmt.Sprintf(queryFmtNetInternetGiB, cfg.ClusterFilter, durStr, minsPerResolution, cfg.ClusterLabel)
  1104. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNetInternetGiB)
  1105. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  1106. return source.NewFuture(source.DecodeNetInternetGiBResult, ctx.QueryAtTime(queryNetInternetGiB, end))
  1107. }
  1108. func (pds *PrometheusMetricsQuerier) QueryNetInternetPricePerGiB(start, end time.Time) *source.Future[source.NetInternetPricePerGiBResult] {
  1109. const queryName = "QueryNetInternetPricePerGiB"
  1110. const queryFmtNetInternetCostPerGiB = `avg(avg_over_time(kubecost_network_internet_egress_cost{%s}[%s])) by (%s)`
  1111. cfg := pds.promConfig
  1112. durStr := timeutil.DurationString(end.Sub(start))
  1113. if durStr == "" {
  1114. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1115. }
  1116. queryNetInternetCostPerGiB := fmt.Sprintf(queryFmtNetInternetCostPerGiB, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  1117. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNetInternetCostPerGiB)
  1118. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  1119. return source.NewFuture(source.DecodeNetInternetPricePerGiBResult, ctx.QueryAtTime(queryNetInternetCostPerGiB, end))
  1120. }
  1121. func (pds *PrometheusMetricsQuerier) QueryNetInternetServiceGiB(start, end time.Time) *source.Future[source.NetInternetServiceGiBResult] {
  1122. const queryName = "QueryNetInternetServiceGiB"
  1123. const queryFmtNetInternetGiB = `sum(increase(kubecost_pod_network_egress_bytes_total{internet="true", %s}[%s:%dm])) by (pod_name, namespace, service, uid, %s) / 1024 / 1024 / 1024`
  1124. cfg := pds.promConfig
  1125. minsPerResolution := cfg.DataResolutionMinutes
  1126. durStr := pds.durationStringFor(start, end, minsPerResolution, true)
  1127. if durStr == "" {
  1128. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1129. }
  1130. queryNetInternetGiB := fmt.Sprintf(queryFmtNetInternetGiB, cfg.ClusterFilter, durStr, minsPerResolution, cfg.ClusterLabel)
  1131. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNetInternetGiB)
  1132. ctx := pds.promContexts.NewNamedContext(NetworkInsightsContextName)
  1133. return source.NewFuture(source.DecodeNetInternetServiceGiBResult, ctx.QueryAtTime(queryNetInternetGiB, end))
  1134. }
  1135. func (pds *PrometheusMetricsQuerier) QueryNetNatGatewayPricePerGiB(start, end time.Time) *source.Future[source.NetNatGatewayPricePerGiBResult] {
  1136. const queryName = "QueryNetNatGatewayPricePerGiB"
  1137. const queryFmtNetNatGatewayPricePerGiB = `avg(avg_over_time(kubecost_network_nat_gateway_egress_cost{%s}[%s])) by (%s)`
  1138. cfg := pds.promConfig
  1139. durStr := timeutil.DurationString(end.Sub(start))
  1140. if durStr == "" {
  1141. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1142. }
  1143. queryNetNatGatewayPricePerGiB := fmt.Sprintf(queryFmtNetNatGatewayPricePerGiB, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  1144. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNetNatGatewayPricePerGiB)
  1145. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  1146. return source.NewFuture(source.DecodeNetNatGatewayPricePerGiBResult, ctx.QueryAtTime(queryNetNatGatewayPricePerGiB, end))
  1147. }
  1148. func (pds *PrometheusMetricsQuerier) QueryNetNatGatewayGiB(start, end time.Time) *source.Future[source.NetNatGatewayGiBResult] {
  1149. const queryName = "QueryNetNatGatewayGiB"
  1150. const queryFmtNetNatGatewayGiB = `sum(increase(kubecost_pod_network_egress_bytes_total{nat_gateway="true", %s}[%s:%dm])) by (pod_name, namespace, service, uid, %s) / 1024 / 1024 / 1024`
  1151. cfg := pds.promConfig
  1152. minsPerResolution := cfg.DataResolutionMinutes
  1153. durStr := pds.durationStringFor(start, end, minsPerResolution, true)
  1154. if durStr == "" {
  1155. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1156. }
  1157. queryNetNatGatewayGiB := fmt.Sprintf(queryFmtNetNatGatewayGiB, cfg.ClusterFilter, durStr, minsPerResolution, cfg.ClusterLabel)
  1158. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNetNatGatewayGiB)
  1159. ctx := pds.promContexts.NewNamedContext(NetworkInsightsContextName)
  1160. return source.NewFuture(source.DecodeNetNatGatewayGiBResult, ctx.QueryAtTime(queryNetNatGatewayGiB, end))
  1161. }
  1162. func (pds *PrometheusMetricsQuerier) QueryNetTransferBytes(start, end time.Time) *source.Future[source.NetTransferBytesResult] {
  1163. const queryName = "QueryNetTransferBytes"
  1164. const queryFmtNetTransferBytes = `sum(increase(container_network_transmit_bytes_total{pod!="", %s}[%s:%dm])) by (pod_name, pod, namespace, uid, %s)`
  1165. cfg := pds.promConfig
  1166. minsPerResolution := cfg.DataResolutionMinutes
  1167. durStr := pds.durationStringFor(start, end, minsPerResolution, true)
  1168. if durStr == "" {
  1169. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1170. }
  1171. queryNetTransferBytes := fmt.Sprintf(queryFmtNetTransferBytes, cfg.ClusterFilter, durStr, minsPerResolution, cfg.ClusterLabel)
  1172. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNetTransferBytes)
  1173. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  1174. return source.NewFuture(source.DecodeNetTransferBytesResult, ctx.QueryAtTime(queryNetTransferBytes, end))
  1175. }
  1176. func (pds *PrometheusMetricsQuerier) QueryNetZoneIngressGiB(start, end time.Time) *source.Future[source.NetZoneIngressGiBResult] {
  1177. const queryName = "QueryNetZoneIngressGiB"
  1178. const queryFmtIngNetZoneGiB = `sum(increase(kubecost_pod_network_ingress_bytes_total{internet="false", same_zone="false", same_region="true", %s}[%s:%dm])) by (pod_name, namespace, uid, %s) / 1024 / 1024 / 1024`
  1179. cfg := pds.promConfig
  1180. minsPerResolution := cfg.DataResolutionMinutes
  1181. durStr := pds.durationStringFor(start, end, minsPerResolution, true)
  1182. if durStr == "" {
  1183. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1184. }
  1185. queryNetZoneCostPerGiB := fmt.Sprintf(queryFmtIngNetZoneGiB, cfg.ClusterFilter, durStr, minsPerResolution, cfg.ClusterLabel)
  1186. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNetZoneCostPerGiB)
  1187. ctx := pds.promContexts.NewNamedContext(NetworkInsightsContextName)
  1188. return source.NewFuture(source.DecodeNetZoneIngressGiBResult, ctx.QueryAtTime(queryNetZoneCostPerGiB, end))
  1189. }
  1190. func (pds *PrometheusMetricsQuerier) QueryNetRegionIngressGiB(start, end time.Time) *source.Future[source.NetRegionIngressGiBResult] {
  1191. const queryName = "QueryNetRegionIngressGiB"
  1192. const queryFmtIngNetRegionGiB = `sum(increase(kubecost_pod_network_ingress_bytes_total{internet="false", same_zone="false", same_region="false", %s}[%s:%dm])) by (pod_name, namespace, uid, %s) / 1024 / 1024 / 1024`
  1193. cfg := pds.promConfig
  1194. minsPerResolution := cfg.DataResolutionMinutes
  1195. durStr := pds.durationStringFor(start, end, minsPerResolution, true)
  1196. if durStr == "" {
  1197. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1198. }
  1199. queryNetRegionIngGiB := fmt.Sprintf(queryFmtIngNetRegionGiB, cfg.ClusterFilter, durStr, minsPerResolution, cfg.ClusterLabel)
  1200. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNetRegionIngGiB)
  1201. ctx := pds.promContexts.NewNamedContext(NetworkInsightsContextName)
  1202. return source.NewFuture(source.DecodeNetRegionIngressGiBResult, ctx.QueryAtTime(queryNetRegionIngGiB, end))
  1203. }
  1204. func (pds *PrometheusMetricsQuerier) QueryNetInternetIngressGiB(start, end time.Time) *source.Future[source.NetInternetIngressGiBResult] {
  1205. const queryName = "QueryNetInternetIngressGiB"
  1206. const queryFmtNetIngInternetGiB = `sum(increase(kubecost_pod_network_ingress_bytes_total{internet="true", %s}[%s:%dm])) by (pod_name, namespace, uid, %s) / 1024 / 1024 / 1024`
  1207. cfg := pds.promConfig
  1208. minsPerResolution := cfg.DataResolutionMinutes
  1209. durStr := pds.durationStringFor(start, end, minsPerResolution, true)
  1210. if durStr == "" {
  1211. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1212. }
  1213. queryNetIngInternetGiB := fmt.Sprintf(queryFmtNetIngInternetGiB, cfg.ClusterFilter, durStr, minsPerResolution, cfg.ClusterLabel)
  1214. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNetIngInternetGiB)
  1215. ctx := pds.promContexts.NewNamedContext(NetworkInsightsContextName)
  1216. return source.NewFuture(source.DecodeNetInternetIngressGiBResult, ctx.QueryAtTime(queryNetIngInternetGiB, end))
  1217. }
  1218. func (pds *PrometheusMetricsQuerier) QueryNetInternetServiceIngressGiB(start, end time.Time) *source.Future[source.NetInternetServiceIngressGiBResult] {
  1219. const queryName = "QueryNetInternetServiceIngressGiB"
  1220. const queryFmtIngNetInternetGiB = `sum(increase(kubecost_pod_network_ingress_bytes_total{internet="true", %s}[%s:%dm])) by (pod_name, namespace, service, uid, %s) / 1024 / 1024 / 1024`
  1221. cfg := pds.promConfig
  1222. minsPerResolution := cfg.DataResolutionMinutes
  1223. durStr := pds.durationStringFor(start, end, minsPerResolution, true)
  1224. if durStr == "" {
  1225. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1226. }
  1227. queryNetIngInternetGiB := fmt.Sprintf(queryFmtIngNetInternetGiB, cfg.ClusterFilter, durStr, minsPerResolution, cfg.ClusterLabel)
  1228. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNetIngInternetGiB)
  1229. ctx := pds.promContexts.NewNamedContext(NetworkInsightsContextName)
  1230. return source.NewFuture(source.DecodeNetInternetServiceIngressGiBResult, ctx.QueryAtTime(queryNetIngInternetGiB, end))
  1231. }
  1232. func (pds *PrometheusMetricsQuerier) QueryNetNatGatewayIngressPricePerGiB(start, end time.Time) *source.Future[source.NetNatGatewayPricePerGiBResult] {
  1233. const queryName = "QueryNetNatGatewayIngressPricePerGiB"
  1234. const queryFmtNetNatGatewayIngressPricePerGiB = `avg(avg_over_time(kubecost_network_nat_gateway_ingress_cost{%s}[%s])) by (%s)`
  1235. cfg := pds.promConfig
  1236. durStr := timeutil.DurationString(end.Sub(start))
  1237. if durStr == "" {
  1238. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1239. }
  1240. queryNetNatGatewayIngressPricePerGiB := fmt.Sprintf(queryFmtNetNatGatewayIngressPricePerGiB, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  1241. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNetNatGatewayIngressPricePerGiB)
  1242. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  1243. return source.NewFuture(source.DecodeNetNatGatewayPricePerGiBResult, ctx.QueryAtTime(queryNetNatGatewayIngressPricePerGiB, end))
  1244. }
  1245. func (pds *PrometheusMetricsQuerier) QueryNetNatGatewayIngressGiB(start, end time.Time) *source.Future[source.NetNatGatewayIngressGiBResult] {
  1246. const queryName = "QueryNetNatGatewayIngressGiB"
  1247. const queryFmtNetNatGatewayIngressGiB = `sum(increase(kubecost_pod_network_ingress_bytes_total{nat_gateway="true", %s}[%s:%dm])) by (pod_name, namespace, service, uid, %s) / 1024 / 1024 / 1024`
  1248. cfg := pds.promConfig
  1249. minsPerResolution := cfg.DataResolutionMinutes
  1250. durStr := pds.durationStringFor(start, end, minsPerResolution, true)
  1251. if durStr == "" {
  1252. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1253. }
  1254. queryNetNatGatewayIngressGiB := fmt.Sprintf(queryFmtNetNatGatewayIngressGiB, cfg.ClusterFilter, durStr, minsPerResolution, cfg.ClusterLabel)
  1255. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNetNatGatewayIngressGiB)
  1256. ctx := pds.promContexts.NewNamedContext(NetworkInsightsContextName)
  1257. return source.NewFuture(source.DecodeNetNatGatewayIngressGiBResult, ctx.QueryAtTime(queryNetNatGatewayIngressGiB, end))
  1258. }
  1259. func (pds *PrometheusMetricsQuerier) QueryNetReceiveBytes(start, end time.Time) *source.Future[source.NetReceiveBytesResult] {
  1260. const queryName = "QueryNetReceiveBytes"
  1261. const queryFmtNetReceiveBytes = `sum(increase(container_network_receive_bytes_total{pod!="", %s}[%s:%dm])) by (pod_name, pod, namespace, uid, %s)`
  1262. cfg := pds.promConfig
  1263. minsPerResolution := cfg.DataResolutionMinutes
  1264. durStr := pds.durationStringFor(start, end, minsPerResolution, true)
  1265. if durStr == "" {
  1266. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1267. }
  1268. queryNetReceiveBytes := fmt.Sprintf(queryFmtNetReceiveBytes, cfg.ClusterFilter, durStr, minsPerResolution, cfg.ClusterLabel)
  1269. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNetReceiveBytes)
  1270. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  1271. return source.NewFuture(source.DecodeNetReceiveBytesResult, ctx.QueryAtTime(queryNetReceiveBytes, end))
  1272. }
  1273. func (pds *PrometheusMetricsQuerier) QueryNamespaceInfo(start, end time.Time) *source.Future[source.NamespaceInfoResult] {
  1274. const queryName = "QueryNamespaceInfo"
  1275. const queryFmtNamespaceInfo = `avg(avg_over_time(namespace_info{%s}[%s])) by (%s, uid, namespace)`
  1276. cfg := pds.promConfig
  1277. durStr := timeutil.DurationString(end.Sub(start))
  1278. if durStr == "" {
  1279. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1280. }
  1281. queryNamespaceInfo := fmt.Sprintf(queryFmtNamespaceInfo, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  1282. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNamespaceInfo)
  1283. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  1284. return source.NewFuture(source.DecodeNamespaceInfoResult, ctx.QueryAtTime(queryNamespaceInfo, end))
  1285. }
  1286. // Note: namespace_info is not currently emitted
  1287. func (pds *PrometheusMetricsQuerier) QueryNamespaceUptime(start, end time.Time) *source.Future[source.UptimeResult] {
  1288. const queryName = "QueryNamespaceUptime"
  1289. const queryFmtNamespaceUptime = `avg(namespace_info{%s}) by (%s, uid)[%s:%dm]`
  1290. cfg := pds.promConfig
  1291. minsPerResolution := cfg.DataResolutionMinutes
  1292. durStr := pds.durationStringFor(start, end, minsPerResolution, false)
  1293. if durStr == "" {
  1294. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1295. }
  1296. queryNamespaceUptime := fmt.Sprintf(queryFmtNamespaceUptime, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution)
  1297. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryFmtNamespaceUptime)
  1298. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  1299. return source.NewFuture(source.DecodeUptimeResult, ctx.QueryAtTime(queryNamespaceUptime, end))
  1300. }
  1301. func (pds *PrometheusMetricsQuerier) QueryNamespaceLabels(start, end time.Time) *source.Future[source.NamespaceLabelsResult] {
  1302. const queryName = "QueryNamespaceLabels"
  1303. const queryFmtNamespaceLabels = `avg_over_time(kube_namespace_labels{%s}[%s])`
  1304. cfg := pds.promConfig
  1305. durStr := timeutil.DurationString(end.Sub(start))
  1306. if durStr == "" {
  1307. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1308. }
  1309. queryNamespaceLabels := fmt.Sprintf(queryFmtNamespaceLabels, cfg.ClusterFilter, durStr)
  1310. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNamespaceLabels)
  1311. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  1312. return source.NewFuture(source.DecodeNamespaceLabelsResult, ctx.QueryAtTime(queryNamespaceLabels, end))
  1313. }
  1314. func (pds *PrometheusMetricsQuerier) QueryNamespaceAnnotations(start, end time.Time) *source.Future[source.NamespaceAnnotationsResult] {
  1315. const queryName = "QueryNamespaceAnnotations"
  1316. const queryFmtNamespaceAnnotations = `avg_over_time(kube_namespace_annotations{%s}[%s])`
  1317. cfg := pds.promConfig
  1318. durStr := timeutil.DurationString(end.Sub(start))
  1319. if durStr == "" {
  1320. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1321. }
  1322. queryNamespaceAnnotations := fmt.Sprintf(queryFmtNamespaceAnnotations, cfg.ClusterFilter, durStr)
  1323. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNamespaceAnnotations)
  1324. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  1325. return source.NewFuture(source.DecodeNamespaceAnnotationsResult, ctx.QueryAtTime(queryNamespaceAnnotations, end))
  1326. }
  1327. func (pds *PrometheusMetricsQuerier) QueryPodLabels(start, end time.Time) *source.Future[source.PodLabelsResult] {
  1328. const queryName = "QueryPodLabels"
  1329. const queryFmtPodLabels = `avg_over_time(kube_pod_labels{%s}[%s])`
  1330. cfg := pds.promConfig
  1331. durStr := timeutil.DurationString(end.Sub(start))
  1332. if durStr == "" {
  1333. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1334. }
  1335. queryPodLabels := fmt.Sprintf(queryFmtPodLabels, cfg.ClusterFilter, durStr)
  1336. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryPodLabels)
  1337. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  1338. return source.NewFuture(source.DecodePodLabelsResult, ctx.QueryAtTime(queryPodLabels, end))
  1339. }
  1340. func (pds *PrometheusMetricsQuerier) QueryPodAnnotations(start, end time.Time) *source.Future[source.PodAnnotationsResult] {
  1341. const queryName = "QueryPodAnnotations"
  1342. const queryFmtPodAnnotations = `avg_over_time(kube_pod_annotations{%s}[%s])`
  1343. cfg := pds.promConfig
  1344. durStr := timeutil.DurationString(end.Sub(start))
  1345. if durStr == "" {
  1346. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1347. }
  1348. queryPodAnnotations := fmt.Sprintf(queryFmtPodAnnotations, cfg.ClusterFilter, durStr)
  1349. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryPodAnnotations)
  1350. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  1351. return source.NewFuture(source.DecodePodAnnotationsResult, ctx.QueryAtTime(queryPodAnnotations, end))
  1352. }
  1353. func (pds *PrometheusMetricsQuerier) QueryServiceInfo(start, end time.Time) *source.Future[source.ServiceInfoResult] {
  1354. const queryName = "QueryServiceInfo"
  1355. const queryFmtServiceInfo = `avg(avg_over_time(service_selector_labels{%s}[%s])) by (%s, uid, namespace_uid, service, service_type)`
  1356. cfg := pds.promConfig
  1357. durStr := timeutil.DurationString(end.Sub(start))
  1358. if durStr == "" {
  1359. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1360. }
  1361. queryServiceInfo := fmt.Sprintf(queryFmtServiceInfo, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  1362. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryServiceInfo)
  1363. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  1364. return source.NewFuture(source.DecodeServiceInfoResult, ctx.QueryAtTime(queryServiceInfo, end))
  1365. }
  1366. func (pds *PrometheusMetricsQuerier) QueryServiceUptime(start, end time.Time) *source.Future[source.UptimeResult] {
  1367. const queryName = "QueryServiceUptime"
  1368. const queryFmtServiceUptime = `avg(service_selector_labels{%s}) by (%s, uid)[%s:%dm]`
  1369. cfg := pds.promConfig
  1370. minsPerResolution := cfg.DataResolutionMinutes
  1371. durStr := pds.durationStringFor(start, end, minsPerResolution, false)
  1372. if durStr == "" {
  1373. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1374. }
  1375. queryServiceUptime := fmt.Sprintf(queryFmtServiceUptime, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution)
  1376. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryServiceUptime)
  1377. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  1378. return source.NewFuture(source.DecodeUptimeResult, ctx.QueryAtTime(queryServiceUptime, end))
  1379. }
  1380. func (pds *PrometheusMetricsQuerier) QueryServiceSelectorLabels(start, end time.Time) *source.Future[source.ServiceLabelsResult] {
  1381. const queryName = "QueryServiceSelectorLabels"
  1382. const queryFmtServiceSelectorLabels = `avg_over_time(service_selector_labels{%s}[%s])`
  1383. cfg := pds.promConfig
  1384. durStr := timeutil.DurationString(end.Sub(start))
  1385. if durStr == "" {
  1386. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1387. }
  1388. queryServiceSelectorLabels := fmt.Sprintf(queryFmtServiceSelectorLabels, cfg.ClusterFilter, durStr)
  1389. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryServiceSelectorLabels)
  1390. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  1391. return source.NewFuture(source.DecodeServiceLabelsResult, ctx.QueryAtTime(queryServiceSelectorLabels, end))
  1392. }
  1393. func (pds *PrometheusMetricsQuerier) QueryDeploymentInfo(start, end time.Time) *source.Future[source.DeploymentInfoResult] {
  1394. const queryName = "QueryDeploymentInfo"
  1395. const queryFmtDeploymentInfo = `avg(avg_over_time(deployment_info{%s}[%s])) by (%s, uid, namespace_uid, deployment)`
  1396. cfg := pds.promConfig
  1397. durStr := timeutil.DurationString(end.Sub(start))
  1398. if durStr == "" {
  1399. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1400. }
  1401. queryDeploymentInfo := fmt.Sprintf(queryFmtDeploymentInfo, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  1402. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryDeploymentInfo)
  1403. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  1404. return source.NewFuture(source.DecodeDeploymentInfoResult, ctx.QueryAtTime(queryDeploymentInfo, end))
  1405. }
  1406. func (pds *PrometheusMetricsQuerier) QueryDeploymentUptime(start, end time.Time) *source.Future[source.UptimeResult] {
  1407. const queryName = "QueryDeploymentUptime"
  1408. const queryFmtDeploymentUptime = `avg(deployment_info{%s}) by (%s, uid)[%s:%dm]`
  1409. cfg := pds.promConfig
  1410. minsPerResolution := cfg.DataResolutionMinutes
  1411. durStr := pds.durationStringFor(start, end, minsPerResolution, false)
  1412. if durStr == "" {
  1413. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1414. }
  1415. queryDeploymentUptime := fmt.Sprintf(queryFmtDeploymentUptime, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution)
  1416. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryDeploymentUptime)
  1417. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  1418. return source.NewFuture(source.DecodeUptimeResult, ctx.QueryAtTime(queryDeploymentUptime, end))
  1419. }
  1420. func (pds *PrometheusMetricsQuerier) QueryDeploymentLabels(start, end time.Time) *source.Future[source.LabelsResult] {
  1421. const queryName = "QueryDeploymentLabels"
  1422. const queryFmtDeploymentLabels = `avg_over_time(deployment_labels{%s}[%s])`
  1423. cfg := pds.promConfig
  1424. durStr := timeutil.DurationString(end.Sub(start))
  1425. if durStr == "" {
  1426. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1427. }
  1428. queryDeploymentLabels := fmt.Sprintf(queryFmtDeploymentLabels, cfg.ClusterFilter, durStr)
  1429. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryDeploymentLabels)
  1430. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  1431. return source.NewFuture(source.DecodeLabelsResult, ctx.QueryAtTime(queryDeploymentLabels, end))
  1432. }
  1433. func (pds *PrometheusMetricsQuerier) QueryDeploymentAnnotations(start, end time.Time) *source.Future[source.AnnotationsResult] {
  1434. const queryName = "QueryDeploymentAnnotations"
  1435. const queryFmtDeploymentAnnotations = `avg_over_time(deployment_annotations{%s}[%s])`
  1436. cfg := pds.promConfig
  1437. durStr := timeutil.DurationString(end.Sub(start))
  1438. if durStr == "" {
  1439. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1440. }
  1441. queryDeploymentAnnotations := fmt.Sprintf(queryFmtDeploymentAnnotations, cfg.ClusterFilter, durStr)
  1442. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryDeploymentAnnotations)
  1443. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  1444. return source.NewFuture(source.DecodeAnnotationsResult, ctx.QueryAtTime(queryDeploymentAnnotations, end))
  1445. }
  1446. func (pds *PrometheusMetricsQuerier) QueryDeploymentMatchLabels(start, end time.Time) *source.Future[source.DeploymentLabelsResult] {
  1447. const queryName = "QueryDeploymentMatchLabels"
  1448. const queryFmtDeploymentLabels = `avg_over_time(deployment_match_labels{%s}[%s])`
  1449. cfg := pds.promConfig
  1450. durStr := timeutil.DurationString(end.Sub(start))
  1451. if durStr == "" {
  1452. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1453. }
  1454. queryDeploymentMatchLabels := fmt.Sprintf(queryFmtDeploymentLabels, cfg.ClusterFilter, durStr)
  1455. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryDeploymentMatchLabels)
  1456. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  1457. return source.NewFuture(source.DecodeDeploymentLabelsResult, ctx.QueryAtTime(queryDeploymentMatchLabels, end))
  1458. }
  1459. func (pds *PrometheusMetricsQuerier) QueryStatefulSetInfo(start, end time.Time) *source.Future[source.StatefulSetInfoResult] {
  1460. const queryName = "QueryStatefulSetInfo"
  1461. const queryFmtStatefulSetInfo = `avg(avg_over_time(statefulset_info{%s}[%s])) by (%s, uid, namespace_uid, statefulSet)`
  1462. cfg := pds.promConfig
  1463. durStr := timeutil.DurationString(end.Sub(start))
  1464. if durStr == "" {
  1465. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1466. }
  1467. queryStatefulSetInfo := fmt.Sprintf(queryFmtStatefulSetInfo, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  1468. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryStatefulSetInfo)
  1469. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  1470. return source.NewFuture(source.DecodeStatefulSetInfoResult, ctx.QueryAtTime(queryStatefulSetInfo, end))
  1471. }
  1472. func (pds *PrometheusMetricsQuerier) QueryStatefulSetUptime(start, end time.Time) *source.Future[source.UptimeResult] {
  1473. const queryName = "QueryStatefulSetUptime"
  1474. const queryFmtStatefulSetUptime = `avg(statefulset_info{%s}) by (%s, uid)[%s:%dm]`
  1475. cfg := pds.promConfig
  1476. minsPerResolution := cfg.DataResolutionMinutes
  1477. durStr := pds.durationStringFor(start, end, minsPerResolution, false)
  1478. if durStr == "" {
  1479. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1480. }
  1481. queryStatefulSetUptime := fmt.Sprintf(queryFmtStatefulSetUptime, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution)
  1482. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryStatefulSetUptime)
  1483. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  1484. return source.NewFuture(source.DecodeUptimeResult, ctx.QueryAtTime(queryStatefulSetUptime, end))
  1485. }
  1486. func (pds *PrometheusMetricsQuerier) QueryStatefulSetLabels(start, end time.Time) *source.Future[source.LabelsResult] {
  1487. const queryName = "QueryStatefulSetLabels"
  1488. const queryFmtStatefulSetLabels = `avg_over_time(statefulset_labels{%s}[%s])`
  1489. cfg := pds.promConfig
  1490. durStr := timeutil.DurationString(end.Sub(start))
  1491. if durStr == "" {
  1492. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1493. }
  1494. queryStatefulSetLabels := fmt.Sprintf(queryFmtStatefulSetLabels, cfg.ClusterFilter, durStr)
  1495. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryStatefulSetLabels)
  1496. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  1497. return source.NewFuture(source.DecodeLabelsResult, ctx.QueryAtTime(queryStatefulSetLabels, end))
  1498. }
  1499. func (pds *PrometheusMetricsQuerier) QueryStatefulSetAnnotations(start, end time.Time) *source.Future[source.AnnotationsResult] {
  1500. const queryName = "QueryStatefulSetAnnotations"
  1501. const queryFmtStatefulSetAnnotations = `avg_over_time(statefulset_annotations{%s}[%s])`
  1502. cfg := pds.promConfig
  1503. durStr := timeutil.DurationString(end.Sub(start))
  1504. if durStr == "" {
  1505. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1506. }
  1507. queryStatefulSetAnnotations := fmt.Sprintf(queryFmtStatefulSetAnnotations, cfg.ClusterFilter, durStr)
  1508. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryStatefulSetAnnotations)
  1509. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  1510. return source.NewFuture(source.DecodeAnnotationsResult, ctx.QueryAtTime(queryStatefulSetAnnotations, end))
  1511. }
  1512. func (pds *PrometheusMetricsQuerier) QueryStatefulSetMatchLabels(start, end time.Time) *source.Future[source.StatefulSetLabelsResult] {
  1513. const queryName = "QueryStatefulSetMatchLabels"
  1514. const queryFmtStatefulSetMatchLabels = `avg_over_time(statefulSet_match_labels{%s}[%s])`
  1515. cfg := pds.promConfig
  1516. durStr := timeutil.DurationString(end.Sub(start))
  1517. if durStr == "" {
  1518. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1519. }
  1520. queryStatefulSetMatchLabels := fmt.Sprintf(queryFmtStatefulSetMatchLabels, cfg.ClusterFilter, durStr)
  1521. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryStatefulSetMatchLabels)
  1522. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  1523. return source.NewFuture(source.DecodeStatefulSetLabelsResult, ctx.QueryAtTime(queryStatefulSetMatchLabels, end))
  1524. }
  1525. func (pds *PrometheusMetricsQuerier) QueryDaemonSetInfo(start, end time.Time) *source.Future[source.DaemonSetInfoResult] {
  1526. const queryName = "QueryDaemonSetInfo"
  1527. const queryFmtDaemonSetInfo = `avg(avg_over_time(daemonset_info{%s}[%s])) by (%s, uid, namespace_uid, daemonset)`
  1528. cfg := pds.promConfig
  1529. durStr := timeutil.DurationString(end.Sub(start))
  1530. if durStr == "" {
  1531. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1532. }
  1533. queryDaemonSetInfo := fmt.Sprintf(queryFmtDaemonSetInfo, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  1534. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryDaemonSetInfo)
  1535. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  1536. return source.NewFuture(source.DecodeDaemonSetInfoResult, ctx.QueryAtTime(queryDaemonSetInfo, end))
  1537. }
  1538. func (pds *PrometheusMetricsQuerier) QueryDaemonSetUptime(start, end time.Time) *source.Future[source.UptimeResult] {
  1539. const queryName = "QueryDaemonSetUptime"
  1540. const queryFmtDaemonSetUptime = `avg(daemonset_info{%s}) by (%s, uid)[%s:%dm]`
  1541. cfg := pds.promConfig
  1542. minsPerResolution := cfg.DataResolutionMinutes
  1543. durStr := pds.durationStringFor(start, end, minsPerResolution, false)
  1544. if durStr == "" {
  1545. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1546. }
  1547. queryDaemonSetUptime := fmt.Sprintf(queryFmtDaemonSetUptime, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution)
  1548. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryDaemonSetUptime)
  1549. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  1550. return source.NewFuture(source.DecodeUptimeResult, ctx.QueryAtTime(queryDaemonSetUptime, end))
  1551. }
  1552. func (pds *PrometheusMetricsQuerier) QueryDaemonSetLabels(start, end time.Time) *source.Future[source.LabelsResult] {
  1553. const queryName = "QueryDaemonSetLabels"
  1554. const queryFmtDaemonSetLabels = `avg_over_time(daemonset_labels{%s}[%s])`
  1555. cfg := pds.promConfig
  1556. durStr := timeutil.DurationString(end.Sub(start))
  1557. if durStr == "" {
  1558. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1559. }
  1560. queryDaemonSetLabels := fmt.Sprintf(queryFmtDaemonSetLabels, cfg.ClusterFilter, durStr)
  1561. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryDaemonSetLabels)
  1562. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  1563. return source.NewFuture(source.DecodeLabelsResult, ctx.QueryAtTime(queryDaemonSetLabels, end))
  1564. }
  1565. func (pds *PrometheusMetricsQuerier) QueryDaemonSetAnnotations(start, end time.Time) *source.Future[source.AnnotationsResult] {
  1566. const queryName = "QueryDaemonSetAnnotations"
  1567. const queryFmtDaemonSetAnnotations = `avg_over_time(daemonset_annotations{%s}[%s])`
  1568. cfg := pds.promConfig
  1569. durStr := timeutil.DurationString(end.Sub(start))
  1570. if durStr == "" {
  1571. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1572. }
  1573. queryDaemonSetAnnotations := fmt.Sprintf(queryFmtDaemonSetAnnotations, cfg.ClusterFilter, durStr)
  1574. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryDaemonSetAnnotations)
  1575. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  1576. return source.NewFuture(source.DecodeAnnotationsResult, ctx.QueryAtTime(queryDaemonSetAnnotations, end))
  1577. }
  1578. func (pds *PrometheusMetricsQuerier) QueryJobInfo(start, end time.Time) *source.Future[source.JobInfoResult] {
  1579. const queryName = "QueryJobInfo"
  1580. const queryFmtJobInfo = `avg(avg_over_time(job_info{%s}[%s])) by (%s, uid, namespace_uid, job)`
  1581. cfg := pds.promConfig
  1582. durStr := timeutil.DurationString(end.Sub(start))
  1583. if durStr == "" {
  1584. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1585. }
  1586. queryJobInfo := fmt.Sprintf(queryFmtJobInfo, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  1587. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryJobInfo)
  1588. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  1589. return source.NewFuture(source.DecodeJobInfoResult, ctx.QueryAtTime(queryJobInfo, end))
  1590. }
  1591. func (pds *PrometheusMetricsQuerier) QueryJobUptime(start, end time.Time) *source.Future[source.UptimeResult] {
  1592. const queryName = "QueryJobUptime"
  1593. const queryFmtJobUptime = `avg(job_info{%s}) by (%s, uid)[%s:%dm]`
  1594. cfg := pds.promConfig
  1595. minsPerResolution := cfg.DataResolutionMinutes
  1596. durStr := pds.durationStringFor(start, end, minsPerResolution, false)
  1597. if durStr == "" {
  1598. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1599. }
  1600. queryJobUptime := fmt.Sprintf(queryFmtJobUptime, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution)
  1601. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryJobUptime)
  1602. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  1603. return source.NewFuture(source.DecodeUptimeResult, ctx.QueryAtTime(queryJobUptime, end))
  1604. }
  1605. func (pds *PrometheusMetricsQuerier) QueryJobLabels(start, end time.Time) *source.Future[source.LabelsResult] {
  1606. const queryName = "QueryJobLabels"
  1607. const queryFmtJobLabels = `avg_over_time(job_labels{%s}[%s])`
  1608. cfg := pds.promConfig
  1609. durStr := timeutil.DurationString(end.Sub(start))
  1610. if durStr == "" {
  1611. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1612. }
  1613. queryJobLabels := fmt.Sprintf(queryFmtJobLabels, cfg.ClusterFilter, durStr)
  1614. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryJobLabels)
  1615. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  1616. return source.NewFuture(source.DecodeLabelsResult, ctx.QueryAtTime(queryJobLabels, end))
  1617. }
  1618. func (pds *PrometheusMetricsQuerier) QueryJobAnnotations(start, end time.Time) *source.Future[source.AnnotationsResult] {
  1619. const queryName = "QueryJobAnnotations"
  1620. const queryFmtJobAnnotations = `avg_over_time(job_annotations{%s}[%s])`
  1621. cfg := pds.promConfig
  1622. durStr := timeutil.DurationString(end.Sub(start))
  1623. if durStr == "" {
  1624. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1625. }
  1626. queryJobAnnotations := fmt.Sprintf(queryFmtJobAnnotations, cfg.ClusterFilter, durStr)
  1627. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryJobAnnotations)
  1628. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  1629. return source.NewFuture(source.DecodeAnnotationsResult, ctx.QueryAtTime(queryJobAnnotations, end))
  1630. }
  1631. func (pds *PrometheusMetricsQuerier) QueryCronJobInfo(start, end time.Time) *source.Future[source.CronJobInfoResult] {
  1632. const queryName = "QueryCronJobInfo"
  1633. const queryFmtCronJobInfo = `avg(avg_over_time(cronjob_info{%s}[%s])) by (%s, uid, namespace_uid, cronjob)`
  1634. cfg := pds.promConfig
  1635. durStr := timeutil.DurationString(end.Sub(start))
  1636. if durStr == "" {
  1637. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1638. }
  1639. queryCronJobInfo := fmt.Sprintf(queryFmtCronJobInfo, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  1640. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryCronJobInfo)
  1641. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  1642. return source.NewFuture(source.DecodeCronJobInfoResult, ctx.QueryAtTime(queryCronJobInfo, end))
  1643. }
  1644. func (pds *PrometheusMetricsQuerier) QueryCronJobUptime(start, end time.Time) *source.Future[source.UptimeResult] {
  1645. const queryName = "QueryCronJobUptime"
  1646. const queryFmtCronJobUptime = `avg(cronjob_info{%s}) by (%s, uid)[%s:%dm]`
  1647. cfg := pds.promConfig
  1648. minsPerResolution := cfg.DataResolutionMinutes
  1649. durStr := pds.durationStringFor(start, end, minsPerResolution, false)
  1650. if durStr == "" {
  1651. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1652. }
  1653. queryCronJobUptime := fmt.Sprintf(queryFmtCronJobUptime, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution)
  1654. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryCronJobUptime)
  1655. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  1656. return source.NewFuture(source.DecodeUptimeResult, ctx.QueryAtTime(queryCronJobUptime, end))
  1657. }
  1658. func (pds *PrometheusMetricsQuerier) QueryCronJobLabels(start, end time.Time) *source.Future[source.LabelsResult] {
  1659. const queryName = "QueryCronJobLabels"
  1660. const queryFmtCronJobLabels = `avg_over_time(cronjob_labels{%s}[%s])`
  1661. cfg := pds.promConfig
  1662. durStr := timeutil.DurationString(end.Sub(start))
  1663. if durStr == "" {
  1664. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1665. }
  1666. queryCronJobLabels := fmt.Sprintf(queryFmtCronJobLabels, cfg.ClusterFilter, durStr)
  1667. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryCronJobLabels)
  1668. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  1669. return source.NewFuture(source.DecodeLabelsResult, ctx.QueryAtTime(queryCronJobLabels, end))
  1670. }
  1671. func (pds *PrometheusMetricsQuerier) QueryCronJobAnnotations(start, end time.Time) *source.Future[source.AnnotationsResult] {
  1672. const queryName = "QueryCronJobAnnotations"
  1673. const queryFmtCronJobAnnotations = `avg_over_time(cronjob_annotations{%s}[%s])`
  1674. cfg := pds.promConfig
  1675. durStr := timeutil.DurationString(end.Sub(start))
  1676. if durStr == "" {
  1677. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1678. }
  1679. queryCronJobAnnotations := fmt.Sprintf(queryFmtCronJobAnnotations, cfg.ClusterFilter, durStr)
  1680. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryCronJobAnnotations)
  1681. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  1682. return source.NewFuture(source.DecodeAnnotationsResult, ctx.QueryAtTime(queryCronJobAnnotations, end))
  1683. }
  1684. func (pds *PrometheusMetricsQuerier) QueryReplicaSetInfo(start, end time.Time) *source.Future[source.ReplicaSetInfoResult] {
  1685. const queryName = "QueryReplicaSetInfo"
  1686. const queryFmtReplicaSetInfo = `avg(avg_over_time(replicaset_info{%s}[%s])) by (%s, uid, namespace_uid, replicaset)`
  1687. cfg := pds.promConfig
  1688. durStr := timeutil.DurationString(end.Sub(start))
  1689. if durStr == "" {
  1690. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1691. }
  1692. queryReplicaSetInfo := fmt.Sprintf(queryFmtReplicaSetInfo, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  1693. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryReplicaSetInfo)
  1694. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  1695. return source.NewFuture(source.DecodeReplicaSetInfoResult, ctx.QueryAtTime(queryReplicaSetInfo, end))
  1696. }
  1697. func (pds *PrometheusMetricsQuerier) QueryReplicaSetUptime(start, end time.Time) *source.Future[source.UptimeResult] {
  1698. const queryName = "QueryReplicaSetUptime"
  1699. const queryFmtReplicaSetUptime = `avg(replicaset_info{%s}) by (%s, uid)[%s:%dm]`
  1700. cfg := pds.promConfig
  1701. minsPerResolution := cfg.DataResolutionMinutes
  1702. durStr := pds.durationStringFor(start, end, minsPerResolution, false)
  1703. if durStr == "" {
  1704. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1705. }
  1706. queryReplicaSetUptime := fmt.Sprintf(queryFmtReplicaSetUptime, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution)
  1707. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryReplicaSetUptime)
  1708. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  1709. return source.NewFuture(source.DecodeUptimeResult, ctx.QueryAtTime(queryReplicaSetUptime, end))
  1710. }
  1711. func (pds *PrometheusMetricsQuerier) QueryReplicaSetLabels(start, end time.Time) *source.Future[source.LabelsResult] {
  1712. const queryName = "QueryReplicaSetLabels"
  1713. const queryFmtReplicaSetLabels = `avg_over_time(replicaset_labels{%s}[%s])`
  1714. cfg := pds.promConfig
  1715. durStr := timeutil.DurationString(end.Sub(start))
  1716. if durStr == "" {
  1717. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1718. }
  1719. queryReplicaSetLabels := fmt.Sprintf(queryFmtReplicaSetLabels, cfg.ClusterFilter, durStr)
  1720. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryReplicaSetLabels)
  1721. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  1722. return source.NewFuture(source.DecodeLabelsResult, ctx.QueryAtTime(queryReplicaSetLabels, end))
  1723. }
  1724. func (pds *PrometheusMetricsQuerier) QueryReplicaSetAnnotations(start, end time.Time) *source.Future[source.AnnotationsResult] {
  1725. const queryName = "QueryReplicaSetAnnotations"
  1726. const queryFmtReplicaSetAnnotations = `avg_over_time(replicaset_annotations{%s}[%s])`
  1727. cfg := pds.promConfig
  1728. durStr := timeutil.DurationString(end.Sub(start))
  1729. if durStr == "" {
  1730. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1731. }
  1732. queryReplicaSetAnnotations := fmt.Sprintf(queryFmtReplicaSetAnnotations, cfg.ClusterFilter, durStr)
  1733. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryReplicaSetAnnotations)
  1734. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  1735. return source.NewFuture(source.DecodeAnnotationsResult, ctx.QueryAtTime(queryReplicaSetAnnotations, end))
  1736. }
  1737. func (pds *PrometheusMetricsQuerier) QueryReplicaSetOwners(start, end time.Time) *source.Future[source.OwnerResult] {
  1738. const queryName = "QueryReplicaSetOwners"
  1739. const queryFmtReplicaSetOwners = `avg(avg_over_time(kube_replicaset_owner{%s}[%s])) by (%s, uid, owner_uid, owner_kind)`
  1740. cfg := pds.promConfig
  1741. durStr := timeutil.DurationString(end.Sub(start))
  1742. if durStr == "" {
  1743. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1744. }
  1745. queryReplicaSetOwners := fmt.Sprintf(queryFmtReplicaSetOwners, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  1746. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryReplicaSetOwners)
  1747. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  1748. return source.NewFuture(source.DecodeOwnerResult, ctx.QueryAtTime(queryReplicaSetOwners, end))
  1749. }
  1750. func (pds *PrometheusMetricsQuerier) QueryPodsWithDaemonSetOwner(start, end time.Time) *source.Future[source.PodsWithDaemonSetOwnerResult] {
  1751. const queryName = "QueryPodsWithDaemonSetOwner"
  1752. const queryFmtPodsWithDaemonSetOwner = `sum(avg_over_time(kube_pod_owner{owner_kind="DaemonSet", %s}[%s])) by (pod, owner_name, namespace, uid, %s)`
  1753. cfg := pds.promConfig
  1754. durStr := timeutil.DurationString(end.Sub(start))
  1755. if durStr == "" {
  1756. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1757. }
  1758. queryPodsWithDaemonSetOwner := fmt.Sprintf(queryFmtPodsWithDaemonSetOwner, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  1759. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryPodsWithDaemonSetOwner)
  1760. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  1761. return source.NewFuture(source.DecodePodsWithDaemonSetOwnerResult, ctx.QueryAtTime(queryPodsWithDaemonSetOwner, end))
  1762. }
  1763. func (pds *PrometheusMetricsQuerier) QueryPodsWithJobOwner(start, end time.Time) *source.Future[source.PodsWithJobOwnerResult] {
  1764. const queryName = "QueryPodsWithJobOwner"
  1765. const queryFmtPodsWithJobOwner = `sum(avg_over_time(kube_pod_owner{owner_kind="Job", %s}[%s])) by (pod, owner_name, namespace, uid, %s)`
  1766. cfg := pds.promConfig
  1767. durStr := timeutil.DurationString(end.Sub(start))
  1768. if durStr == "" {
  1769. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1770. }
  1771. queryPodsWithJobOwner := fmt.Sprintf(queryFmtPodsWithJobOwner, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  1772. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryPodsWithJobOwner)
  1773. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  1774. return source.NewFuture(source.DecodePodsWithJobOwnerResult, ctx.QueryAtTime(queryPodsWithJobOwner, end))
  1775. }
  1776. func (pds *PrometheusMetricsQuerier) QueryPodsWithReplicaSetOwner(start, end time.Time) *source.Future[source.PodsWithReplicaSetOwnerResult] {
  1777. const queryName = "QueryPodsWithReplicaSetOwner"
  1778. const queryFmtPodsWithReplicaSetOwner = `sum(avg_over_time(kube_pod_owner{owner_kind="ReplicaSet", %s}[%s])) by (pod, owner_name, namespace, uid, %s)`
  1779. cfg := pds.promConfig
  1780. durStr := timeutil.DurationString(end.Sub(start))
  1781. if durStr == "" {
  1782. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1783. }
  1784. queryPodsWithReplicaSetOwner := fmt.Sprintf(queryFmtPodsWithReplicaSetOwner, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  1785. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryPodsWithReplicaSetOwner)
  1786. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  1787. return source.NewFuture(source.DecodePodsWithReplicaSetOwnerResult, ctx.QueryAtTime(queryPodsWithReplicaSetOwner, end))
  1788. }
  1789. func (pds *PrometheusMetricsQuerier) QueryReplicaSetsWithoutOwners(start, end time.Time) *source.Future[source.ReplicaSetsWithoutOwnersResult] {
  1790. const queryName = "QueryReplicaSetsWithoutOwners"
  1791. const queryFmtReplicaSetsWithoutOwners = `avg(avg_over_time(kube_replicaset_owner{owner_kind="<none>", owner_name="<none>", %s}[%s])) by (replicaset, namespace, uid, %s)`
  1792. cfg := pds.promConfig
  1793. durStr := timeutil.DurationString(end.Sub(start))
  1794. if durStr == "" {
  1795. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1796. }
  1797. queryReplicaSetsWithoutOwners := fmt.Sprintf(queryFmtReplicaSetsWithoutOwners, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  1798. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryReplicaSetsWithoutOwners)
  1799. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  1800. return source.NewFuture(source.DecodeReplicaSetsWithoutOwnersResult, ctx.QueryAtTime(queryReplicaSetsWithoutOwners, end))
  1801. }
  1802. func (pds *PrometheusMetricsQuerier) QueryReplicaSetsWithRollout(start, end time.Time) *source.Future[source.ReplicaSetsWithRolloutResult] {
  1803. const queryName = "QueryReplicaSetsWithRollout"
  1804. const queryFmtReplicaSetsWithRolloutOwner = `avg(avg_over_time(kube_replicaset_owner{owner_kind="Rollout", %s}[%s])) by (replicaset, namespace, owner_kind, owner_name, uid, %s)`
  1805. cfg := pds.promConfig
  1806. durStr := timeutil.DurationString(end.Sub(start))
  1807. if durStr == "" {
  1808. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1809. }
  1810. queryReplicaSetsWithRolloutOwner := fmt.Sprintf(queryFmtReplicaSetsWithRolloutOwner, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  1811. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryReplicaSetsWithRolloutOwner)
  1812. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  1813. return source.NewFuture(source.DecodeReplicaSetsWithRolloutResult, ctx.QueryAtTime(queryReplicaSetsWithRolloutOwner, end))
  1814. }
  1815. // Note: The ResourceQuota metrics are _not_ emitted at the moment. Leaving the query implementations here in case we add metric emission later on.
  1816. func (pds *PrometheusMetricsQuerier) QueryResourceQuotaInfo(start, end time.Time) *source.Future[source.ResourceQuotaInfoResult] {
  1817. const queryName = "QueryResourceQuotaInfo"
  1818. const queryFmtResourceQuotaInfo = `avg(avg_over_time(resourcequota_info{%s}[%s])) by (%s, uid, namespace_uid, resourcequota)`
  1819. cfg := pds.promConfig
  1820. durStr := timeutil.DurationString(end.Sub(start))
  1821. if durStr == "" {
  1822. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1823. }
  1824. queryResourceQuotaInfo := fmt.Sprintf(queryFmtResourceQuotaInfo, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  1825. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryResourceQuotaInfo)
  1826. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  1827. return source.NewFuture(source.DecodeResourceQuotaInfoResult, ctx.QueryAtTime(queryResourceQuotaInfo, end))
  1828. }
  1829. func (pds *PrometheusMetricsQuerier) QueryResourceQuotaUptime(start, end time.Time) *source.Future[source.UptimeResult] {
  1830. const queryName = "QueryResourceQuotaUptime"
  1831. const queryFmtResourceQuotaUptime = `avg(resourcequota_info{%s}) by (%s, uid)[%s:%dm]`
  1832. cfg := pds.promConfig
  1833. minsPerResolution := cfg.DataResolutionMinutes
  1834. durStr := pds.durationStringFor(start, end, minsPerResolution, false)
  1835. if durStr == "" {
  1836. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1837. }
  1838. queryResourceQuotaUptime := fmt.Sprintf(queryFmtResourceQuotaUptime, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution)
  1839. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryFmtResourceQuotaUptime)
  1840. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  1841. return source.NewFuture(source.DecodeUptimeResult, ctx.QueryAtTime(queryResourceQuotaUptime, end))
  1842. }
  1843. func (pds *PrometheusMetricsQuerier) QueryResourceQuotaSpecCPURequestAverage(start, end time.Time) *source.Future[source.ResourceResult] {
  1844. const queryName = "QueryResourceQuotaSpecCPURequestAverage"
  1845. const queryFmtResourceQuotaSpecCPURequests = `avg(avg_over_time(resourcequota_spec_resource_requests{resource="cpu",unit="core", %s}[%s])) by (resourcequota, namespace, uid, %s)`
  1846. cfg := pds.promConfig
  1847. durStr := timeutil.DurationString(end.Sub(start))
  1848. if durStr == "" {
  1849. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1850. }
  1851. queryResourceQuotaSpecCPURequests := fmt.Sprintf(queryFmtResourceQuotaSpecCPURequests, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  1852. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryResourceQuotaSpecCPURequests)
  1853. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  1854. return source.NewFuture(source.DecodeResourceResult, ctx.QueryAtTime(queryResourceQuotaSpecCPURequests, end))
  1855. }
  1856. func (pds *PrometheusMetricsQuerier) QueryResourceQuotaSpecCPURequestMax(start, end time.Time) *source.Future[source.ResourceResult] {
  1857. const queryName = "QueryResourceQuotaSpecCPURequestMax"
  1858. const queryFmtResourceQuotaSpecCPURequests = `max(max_over_time(resourcequota_spec_resource_requests{resource="cpu",unit="core", %s}[%s])) by (resourcequota, namespace, uid, %s)`
  1859. cfg := pds.promConfig
  1860. durStr := timeutil.DurationString(end.Sub(start))
  1861. if durStr == "" {
  1862. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1863. }
  1864. queryResourceQuotaSpecCPURequests := fmt.Sprintf(queryFmtResourceQuotaSpecCPURequests, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  1865. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryResourceQuotaSpecCPURequests)
  1866. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  1867. return source.NewFuture(source.DecodeResourceResult, ctx.QueryAtTime(queryResourceQuotaSpecCPURequests, end))
  1868. }
  1869. func (pds *PrometheusMetricsQuerier) QueryResourceQuotaSpecRAMRequestAverage(start, end time.Time) *source.Future[source.ResourceResult] {
  1870. const queryName = "QueryResourceQuotaSpecRAMRequestAverage"
  1871. const queryFmtResourceQuotaSpecRAMRequests = `avg(avg_over_time(resourcequota_spec_resource_requests{resource="memory",unit="byte", %s}[%s])) by (resourcequota, namespace, uid, %s)`
  1872. cfg := pds.promConfig
  1873. durStr := timeutil.DurationString(end.Sub(start))
  1874. if durStr == "" {
  1875. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1876. }
  1877. queryResourceQuotaSpecRAMRequests := fmt.Sprintf(queryFmtResourceQuotaSpecRAMRequests, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  1878. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryResourceQuotaSpecRAMRequests)
  1879. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  1880. return source.NewFuture(source.DecodeResourceResult, ctx.QueryAtTime(queryResourceQuotaSpecRAMRequests, end))
  1881. }
  1882. func (pds *PrometheusMetricsQuerier) QueryResourceQuotaSpecRAMRequestMax(start, end time.Time) *source.Future[source.ResourceResult] {
  1883. const queryName = "QueryResourceQuotaSpecRAMRequestMax"
  1884. const queryFmtResourceQuotaSpecRAMRequests = `max(max_over_time(resourcequota_spec_resource_requests{resource="memory",unit="byte", %s}[%s])) by (resourcequota, namespace, uid, %s)`
  1885. cfg := pds.promConfig
  1886. durStr := timeutil.DurationString(end.Sub(start))
  1887. if durStr == "" {
  1888. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1889. }
  1890. queryResourceQuotaSpecRAMRequests := fmt.Sprintf(queryFmtResourceQuotaSpecRAMRequests, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  1891. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryResourceQuotaSpecRAMRequests)
  1892. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  1893. return source.NewFuture(source.DecodeResourceResult, ctx.QueryAtTime(queryResourceQuotaSpecRAMRequests, end))
  1894. }
  1895. func (pds *PrometheusMetricsQuerier) QueryResourceQuotaSpecCPULimitAverage(start, end time.Time) *source.Future[source.ResourceResult] {
  1896. const queryName = "QueryResourceQuotaSpecCPULimitAverage"
  1897. const queryFmtResourceQuotaSpecCPULimits = `avg(avg_over_time(resourcequota_spec_resource_limits{resource="cpu",unit="core", %s}[%s])) by (resourcequota, namespace, uid, %s)`
  1898. cfg := pds.promConfig
  1899. durStr := timeutil.DurationString(end.Sub(start))
  1900. if durStr == "" {
  1901. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1902. }
  1903. queryResourceQuotaSpecCPULimits := fmt.Sprintf(queryFmtResourceQuotaSpecCPULimits, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  1904. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryResourceQuotaSpecCPULimits)
  1905. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  1906. return source.NewFuture(source.DecodeResourceResult, ctx.QueryAtTime(queryResourceQuotaSpecCPULimits, end))
  1907. }
  1908. func (pds *PrometheusMetricsQuerier) QueryResourceQuotaSpecCPULimitMax(start, end time.Time) *source.Future[source.ResourceResult] {
  1909. const queryName = "QueryResourceQuotaSpecCPULimitMax"
  1910. const queryFmtResourceQuotaSpecCPULimits = `max(max_over_time(resourcequota_spec_resource_limits{resource="cpu",unit="core", %s}[%s])) by (resourcequota, namespace, uid, %s)`
  1911. cfg := pds.promConfig
  1912. durStr := timeutil.DurationString(end.Sub(start))
  1913. if durStr == "" {
  1914. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1915. }
  1916. queryResourceQuotaSpecCPULimits := fmt.Sprintf(queryFmtResourceQuotaSpecCPULimits, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  1917. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryResourceQuotaSpecCPULimits)
  1918. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  1919. return source.NewFuture(source.DecodeResourceResult, ctx.QueryAtTime(queryResourceQuotaSpecCPULimits, end))
  1920. }
  1921. func (pds *PrometheusMetricsQuerier) QueryResourceQuotaSpecRAMLimitAverage(start, end time.Time) *source.Future[source.ResourceResult] {
  1922. const queryName = "QueryResourceQuotaSpecRAMLimitAverage"
  1923. const queryFmtResourceQuotaSpecRAMLimits = `avg(avg_over_time(resourcequota_spec_resource_limits{resource="memory",unit="byte", %s}[%s])) by (resourcequota, namespace, uid, %s)`
  1924. cfg := pds.promConfig
  1925. durStr := timeutil.DurationString(end.Sub(start))
  1926. if durStr == "" {
  1927. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1928. }
  1929. queryResourceQuotaSpecRAMLimits := fmt.Sprintf(queryFmtResourceQuotaSpecRAMLimits, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  1930. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryResourceQuotaSpecRAMLimits)
  1931. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  1932. return source.NewFuture(source.DecodeResourceResult, ctx.QueryAtTime(queryResourceQuotaSpecRAMLimits, end))
  1933. }
  1934. func (pds *PrometheusMetricsQuerier) QueryResourceQuotaSpecRAMLimitMax(start, end time.Time) *source.Future[source.ResourceResult] {
  1935. const queryName = "QueryResourceQuotaSpecRAMLimitMax"
  1936. const queryFmtResourceQuotaSpecRAMLimits = `max(max_over_time(resourcequota_spec_resource_limits{resource="memory",unit="byte", %s}[%s])) by (resourcequota, namespace, uid, %s)`
  1937. cfg := pds.promConfig
  1938. durStr := timeutil.DurationString(end.Sub(start))
  1939. if durStr == "" {
  1940. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1941. }
  1942. queryResourceQuotaSpecRAMLimits := fmt.Sprintf(queryFmtResourceQuotaSpecRAMLimits, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  1943. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryResourceQuotaSpecRAMLimits)
  1944. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  1945. return source.NewFuture(source.DecodeResourceResult, ctx.QueryAtTime(queryResourceQuotaSpecRAMLimits, end))
  1946. }
  1947. func (pds *PrometheusMetricsQuerier) QueryResourceQuotaStatusUsedCPURequestAverage(start, end time.Time) *source.Future[source.ResourceResult] {
  1948. const queryName = "QueryResourceQuotaStatusUsedCPURequestAverage"
  1949. const queryFmtResourceQuotaStatusUsedCPURequests = `avg(avg_over_time(resourcequota_status_used_resource_requests{resource="cpu",unit="core", %s}[%s])) by (resourcequota, namespace, uid, %s)`
  1950. cfg := pds.promConfig
  1951. durStr := timeutil.DurationString(end.Sub(start))
  1952. if durStr == "" {
  1953. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1954. }
  1955. queryResourceQuotaStatusUsedCPURequests := fmt.Sprintf(queryFmtResourceQuotaStatusUsedCPURequests, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  1956. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryResourceQuotaStatusUsedCPURequests)
  1957. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  1958. return source.NewFuture(source.DecodeResourceResult, ctx.QueryAtTime(queryResourceQuotaStatusUsedCPURequests, end))
  1959. }
  1960. func (pds *PrometheusMetricsQuerier) QueryResourceQuotaStatusUsedCPURequestMax(start, end time.Time) *source.Future[source.ResourceResult] {
  1961. const queryName = "QueryResourceQuotaStatusUsedCPURequestMax"
  1962. const queryFmtResourceQuotaStatusUsedCPURequests = `max(max_over_time(resourcequota_status_used_resource_requests{resource="cpu",unit="core", %s}[%s])) by (resourcequota, namespace, uid, %s)`
  1963. cfg := pds.promConfig
  1964. durStr := timeutil.DurationString(end.Sub(start))
  1965. if durStr == "" {
  1966. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1967. }
  1968. queryResourceQuotaStatusUsedCPURequests := fmt.Sprintf(queryFmtResourceQuotaStatusUsedCPURequests, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  1969. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryResourceQuotaStatusUsedCPURequests)
  1970. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  1971. return source.NewFuture(source.DecodeResourceResult, ctx.QueryAtTime(queryResourceQuotaStatusUsedCPURequests, end))
  1972. }
  1973. func (pds *PrometheusMetricsQuerier) QueryResourceQuotaStatusUsedRAMRequestAverage(start, end time.Time) *source.Future[source.ResourceResult] {
  1974. const queryName = "QueryResourceQuotaStatusUsedRAMRequestAverage"
  1975. const queryFmtResourceQuotaStatusUsedRAMRequests = `avg(avg_over_time(resourcequota_status_used_resource_requests{resource="memory",unit="byte", %s}[%s])) by (resourcequota, namespace, uid, %s)`
  1976. cfg := pds.promConfig
  1977. durStr := timeutil.DurationString(end.Sub(start))
  1978. if durStr == "" {
  1979. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1980. }
  1981. queryResourceQuotaStatusUsedRAMRequests := fmt.Sprintf(queryFmtResourceQuotaStatusUsedRAMRequests, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  1982. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryResourceQuotaStatusUsedRAMRequests)
  1983. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  1984. return source.NewFuture(source.DecodeResourceResult, ctx.QueryAtTime(queryResourceQuotaStatusUsedRAMRequests, end))
  1985. }
  1986. func (pds *PrometheusMetricsQuerier) QueryResourceQuotaStatusUsedRAMRequestMax(start, end time.Time) *source.Future[source.ResourceResult] {
  1987. const queryName = "QueryResourceQuotaStatusUsedRAMRequestMax"
  1988. const queryFmtResourceQuotaStatusUsedRAMRequests = `max(max_over_time(resourcequota_status_used_resource_requests{resource="memory",unit="byte", %s}[%s])) by (resourcequota, namespace, uid, %s)`
  1989. cfg := pds.promConfig
  1990. durStr := timeutil.DurationString(end.Sub(start))
  1991. if durStr == "" {
  1992. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  1993. }
  1994. queryResourceQuotaStatusUsedRAMRequests := fmt.Sprintf(queryFmtResourceQuotaStatusUsedRAMRequests, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  1995. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryResourceQuotaStatusUsedRAMRequests)
  1996. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  1997. return source.NewFuture(source.DecodeResourceResult, ctx.QueryAtTime(queryResourceQuotaStatusUsedRAMRequests, end))
  1998. }
  1999. func (pds *PrometheusMetricsQuerier) QueryResourceQuotaStatusUsedCPULimitAverage(start, end time.Time) *source.Future[source.ResourceResult] {
  2000. const queryName = "QueryResourceQuotaStatusUsedCPULimitAverage"
  2001. const queryFmtResourceQuotaStatusUsedCPULimits = `avg(avg_over_time(resourcequota_status_used_resource_limits{resource="cpu",unit="core", %s}[%s])) by (resourcequota, namespace, uid, %s)`
  2002. cfg := pds.promConfig
  2003. durStr := timeutil.DurationString(end.Sub(start))
  2004. if durStr == "" {
  2005. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  2006. }
  2007. queryResourceQuotaStatusUsedCPULimits := fmt.Sprintf(queryFmtResourceQuotaStatusUsedCPULimits, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  2008. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryResourceQuotaStatusUsedCPULimits)
  2009. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  2010. return source.NewFuture(source.DecodeResourceResult, ctx.QueryAtTime(queryResourceQuotaStatusUsedCPULimits, end))
  2011. }
  2012. func (pds *PrometheusMetricsQuerier) QueryResourceQuotaStatusUsedCPULimitMax(start, end time.Time) *source.Future[source.ResourceResult] {
  2013. const queryName = "QueryResourceQuotaStatusUsedCPULimitMax"
  2014. const queryFmtResourceQuotaStatusUsedCPULimits = `max(max_over_time(resourcequota_status_used_resource_limits{resource="cpu",unit="core", %s}[%s])) by (resourcequota, namespace, uid, %s)`
  2015. cfg := pds.promConfig
  2016. durStr := timeutil.DurationString(end.Sub(start))
  2017. if durStr == "" {
  2018. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  2019. }
  2020. queryResourceQuotaStatusUsedCPULimits := fmt.Sprintf(queryFmtResourceQuotaStatusUsedCPULimits, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  2021. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryResourceQuotaStatusUsedCPULimits)
  2022. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  2023. return source.NewFuture(source.DecodeResourceResult, ctx.QueryAtTime(queryResourceQuotaStatusUsedCPULimits, end))
  2024. }
  2025. func (pds *PrometheusMetricsQuerier) QueryResourceQuotaStatusUsedRAMLimitAverage(start, end time.Time) *source.Future[source.ResourceResult] {
  2026. const queryName = "QueryResourceQuotaStatusUsedRAMLimitAverage"
  2027. const queryFmtResourceQuotaStatusUsedRAMLimits = `avg(avg_over_time(resourcequota_status_used_resource_limits{resource="memory",unit="byte", %s}[%s])) by (resourcequota, namespace, uid, %s)`
  2028. cfg := pds.promConfig
  2029. durStr := timeutil.DurationString(end.Sub(start))
  2030. if durStr == "" {
  2031. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  2032. }
  2033. queryResourceQuotaStatusUsedRAMLimits := fmt.Sprintf(queryFmtResourceQuotaStatusUsedRAMLimits, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  2034. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryResourceQuotaStatusUsedRAMLimits)
  2035. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  2036. return source.NewFuture(source.DecodeResourceResult, ctx.QueryAtTime(queryResourceQuotaStatusUsedRAMLimits, end))
  2037. }
  2038. func (pds *PrometheusMetricsQuerier) QueryResourceQuotaStatusUsedRAMLimitMax(start, end time.Time) *source.Future[source.ResourceResult] {
  2039. const queryName = "QueryResourceQuotaStatusUsedRAMLimitMax"
  2040. const queryFmtResourceQuotaStatusUsedRAMLimits = `max(max_over_time(resourcequota_status_used_resource_limits{resource="memory",unit="byte", %s}[%s])) by (resourcequota, namespace, uid, %s)`
  2041. cfg := pds.promConfig
  2042. durStr := timeutil.DurationString(end.Sub(start))
  2043. if durStr == "" {
  2044. panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName))
  2045. }
  2046. queryResourceQuotaStatusUsedRAMLimits := fmt.Sprintf(queryFmtResourceQuotaStatusUsedRAMLimits, cfg.ClusterFilter, durStr, cfg.ClusterLabel)
  2047. log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryResourceQuotaStatusUsedRAMLimits)
  2048. ctx := pds.promContexts.NewNamedContext(KubeModelContextName)
  2049. return source.NewFuture(source.DecodeResourceResult, ctx.QueryAtTime(queryResourceQuotaStatusUsedRAMLimits, end))
  2050. }
  2051. func (pds *PrometheusMetricsQuerier) QueryDataCoverage(limitDays int) (time.Time, time.Time, error) {
  2052. const (
  2053. queryName = "QueryDataCoverage"
  2054. queryFmtOldestSample = `min_over_time(timestamp(group(node_cpu_hourly_cost{%s}))[%s:%s])`
  2055. queryFmtNewestSample = `max_over_time(timestamp(group(node_cpu_hourly_cost{%s}))[%s:%s])`
  2056. )
  2057. cfg := pds.promConfig
  2058. minutesPerDuration := 60
  2059. dur := time.Duration(limitDays) * timeutil.Day
  2060. end := time.Now().UTC().Truncate(timeutil.Day).Add(timeutil.Day)
  2061. start := end.Add(-dur)
  2062. durStr := pds.durationStringFor(start, end, minutesPerDuration, false)
  2063. ctx := pds.promContexts.NewNamedContext(AllocationContextName)
  2064. queryOldest := fmt.Sprintf(queryFmtOldestSample, cfg.ClusterFilter, durStr, "1h")
  2065. log.Debugf("[Prometheus][%s[Oldest]][At Time: %d]: %s", queryName, end.Unix(), queryOldest)
  2066. resOldestFut := ctx.QueryAtTime(queryOldest, end)
  2067. resOldest, err := resOldestFut.Await()
  2068. if err != nil {
  2069. return time.Time{}, time.Time{}, fmt.Errorf("querying oldest sample: %w", err)
  2070. }
  2071. if len(resOldest) == 0 || len(resOldest[0].Values) == 0 {
  2072. // If node_cpu_hourly_cost metric is not available, fallback to a reasonable time range
  2073. // This prevents CSV export from failing when the metric doesn't exist yet
  2074. log.Warnf("QueryDataCoverage: node_cpu_hourly_cost metric not available, using fallback time range")
  2075. // Use a reasonable fallback: start from 1 day ago to account for metric collection delay
  2076. fallbackEnd := time.Now().UTC().Truncate(timeutil.Day)
  2077. fallbackStart := fallbackEnd.AddDate(0, 0, -1) // 1 day ago
  2078. return fallbackStart, fallbackEnd, nil
  2079. }
  2080. oldest := time.Unix(int64(resOldest[0].Values[0].Value), 0)
  2081. queryNewest := fmt.Sprintf(queryFmtNewestSample, cfg.ClusterFilter, durStr, "1h")
  2082. log.Debugf("[Prometheus][%s[Newest]][At Time: %d]: %s", queryName, end.Unix(), queryNewest)
  2083. resNewestFut := ctx.QueryAtTime(queryNewest, end)
  2084. resNewest, err := resNewestFut.Await()
  2085. if err != nil {
  2086. return time.Time{}, time.Time{}, fmt.Errorf("querying newest sample: %w", err)
  2087. }
  2088. if len(resNewest) == 0 || len(resNewest[0].Values) == 0 {
  2089. // If newest query fails but oldest succeeded, use oldest as both start and end
  2090. // This allows CSV export to proceed with at least some time range
  2091. log.Warnf("QueryDataCoverage: newest sample query returned no results, using oldest timestamp")
  2092. return oldest, oldest, nil
  2093. }
  2094. newest := time.Unix(int64(resNewest[0].Values[0].Value), 0)
  2095. return oldest, newest, nil
  2096. }
  2097. // durationStringFor simplifies the determination of query duration based on the version of prom and if the function
  2098. // in the query needs all data points in the vector it is provided or if it will extrapolate its own. Functions
  2099. // that extrapolate will add on another resolution if given a duration that is one resolution longer than the intended
  2100. // duration.
  2101. func (pds *PrometheusMetricsQuerier) durationStringFor(start, end time.Time, minsPerResolution int, extrapolated bool) string {
  2102. dur := end.Sub(start)
  2103. // If using a version of Prometheus where the resolution needs duration offset,
  2104. // we need to apply that here.
  2105. //
  2106. // E.g. avg(node_total_hourly_cost{}) by (node, provider_id)[60m:5m] with
  2107. // time=01:00:00 will return, for a node running the entire time, 12
  2108. // timestamps where the first is 00:05:00 and the last is 01:00:00.
  2109. // However, OpenCost expects for there to be 13 timestamps where the first
  2110. // begins at 00:00:00. To achieve this, we must modify our query to
  2111. // avg(node_total_hourly_cost{}) by (node, provider_id)[65m:5m]
  2112. if pds.promConfig.IsOffsetResolution && !extrapolated {
  2113. // increase the query time by the resolution
  2114. dur = dur + (time.Duration(minsPerResolution) * time.Minute)
  2115. }
  2116. return timeutil.DurationString(dur)
  2117. }