package prom import ( "fmt" "time" "github.com/opencost/opencost/core/pkg/log" "github.com/opencost/opencost/core/pkg/source" "github.com/opencost/opencost/core/pkg/util/timeutil" prometheus "github.com/prometheus/client_golang/api" ) //-------------------------------------------------------------------------- // PrometheusMetricsQuerier //-------------------------------------------------------------------------- // PrometheusMetricsQueryLogFormat is the log format used to log metric queries before being sent to the prometheus // instance const PrometheusMetricsQueryLogFormat = `[PrometheusMetricsQuerier][%s][At Time: %d]: %s` // PrometheusMetricsQuerier is the implementation of the data source's MetricsQuerier interface for Prometheus. type PrometheusMetricsQuerier struct { promConfig *OpenCostPrometheusConfig promClient prometheus.Client promContexts *ContextFactory } func newPrometheusMetricsQuerier( promConfig *OpenCostPrometheusConfig, promClient prometheus.Client, promContexts *ContextFactory, ) *PrometheusMetricsQuerier { return &PrometheusMetricsQuerier{ promConfig: promConfig, promClient: promClient, promContexts: promContexts, } } func (pds *PrometheusMetricsQuerier) QueryPVPricePerGiBHour(start, end time.Time) *source.Future[source.PVPricePerGiBHourResult] { const queryName = "QueryPVPricePerGiBHour" const pvCostQuery = `avg(avg_over_time(pv_hourly_cost{%s}[%s])) by (%s, persistentvolume, volumename, uid, provider_id)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryPVCost := fmt.Sprintf(pvCostQuery, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryPVCost) ctx := pds.promContexts.NewNamedContext(ClusterContextName) return source.NewFuture(source.DecodePVPricePerGiBHourResult, ctx.QueryAtTime(queryPVCost, end)) } func (pds *PrometheusMetricsQuerier) QueryPVUsedAverage(start, end time.Time) *source.Future[source.PVUsedAvgResult] { const queryName = "QueryPVUsedAverage" const pvUsedAverageQuery = `avg(avg_over_time(kubelet_volume_stats_used_bytes{%s}[%s])) by (%s, persistentvolumeclaim, namespace, uid)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryPVUsedAvg := fmt.Sprintf(pvUsedAverageQuery, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryPVUsedAvg) ctx := pds.promContexts.NewNamedContext(ClusterContextName) return source.NewFuture(source.DecodePVUsedAvgResult, ctx.QueryAtTime(queryPVUsedAvg, end)) } func (pds *PrometheusMetricsQuerier) QueryPVUsedMax(start, end time.Time) *source.Future[source.PVUsedMaxResult] { const queryName = "QueryPVUsedMax" const pvUsedMaxQuery = `max(max_over_time(kubelet_volume_stats_used_bytes{%s}[%s])) by (%s, persistentvolumeclaim, namespace, uid)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryPVUsedMax := fmt.Sprintf(pvUsedMaxQuery, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryPVUsedMax) ctx := pds.promContexts.NewNamedContext(ClusterContextName) return source.NewFuture(source.DecodePVUsedMaxResult, ctx.QueryAtTime(queryPVUsedMax, end)) } func (pds *PrometheusMetricsQuerier) QueryPVCUptime(start, end time.Time) *source.Future[source.UptimeResult] { const queryName = "QueryPVCUptime" const queryFmtPVCUptime = `avg(kube_persistentvolumeclaim_info{%s}) by (%s, uid)[%s:%dm]` cfg := pds.promConfig minsPerResolution := cfg.DataResolutionMinutes durStr := pds.durationStringFor(start, end, minsPerResolution, false) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryPVCUptime := fmt.Sprintf(queryFmtPVCUptime, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryPVCUptime) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeUptimeResult, ctx.QueryAtTime(queryPVCUptime, end)) } func (pds *PrometheusMetricsQuerier) QueryPVCInfo(start, end time.Time) *source.Future[source.PVCInfoResult] { const queryName = "QueryPVCInfo" const queryFmtPVCInfo = `avg(kube_persistentvolumeclaim_info{volumename != "", %s}) by (persistentvolumeclaim, storageclass, volumename, namespace, uid, %s)[%s:%dm]` cfg := pds.promConfig minsPerResolution := cfg.DataResolutionMinutes durStr := pds.durationStringFor(start, end, minsPerResolution, false) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryPVCInfo := fmt.Sprintf(queryFmtPVCInfo, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryPVCInfo) ctx := pds.promContexts.NewNamedContext(AllocationContextName) return source.NewFuture(source.DecodePVCInfoResult, ctx.QueryAtTime(queryPVCInfo, end)) } func (pds *PrometheusMetricsQuerier) QueryKMPVCInfo(start, end time.Time) *source.Future[source.PVCInfoResult] { const queryName = "QueryKMPVCInfo" const queryFmt = `avg(avg_over_time(kube_persistentvolumeclaim_info{volumename != "", %s}[%s])) by (uid, namespace_uid, persistentvolumeclaim, namespace, storageclass, volumename, persistentvolume_uid, %s)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } q := fmt.Sprintf(queryFmt, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), q) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodePVCInfoResult, ctx.QueryAtTime(q, end)) } func (pds *PrometheusMetricsQuerier) QueryKMPVInfo(start, end time.Time) *source.Future[source.PVInfoResult] { const queryName = "QueryKMPVInfo" const queryFmt = `avg(avg_over_time(kubecost_pv_info{%s}[%s])) by (uid, storageclass, persistentvolume, provider_id, csi_volume_handle, %s)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } q := fmt.Sprintf(queryFmt, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), q) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodePVInfoResult, ctx.QueryAtTime(q, end)) } func (pds *PrometheusMetricsQuerier) QueryPVCBytesUsedAverage(start, end time.Time) *source.Future[source.PVCUIDValueResult] { const queryName = "QueryPVCBytesUsedAverage" const queryFmt = `avg(avg_over_time(kubelet_volume_stats_used_bytes{%s}[%s])) by (%s, persistentvolumeclaim_uid)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } q := fmt.Sprintf(queryFmt, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), q) ctx := pds.promContexts.NewNamedContext(ClusterContextName) return source.NewFuture(source.DecodePVCUIDValueResult, ctx.QueryAtTime(q, end)) } func (pds *PrometheusMetricsQuerier) QueryPVCBytesUsedMax(start, end time.Time) *source.Future[source.PVCUIDValueResult] { const queryName = "QueryPVCBytesUsedMax" const queryFmt = `max(max_over_time(kubelet_volume_stats_used_bytes{%s}[%s])) by (%s, persistentvolumeclaim_uid)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } q := fmt.Sprintf(queryFmt, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), q) ctx := pds.promContexts.NewNamedContext(ClusterContextName) return source.NewFuture(source.DecodePVCUIDValueResult, ctx.QueryAtTime(q, end)) } func (pds *PrometheusMetricsQuerier) QueryPVActiveMinutes(start, end time.Time) *source.Future[source.PVActiveMinutesResult] { const queryName = "QueryPVActiveMinutes" const pvActiveMinsQuery = `avg(kube_persistentvolume_capacity_bytes{%s}) by (%s, persistentvolume, uid)[%s:%dm]` cfg := pds.promConfig minsPerResolution := cfg.DataResolutionMinutes durStr := pds.durationStringFor(start, end, minsPerResolution, false) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryPVActiveMins := fmt.Sprintf(pvActiveMinsQuery, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryPVActiveMins) ctx := pds.promContexts.NewNamedContext(ClusterContextName) return source.NewFuture(source.DecodePVActiveMinutesResult, ctx.QueryAtTime(queryPVActiveMins, end)) } func (pds *PrometheusMetricsQuerier) QueryLocalStorageUsedAvg(start, end time.Time) *source.Future[source.LocalStorageUsedAvgResult] { const queryName = "QueryLocalStorageUsedAvg" 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)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryLocalStorageUsedAvg := fmt.Sprintf(localStorageUsedAvgQuery, cfg.ClusterFilter, durStr, cfg.ClusterLabel, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryLocalStorageUsedAvg) ctx := pds.promContexts.NewNamedContext(ClusterContextName) return source.NewFuture(source.DecodeLocalStorageUsedAvgResult, ctx.QueryAtTime(queryLocalStorageUsedAvg, end)) } func (pds *PrometheusMetricsQuerier) QueryLocalStorageUsedMax(start, end time.Time) *source.Future[source.LocalStorageUsedMaxResult] { const queryName = "QueryLocalStorageUsedMax" 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)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic("failed to parse duration string passed to QueryLocalStorageUsedMax") } queryLocalStorageUsedMax := fmt.Sprintf(localStorageUsedMaxQuery, cfg.ClusterFilter, durStr, cfg.ClusterLabel, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryLocalStorageUsedMax) ctx := pds.promContexts.NewNamedContext(ClusterContextName) return source.NewFuture(source.DecodeLocalStorageUsedMaxResult, ctx.QueryAtTime(queryLocalStorageUsedMax, end)) } func (pds *PrometheusMetricsQuerier) QueryLocalStorageBytes(start, end time.Time) *source.Future[source.LocalStorageBytesResult] { const queryName = "QueryLocalStorageBytes" const localStorageBytesQuery = `avg_over_time(sum(container_fs_limit_bytes{device=~"/dev/(nvme|sda).*", id="/", %s}) by (instance, device, %s)[%s:%dm])` cfg := pds.promConfig minsPerResolution := cfg.DataResolutionMinutes durStr := pds.durationStringFor(start, end, minsPerResolution, false) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryLocalStorageBytes := fmt.Sprintf(localStorageBytesQuery, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryLocalStorageBytes) ctx := pds.promContexts.NewNamedContext(ClusterContextName) return source.NewFuture(source.DecodeLocalStorageBytesResult, ctx.QueryAtTime(queryLocalStorageBytes, end)) } func (pds *PrometheusMetricsQuerier) QueryKMLocalStorageUsedAvg(start, end time.Time) *source.Future[source.NodeUIDValueResult] { const queryName = "QueryKMLocalStorageUsedAvg" const queryFmt = `avg(avg_over_time(container_fs_usage_bytes{%s}[%s])) by (node_uid, %s)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } q := fmt.Sprintf(queryFmt, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), q) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeNodeUIDValueResult, ctx.QueryAtTime(q, end)) } func (pds *PrometheusMetricsQuerier) QueryKMLocalStorageUsedMax(start, end time.Time) *source.Future[source.NodeUIDValueResult] { const queryName = "QueryKMLocalStorageUsedMax" const queryFmt = `max(max_over_time(container_fs_usage_bytes{%s}[%s])) by (node_uid, %s)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } q := fmt.Sprintf(queryFmt, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), q) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeNodeUIDValueResult, ctx.QueryAtTime(q, end)) } func (pds *PrometheusMetricsQuerier) QueryKMLocalStorageBytes(start, end time.Time) *source.Future[source.UIDValueResult] { const queryName = "QueryKMLocalStorageBytes" const queryFmt = `avg_over_time(node_fs_capacity_bytes{%s}[%s:%dm]) by (uid, %s)` cfg := pds.promConfig minsPerResolution := cfg.DataResolutionMinutes durStr := pds.durationStringFor(start, end, minsPerResolution, false) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } q := fmt.Sprintf(queryFmt, cfg.ClusterFilter, durStr, minsPerResolution, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), q) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeUIDValueResult, ctx.QueryAtTime(q, end)) } func (pds *PrometheusMetricsQuerier) QueryLocalStorageActiveMinutes(start, end time.Time) *source.Future[source.LocalStorageActiveMinutesResult] { const queryName = "QueryLocalStorageActiveMinutes" const localStorageActiveMinutesQuery = `count(node_total_hourly_cost{%s}) by (%s, node, uid, instance, provider_id)[%s:%dm]` cfg := pds.promConfig minsPerResolution := cfg.DataResolutionMinutes durStr := pds.durationStringFor(start, end, minsPerResolution, false) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryLocalStorageActiveMins := fmt.Sprintf(localStorageActiveMinutesQuery, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryLocalStorageActiveMins) ctx := pds.promContexts.NewNamedContext(ClusterContextName) return source.NewFuture(source.DecodeLocalStorageActiveMinutesResult, ctx.QueryAtTime(queryLocalStorageActiveMins, end)) } func (pds *PrometheusMetricsQuerier) QueryNodeInfo(start, end time.Time) *source.Future[source.NodeInfoResult] { const queryName = "QueryNodeInfo" const queryFmtNodeInfo = `avg(avg_over_time(node_info{%s}[%s])) by (%s, node, uid, provider_id, instance_type)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryNodeInfo := fmt.Sprintf(queryFmtNodeInfo, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNodeInfo) ctx := pds.promContexts.NewNamedContext(ClusterContextName) return source.NewFuture(source.DecodeNodeInfoResult, ctx.QueryAtTime(queryNodeInfo, end)) } func (pds *PrometheusMetricsQuerier) QueryNodeUptime(start, end time.Time) *source.Future[source.UptimeResult] { const queryName = "QueryNodeUptime" const queryFmtNodeUptime = `avg(node_info{%s}) by (%s, uid)[%s:%dm]` cfg := pds.promConfig minsPerResolution := cfg.DataResolutionMinutes durStr := pds.durationStringFor(start, end, minsPerResolution, false) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryNodeUptime := fmt.Sprintf(queryFmtNodeUptime, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNodeUptime) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeUptimeResult, ctx.QueryAtTime(queryNodeUptime, end)) } func (pds *PrometheusMetricsQuerier) QueryNodeCPUCoresCapacity(start, end time.Time) *source.Future[source.NodeCPUCoresCapacityResult] { const queryName = "QueryNodeCPUCoresCapacity" const nodeCPUCoresCapacityQuery = `avg(avg_over_time(kube_node_status_capacity_cpu_cores{%s}[%s])) by (%s, node, uid)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryNodeCPUCoresCapacity := fmt.Sprintf(nodeCPUCoresCapacityQuery, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNodeCPUCoresCapacity) ctx := pds.promContexts.NewNamedContext(ClusterContextName) return source.NewFuture(source.DecodeNodeCPUCoresCapacityResult, ctx.QueryAtTime(queryNodeCPUCoresCapacity, end)) } func (pds *PrometheusMetricsQuerier) QueryNodeCPUCoresAllocatable(start, end time.Time) *source.Future[source.NodeCPUCoresAllocatableResult] { const queryName = "QueryNodeCPUCoresAllocatable" const nodeCPUCoresAllocatableQuery = `avg(avg_over_time(kube_node_status_allocatable_cpu_cores{%s}[%s])) by (%s, node, uid)` // `avg(avg_over_time(container_cpu_allocation{container!="", container!="POD", node!="", %s}[%s])) by (container, pod, namespace, node, %s)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryNodeCPUCoresAllocatable := fmt.Sprintf(nodeCPUCoresAllocatableQuery, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNodeCPUCoresAllocatable) ctx := pds.promContexts.NewNamedContext(ClusterContextName) return source.NewFuture(source.DecodeNodeCPUCoresAllocatableResult, ctx.QueryAtTime(queryNodeCPUCoresAllocatable, end)) } func (pds *PrometheusMetricsQuerier) QueryNodeRAMBytesCapacity(start, end time.Time) *source.Future[source.NodeRAMBytesCapacityResult] { const queryName = "QueryNodeRAMBytesCapacity" const nodeRAMBytesCapacityQuery = `avg(avg_over_time(kube_node_status_capacity_memory_bytes{%s}[%s])) by (%s, node, uid)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryNodeRAMBytesCapacity := fmt.Sprintf(nodeRAMBytesCapacityQuery, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNodeRAMBytesCapacity) ctx := pds.promContexts.NewNamedContext(ClusterContextName) return source.NewFuture(source.DecodeNodeRAMBytesCapacityResult, ctx.QueryAtTime(queryNodeRAMBytesCapacity, end)) } func (pds *PrometheusMetricsQuerier) QueryNodeRAMBytesAllocatable(start, end time.Time) *source.Future[source.NodeRAMBytesAllocatableResult] { const queryName = "QueryNodeRAMBytesAllocatable" const nodeRAMBytesAllocatableQuery = `avg(avg_over_time(kube_node_status_allocatable_memory_bytes{%s}[%s])) by (%s, node, uid)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryNodeRAMBytesAllocatable := fmt.Sprintf(nodeRAMBytesAllocatableQuery, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNodeRAMBytesAllocatable) ctx := pds.promContexts.NewNamedContext(ClusterContextName) return source.NewFuture(source.DecodeNodeRAMBytesAllocatableResult, ctx.QueryAtTime(queryNodeRAMBytesAllocatable, end)) } func (pds *PrometheusMetricsQuerier) QueryNodeGPUCount(start, end time.Time) *source.Future[source.NodeGPUCountResult] { const queryName = "QueryNodeGPUCount" const nodeGPUCountQuery = `avg(avg_over_time(node_gpu_count{%s}[%s])) by (%s, node, uid, provider_id)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryNodeGPUCount := fmt.Sprintf(nodeGPUCountQuery, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNodeGPUCount) ctx := pds.promContexts.NewNamedContext(ClusterContextName) return source.NewFuture(source.DecodeNodeGPUCountResult, ctx.QueryAtTime(queryNodeGPUCount, end)) } func (pds *PrometheusMetricsQuerier) QueryNodeLabels(start, end time.Time) *source.Future[source.NodeLabelsResult] { const queryName = "QueryNodeLabels" const labelsQuery = `avg_over_time(kube_node_labels{%s}[%s])` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryLabels := fmt.Sprintf(labelsQuery, cfg.ClusterFilter, durStr) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryLabels) ctx := pds.promContexts.NewNamedContext(ClusterContextName) return source.NewFuture(source.DecodeNodeLabelsResult, ctx.QueryAtTime(queryLabels, end)) } func (pds *PrometheusMetricsQuerier) QueryNodeActiveMinutes(start, end time.Time) *source.Future[source.NodeActiveMinutesResult] { const queryName = "QueryNodeActiveMinutes" const activeMinsQuery = `avg(node_total_hourly_cost{%s}) by (node, uid, %s, provider_id)[%s:%dm]` cfg := pds.promConfig minsPerResolution := cfg.DataResolutionMinutes durStr := pds.durationStringFor(start, end, minsPerResolution, false) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryActiveMins := fmt.Sprintf(activeMinsQuery, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryActiveMins) ctx := pds.promContexts.NewNamedContext(ClusterContextName) return source.NewFuture(source.DecodeNodeActiveMinutesResult, ctx.QueryAtTime(queryActiveMins, end)) } func (pds *PrometheusMetricsQuerier) QueryNodeCPUModeTotal(start, end time.Time) *source.Future[source.NodeCPUModeTotalResult] { const queryName = "QueryNodeCPUModeTotal" const nodeCPUModeTotalQuery = `sum(rate(node_cpu_seconds_total{%s}[%s:%dm])) by (kubernetes_node, uid, %s, mode)` cfg := pds.promConfig minsPerResolution := cfg.DataResolutionMinutes durStr := pds.durationStringFor(start, end, minsPerResolution, true) if durStr == "" { panic("failed to parse duration string passed to QueryNodeCPUModeTotal") } queryCPUModeTotal := fmt.Sprintf(nodeCPUModeTotalQuery, cfg.ClusterFilter, durStr, minsPerResolution, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryCPUModeTotal) ctx := pds.promContexts.NewNamedContext(ClusterContextName) return source.NewFuture(source.DecodeNodeCPUModeTotalResult, ctx.QueryAtTime(queryCPUModeTotal, end)) } func (pds *PrometheusMetricsQuerier) QueryNodeRAMSystemPercent(start, end time.Time) *source.Future[source.NodeRAMSystemPercentResult] { const queryName = "QueryNodeRAMSystemPercent" 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)` cfg := pds.promConfig minsPerResolution := cfg.DataResolutionMinutes durStr := pds.durationStringFor(start, end, minsPerResolution, false) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryRAMSystemPct := fmt.Sprintf(nodeRAMSystemPctQuery, cfg.ClusterFilter, durStr, minsPerResolution, cfg.ClusterLabel, cfg.ClusterFilter, durStr, minsPerResolution, cfg.ClusterLabel, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryRAMSystemPct) ctx := pds.promContexts.NewNamedContext(ClusterContextName) return source.NewFuture(source.DecodeNodeRAMSystemPercentResult, ctx.QueryAtTime(queryRAMSystemPct, end)) } func (pds *PrometheusMetricsQuerier) QueryNodeRAMUserPercent(start, end time.Time) *source.Future[source.NodeRAMUserPercentResult] { const queryName = "QueryNodeRAMUserPercent" 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)` cfg := pds.promConfig minsPerResolution := cfg.DataResolutionMinutes durStr := pds.durationStringFor(start, end, minsPerResolution, false) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryRAMUserPct := fmt.Sprintf(nodeRAMUserPctQuery, cfg.ClusterFilter, durStr, minsPerResolution, cfg.ClusterLabel, cfg.ClusterFilter, durStr, minsPerResolution, cfg.ClusterLabel, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryRAMUserPct) ctx := pds.promContexts.NewNamedContext(ClusterContextName) return source.NewFuture(source.DecodeNodeRAMUserPercentResult, ctx.QueryAtTime(queryRAMUserPct, end)) } func (pds *PrometheusMetricsQuerier) QueryNodeResourceCapacities(start, end time.Time) *source.Future[source.ResourceResult] { const queryName = "QueryNodeResourceCapacities" const queryFmtNodeResourceCapacities = `avg(avg_over_time(kube_node_status_capacity{%s}[%s])) by (%s, node, uid, resource, unit)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryNodeResourceCapacities := fmt.Sprintf(queryFmtNodeResourceCapacities, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNodeResourceCapacities) ctx := pds.promContexts.NewNamedContext(ClusterContextName) return source.NewFuture(source.DecodeResourceResult, ctx.QueryAtTime(queryNodeResourceCapacities, end)) } func (pds *PrometheusMetricsQuerier) QueryNodeResourcesAllocatable(start, end time.Time) *source.Future[source.ResourceResult] { const queryName = "QueryNodeResourcesAllocatable" const queryFmtNodeResourcesAllocatable = `avg(avg_over_time(kube_node_status_allocatable{%s}[%s])) by (%s, node, uid, resource, unit)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryNodeResourcesAllocatable := fmt.Sprintf(queryFmtNodeResourcesAllocatable, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNodeResourcesAllocatable) ctx := pds.promContexts.NewNamedContext(ClusterContextName) return source.NewFuture(source.DecodeResourceResult, ctx.QueryAtTime(queryNodeResourcesAllocatable, end)) } func (pds *PrometheusMetricsQuerier) QueryLBPricePerHr(start, end time.Time) *source.Future[source.LBPricePerHrResult] { const queryName = "QueryLBPricePerHr" const queryFmtLBCostPerHr = `avg(avg_over_time(kubecost_load_balancer_cost{%s}[%s])) by (namespace, service_name, ingress_ip, uid, %s)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryLBCostPerHr := fmt.Sprintf(queryFmtLBCostPerHr, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryLBCostPerHr) ctx := pds.promContexts.NewNamedContext(AllocationContextName) return source.NewFuture(source.DecodeLBPricePerHrResult, ctx.QueryAtTime(queryLBCostPerHr, end)) } func (pds *PrometheusMetricsQuerier) QueryLBActiveMinutes(start, end time.Time) *source.Future[source.LBActiveMinutesResult] { const queryName = "QueryLBActiveMinutes" const lbActiveMinutesQuery = `avg(kubecost_load_balancer_cost{%s}) by (namespace, service_name, uid, %s, ingress_ip)[%s:%dm]` cfg := pds.promConfig minsPerResolution := cfg.DataResolutionMinutes durStr := pds.durationStringFor(start, end, minsPerResolution, false) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryLBActiveMins := fmt.Sprintf(lbActiveMinutesQuery, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryLBActiveMins) ctx := pds.promContexts.NewNamedContext(ClusterContextName) return source.NewFuture(source.DecodeLBActiveMinutesResult, ctx.QueryAtTime(queryLBActiveMins, end)) } func (pds *PrometheusMetricsQuerier) QueryClusterInfo(start, end time.Time) *source.Future[source.ClusterInfoResult] { const queryName = "QueryClusterInfo" const queryFmtClusterInfo = `avg(avg_over_time(cluster_info{%s}[%s])) by (%s, uid, provider, account_id, provisioner_name, region)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryClusterInfo := fmt.Sprintf(queryFmtClusterInfo, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryClusterInfo) ctx := pds.promContexts.NewNamedContext(ClusterContextName) return source.NewFuture(source.DecodeClusterInfoResult, ctx.QueryAtTime(queryClusterInfo, end)) } func (pds *PrometheusMetricsQuerier) QueryClusterUptime(start, end time.Time) *source.Future[source.UptimeResult] { const queryName = "QueryClusterUptime" const queryFmtClusterUptime = `avg(cluster_info{%s}) by (%s, uid)[%s:%dm]` cfg := pds.promConfig minsPerResolution := cfg.DataResolutionMinutes durStr := pds.durationStringFor(start, end, minsPerResolution, false) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryClusterUptime := fmt.Sprintf(queryFmtClusterUptime, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryFmtClusterUptime) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeUptimeResult, ctx.QueryAtTime(queryClusterUptime, end)) } func (pds *PrometheusMetricsQuerier) QueryClusterManagementDuration(start, end time.Time) *source.Future[source.ClusterManagementDurationResult] { const queryName = "QueryClusterManagementDuration" const clusterManagementDurationQuery = `avg(kubecost_cluster_management_cost{%s}) by (%s, provisioner_name)[%s:%dm]` cfg := pds.promConfig minsPerResolution := cfg.DataResolutionMinutes durStr := pds.durationStringFor(start, end, minsPerResolution, false) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryClusterManagementDuration := fmt.Sprintf(clusterManagementDurationQuery, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryClusterManagementDuration) ctx := pds.promContexts.NewNamedContext(ClusterContextName) return source.NewFuture(source.DecodeClusterManagementDurationResult, ctx.QueryAtTime(queryClusterManagementDuration, end)) } func (pds *PrometheusMetricsQuerier) QueryClusterManagementPricePerHr(start, end time.Time) *source.Future[source.ClusterManagementPricePerHrResult] { const queryName = "QueryClusterManagementPricePerHr" const clusterManagementCostQuery = `avg(avg_over_time(kubecost_cluster_management_cost{%s}[%s])) by (%s, provisioner_name)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryClusterManagementCost := fmt.Sprintf(clusterManagementCostQuery, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryClusterManagementCost) ctx := pds.promContexts.NewNamedContext(ClusterContextName) return source.NewFuture(source.DecodeClusterManagementPricePerHrResult, ctx.QueryAtTime(queryClusterManagementCost, end)) } // AllocationMetricQuerier func (pds *PrometheusMetricsQuerier) QueryPods(start, end time.Time) *source.Future[source.PodsResult] { const queryName = "QueryPods" const queryFmtPods = `avg(kube_pod_container_status_running{%s} != 0) by (pod, namespace, uid, %s)[%s:%dm]` cfg := pds.promConfig minsPerResolution := cfg.DataResolutionMinutes durStr := pds.durationStringFor(start, end, minsPerResolution, false) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryPods := fmt.Sprintf(queryFmtPods, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryPods) ctx := pds.promContexts.NewNamedContext(AllocationContextName) return source.NewFuture(source.DecodePodsResult, ctx.QueryAtTime(queryPods, end)) } func (pds *PrometheusMetricsQuerier) QueryPodsUID(start, end time.Time) *source.Future[source.PodsResult] { const queryName = "QueryPodsUID" const queryFmtPodsUID = `avg(kube_pod_container_status_running{%s} != 0) by (pod, namespace, uid, %s)[%s:%dm]` cfg := pds.promConfig minsPerResolution := cfg.DataResolutionMinutes durStr := pds.durationStringFor(start, end, minsPerResolution, false) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryPodsUID := fmt.Sprintf(queryFmtPodsUID, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryPodsUID) ctx := pds.promContexts.NewNamedContext(AllocationContextName) return source.NewFuture(source.DecodePodsResult, ctx.QueryAtTime(queryPodsUID, end)) } func (pds *PrometheusMetricsQuerier) QueryPodInfo(start, end time.Time) *source.Future[source.PodInfoResult] { const queryName = "QueryPodInfo" const queryFmtPodInfo = `avg(avg_over_time(pod_info{%s}[%s])) by (%s, pod, uid, namespace_uid, node_uid)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryPodInfo := fmt.Sprintf(queryFmtPodInfo, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryPodInfo) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodePodInfoResult, ctx.QueryAtTime(queryPodInfo, end)) } func (pds *PrometheusMetricsQuerier) QueryPodUptime(start, end time.Time) *source.Future[source.UptimeResult] { const queryName = "QueryPodUptime" const queryFmtPodUptime = `avg(pod_info{%s}) by (%s, uid)[%s:%dm]` cfg := pds.promConfig minsPerResolution := cfg.DataResolutionMinutes durStr := pds.durationStringFor(start, end, minsPerResolution, false) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryPodUptime := fmt.Sprintf(queryFmtPodUptime, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryPodUptime) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeUptimeResult, ctx.QueryAtTime(queryPodUptime, end)) } func (pds *PrometheusMetricsQuerier) QueryPodOwners(start, end time.Time) *source.Future[source.OwnerResult] { const queryName = "QueryPodOwners" const queryFmtPodOwners = `avg(avg_over_time(kube_pod_owner{%s}[%s])) by (%s, uid, owner_uid, owner_kind, controller)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryPodOwners := fmt.Sprintf(queryFmtPodOwners, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryPodOwners) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeOwnerResult, ctx.QueryAtTime(queryPodOwners, end)) } func (pds *PrometheusMetricsQuerier) QueryPodPVCVolumes(start, end time.Time) *source.Future[source.PodPVCVolumeResult] { const queryName = "QueryPodPVCVolumes" const queryFmtPodPVCVolumes = `avg(avg_over_time(pod_pvc_volume{%s}[%s])) by (%s, uid, persistentvolumeclaim_uid, pod_volume_name)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryPodPVCVolumes := fmt.Sprintf(queryFmtPodPVCVolumes, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryPodPVCVolumes) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodePodPVCVolumeResult, ctx.QueryAtTime(queryPodPVCVolumes, end)) } func (pds *PrometheusMetricsQuerier) QueryPodNetworkEgressBytes(start, end time.Time) *source.Future[source.PodNetworkBytesResult] { const queryName = "QueryPodNetworkEgressBytes" 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)` cfg := pds.promConfig minsPerResolution := cfg.DataResolutionMinutes durStr := pds.durationStringFor(start, end, minsPerResolution, true) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } q := fmt.Sprintf(queryFmt, cfg.ClusterFilter, durStr, minsPerResolution, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), q) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodePodNetworkBytesResult, ctx.QueryAtTime(q, end)) } func (pds *PrometheusMetricsQuerier) QueryPodNetworkIngressBytes(start, end time.Time) *source.Future[source.PodNetworkBytesResult] { const queryName = "QueryPodNetworkIngressBytes" 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)` cfg := pds.promConfig minsPerResolution := cfg.DataResolutionMinutes durStr := pds.durationStringFor(start, end, minsPerResolution, true) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } q := fmt.Sprintf(queryFmt, cfg.ClusterFilter, durStr, minsPerResolution, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), q) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodePodNetworkBytesResult, ctx.QueryAtTime(q, end)) } func (pds *PrometheusMetricsQuerier) QueryContainerUptime(start, end time.Time) *source.Future[source.ContainerUptimeResult] { const queryName = "QueryContainerUptime" const queryFmtContainerUptime = `avg(kube_pod_container_status_running{container!="", %s} != 0) by (container, uid, %s)[%s:%dm]` cfg := pds.promConfig minsPerResolution := cfg.DataResolutionMinutes durStr := pds.durationStringFor(start, end, minsPerResolution, false) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryContainerUptime := fmt.Sprintf(queryFmtContainerUptime, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryContainerUptime) ctx := pds.promContexts.NewNamedContext(AllocationContextName) return source.NewFuture(source.DecodeContainerUptimeResult, ctx.QueryAtTime(queryContainerUptime, end)) } func (pds *PrometheusMetricsQuerier) QueryContainerResourceRequests(start, end time.Time) *source.Future[source.ContainerResourceResult] { const queryName = "QueryContainerResourceRequests" const queryFmtContainerResourceRequests = `avg(avg_over_time(kube_pod_container_resource_requests{container!="", container!="POD", node!="", %s}[%s])) by (container, uid, resource, unit, %s)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryContainerResourceRequests := fmt.Sprintf(queryFmtContainerResourceRequests, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryContainerResourceRequests) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeContainerResourceResult, ctx.QueryAtTime(queryContainerResourceRequests, end)) } func (pds *PrometheusMetricsQuerier) QueryContainerResourceLimits(start, end time.Time) *source.Future[source.ContainerResourceResult] { const queryName = "QueryContainerResourceLimits" const queryFmtContainerResourceLimits = `avg(avg_over_time(kube_pod_container_resource_limits{container!="", container!="POD", node!="", %s}[%s])) by (container, uid, resource, unit, %s)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryContainerResourceLimits := fmt.Sprintf(queryFmtContainerResourceLimits, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryContainerResourceLimits) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeContainerResourceResult, ctx.QueryAtTime(queryContainerResourceLimits, end)) } func (pds *PrometheusMetricsQuerier) QueryRAMBytesAllocated(start, end time.Time) *source.Future[source.RAMBytesAllocatedResult] { const queryName = "QueryRAMBytesAllocated" 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)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryRAMBytesAllocated := fmt.Sprintf(queryFmtRAMBytesAllocated, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryRAMBytesAllocated) ctx := pds.promContexts.NewNamedContext(AllocationContextName) return source.NewFuture(source.DecodeRAMBytesAllocatedResult, ctx.QueryAtTime(queryRAMBytesAllocated, end)) } func (pds *PrometheusMetricsQuerier) QueryRAMRequests(start, end time.Time) *source.Future[source.RAMRequestsResult] { const queryName = "QueryRAMRequests" 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)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryRAMRequests := fmt.Sprintf(queryFmtRAMRequests, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryRAMRequests) ctx := pds.promContexts.NewNamedContext(AllocationContextName) return source.NewFuture(source.DecodeRAMRequestsResult, ctx.QueryAtTime(queryRAMRequests, end)) } func (pds *PrometheusMetricsQuerier) QueryRAMLimits(start, end time.Time) *source.Future[source.RAMLimitsResult] { const queryName = "QueryRAMLimits" 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)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryRAMLimits := fmt.Sprintf(queryFmtRAMLimits, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryRAMLimits) ctx := pds.promContexts.NewNamedContext(AllocationContextName) return source.NewFuture(source.DecodeRAMLimitsResult, ctx.QueryAtTime(queryRAMLimits, end)) } func (pds *PrometheusMetricsQuerier) QueryRAMUsageAvg(start, end time.Time) *source.Future[source.RAMUsageAvgResult] { const queryName = "QueryRAMUsageAvg" 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)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryRAMUsageAvg := fmt.Sprintf(queryFmtRAMUsageAvg, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryRAMUsageAvg) ctx := pds.promContexts.NewNamedContext(AllocationContextName) return source.NewFuture(source.DecodeRAMUsageAvgResult, ctx.QueryAtTime(queryRAMUsageAvg, end)) } func (pds *PrometheusMetricsQuerier) QueryRAMUsageMax(start, end time.Time) *source.Future[source.RAMUsageMaxResult] { const queryName = "QueryRAMUsageMax" 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)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryRAMUsageMax := fmt.Sprintf(queryFmtRAMUsageMax, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryRAMUsageMax) ctx := pds.promContexts.NewNamedContext(AllocationContextName) return source.NewFuture(source.DecodeRAMUsageMaxResult, ctx.QueryAtTime(queryRAMUsageMax, end)) } func (pds *PrometheusMetricsQuerier) QueryCPUCoresAllocated(start, end time.Time) *source.Future[source.CPUCoresAllocatedResult] { const queryName = "QueryCPUCoresAllocated" const queryFmtCPUCoresAllocated = `avg(avg_over_time(container_cpu_allocation{container!="", container!="POD", node!="", %s}[%s])) by (container, pod, namespace, node, uid, %s)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryCPUCoresAllocated := fmt.Sprintf(queryFmtCPUCoresAllocated, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryCPUCoresAllocated) ctx := pds.promContexts.NewNamedContext(AllocationContextName) return source.NewFuture(source.DecodeCPUCoresAllocatedResult, ctx.QueryAtTime(queryCPUCoresAllocated, end)) } func (pds *PrometheusMetricsQuerier) QueryCPURequests(start, end time.Time) *source.Future[source.CPURequestsResult] { const queryName = "QueryCPURequests" 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)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryCPURequests := fmt.Sprintf(queryFmtCPURequests, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryCPURequests) ctx := pds.promContexts.NewNamedContext(AllocationContextName) return source.NewFuture(source.DecodeCPURequestsResult, ctx.QueryAtTime(queryCPURequests, end)) } func (pds *PrometheusMetricsQuerier) QueryCPULimits(start, end time.Time) *source.Future[source.CPULimitsResult] { const queryName = "QueryCPULimits" 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)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryCPULimits := fmt.Sprintf(queryFmtCPULimits, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryCPULimits) ctx := pds.promContexts.NewNamedContext(AllocationContextName) return source.NewFuture(source.DecodeCPULimitsResult, ctx.QueryAtTime(queryCPULimits, end)) } func (pds *PrometheusMetricsQuerier) QueryCPUUsageAvg(start, end time.Time) *source.Future[source.CPUUsageAvgResult] { const queryName = "QueryCPUUsageAvg" 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)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryCPUUsageAvg := fmt.Sprintf(queryFmtCPUUsageAvg, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryCPUUsageAvg) ctx := pds.promContexts.NewNamedContext(AllocationContextName) return source.NewFuture(source.DecodeCPUUsageAvgResult, ctx.QueryAtTime(queryCPUUsageAvg, end)) } func (pds *PrometheusMetricsQuerier) QueryCPUUsageMax(start, end time.Time) *source.Future[source.CPUUsageMaxResult] { const queryName = "QueryCPUUsageMax" // Because we use container_cpu_usage_seconds_total to calculate CPU usage // at any given "instant" of time, we need to use an irate or rate. To then // calculate a max (or any aggregation) we have to perform an aggregation // query on top of an instant-by-instant maximum. Prometheus supports this // type of query with a "subquery" [1], however it is reportedly expensive // to make such a query. By default, Kubecost's Prometheus config includes // a recording rule that keeps track of the instant-by-instant irate for CPU // usage. The metric in this query is created by that recording rule. // // [1] https://prometheus.io/blog/2019/01/28/subquery-support/ // // If changing the name of the recording rule, make sure to update the // corresponding diagnostic query to avoid confusion. 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)` // This is the subquery equivalent of the above recording rule query. It is // more expensive, but does not require the recording rule. It should be // used as a fallback query if the recording rule data does not exist. // // The parameter after the colon [:] in the subquery affects the // resolution of the subquery. // The parameter after the metric ...{}[] should be set to 2x // the resolution, to make sure the irate always has two points to query // in case the Prom scrape duration has been reduced to be equal to the // query resolution. 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)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryCPUUsageMaxRecordingRule := fmt.Sprintf(queryFmtCPUUsageMaxRecordingRule, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryCPUUsageMaxRecordingRule) ctx := pds.promContexts.NewNamedContext(AllocationContextName) resCPUUsageMaxRR := ctx.QueryAtTime(queryCPUUsageMaxRecordingRule, end) resCPUUsageMax, _ := resCPUUsageMaxRR.Await() if len(resCPUUsageMax) > 0 { return source.NewFutureFrom(source.DecodeAll(resCPUUsageMax, source.DecodeCPUUsageMaxResult)) } minsPerResolution := cfg.DataResolutionMinutes durStr = pds.durationStringFor(start, end, minsPerResolution, false) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryCPUUsageMaxSubquery := fmt.Sprintf(queryFmtCPUUsageMaxSubquery, cfg.ClusterFilter, 2*minsPerResolution, durStr, minsPerResolution, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryCPUUsageMaxSubquery) return source.NewFuture(source.DecodeCPUUsageMaxResult, ctx.QueryAtTime(queryCPUUsageMaxSubquery, end)) } func (pds *PrometheusMetricsQuerier) QueryGPUsRequested(start, end time.Time) *source.Future[source.GPUsRequestedResult] { const queryName = "QueryGPUsRequested" 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)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryGPUsRequested := fmt.Sprintf(queryFmtGPUsRequested, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryGPUsRequested) ctx := pds.promContexts.NewNamedContext(AllocationContextName) return source.NewFuture(source.DecodeGPUsRequestedResult, ctx.QueryAtTime(queryGPUsRequested, end)) } func (pds *PrometheusMetricsQuerier) QueryGPUsUsageAvg(start, end time.Time) *source.Future[source.GPUsUsageAvgResult] { const queryName = "QueryGPUsUsageAvg" const queryFmtGPUsUsageAvg = `avg(avg_over_time(DCGM_FI_PROF_GR_ENGINE_ACTIVE{container!=""}[%s])) by (container, pod, namespace, pod_uid, %s)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryGPUsUsageAvg := fmt.Sprintf(queryFmtGPUsUsageAvg, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryGPUsUsageAvg) ctx := pds.promContexts.NewNamedContext(AllocationContextName) return source.NewFuture(source.DecodeGPUsUsageAvgResult, ctx.QueryAtTime(queryGPUsUsageAvg, end)) } func (pds *PrometheusMetricsQuerier) QueryGPUsUsageMax(start, end time.Time) *source.Future[source.GPUsUsageMaxResult] { const queryName = "QueryGPUsUsageMax" const queryFmtGPUsUsageMax = `max(max_over_time(DCGM_FI_PROF_GR_ENGINE_ACTIVE{container!=""}[%s])) by (container, pod, namespace, pod_uid, %s)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryGPUsUsageMax := fmt.Sprintf(queryFmtGPUsUsageMax, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryGPUsUsageMax) ctx := pds.promContexts.NewNamedContext(AllocationContextName) return source.NewFuture(source.DecodeGPUsUsageMaxResult, ctx.QueryAtTime(queryGPUsUsageMax, end)) } func (pds *PrometheusMetricsQuerier) QueryGPUsAllocated(start, end time.Time) *source.Future[source.GPUsAllocatedResult] { const queryName = "QueryGPUsAllocated" const queryFmtGPUsAllocated = `avg(avg_over_time(container_gpu_allocation{container!="", container!="POD", node!="", %s}[%s])) by (container, pod, namespace, node, uid, %s)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryGPUsAllocated := fmt.Sprintf(queryFmtGPUsAllocated, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryGPUsAllocated) ctx := pds.promContexts.NewNamedContext(AllocationContextName) return source.NewFuture(source.DecodeGPUsAllocatedResult, ctx.QueryAtTime(queryGPUsAllocated, end)) } func (pds *PrometheusMetricsQuerier) QueryIsGPUShared(start, end time.Time) *source.Future[source.IsGPUSharedResult] { const queryName = "QueryIsGPUShared" 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)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryIsGPUShared := fmt.Sprintf(queryFmtIsGPUShared, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryIsGPUShared) ctx := pds.promContexts.NewNamedContext(AllocationContextName) return source.NewFuture(source.DecodeIsGPUSharedResult, ctx.QueryAtTime(queryIsGPUShared, end)) } func (pds *PrometheusMetricsQuerier) QueryDCGMDeviceInfo(start, end time.Time) *source.Future[source.DCGMDeviceInfoResult] { const queryName = "QueryDCGMDeviceInfo" const queryFmtDCGMDeviceInfo = `avg(avg_over_time(DCGM_FI_DEV_DEC_UTIL{%s}[%s])) by (UUID, device, modelName, Hostname, %s)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryDCGMDeviceInfo := fmt.Sprintf(queryFmtDCGMDeviceInfo, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryDCGMDeviceInfo) ctx := pds.promContexts.NewNamedContext(ComputeCostDataContextName) return source.NewFuture(source.DecodeDCGMDeviceInfoResult, ctx.QueryAtTime(queryDCGMDeviceInfo, end)) } func (pds *PrometheusMetricsQuerier) QueryDCGMDeviceUptime(start, end time.Time) *source.Future[source.DCGMDeviceUptimeResult] { const queryName = "QueryDCGMDeviceUptime" const queryFmtDCGMDeviceUptime = `avg(DCGM_FI_DEV_DEC_UTIL{%s}) by (UUID, %s)[%s:%dm]` cfg := pds.promConfig minsPerResolution := cfg.DataResolutionMinutes durStr := pds.durationStringFor(start, end, minsPerResolution, false) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryDCGMDeviceUptime := fmt.Sprintf(queryFmtDCGMDeviceUptime, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryDCGMDeviceUptime) ctx := pds.promContexts.NewNamedContext(ComputeCostDataContextName) return source.NewFuture(source.DecodeDCGMDeviceUptimeResult, ctx.QueryAtTime(queryDCGMDeviceUptime, end)) } func (pds *PrometheusMetricsQuerier) QueryDCGMContainerUsageAvg(start, end time.Time) *source.Future[source.DCGMDeviceContainerUsageResult] { const queryName = "QueryDCGMContainerUsageAvg" const queryFmtDCGMContainerUsageAvg = `avg(avg_over_time(DCGM_FI_PROF_GR_ENGINE_ACTIVE{container!="", %s}[%s])) by (UUID, pod_uid, container, %s)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryDCGMContainerUsageAvg := fmt.Sprintf(queryFmtDCGMContainerUsageAvg, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryDCGMContainerUsageAvg) ctx := pds.promContexts.NewNamedContext(ComputeCostDataContextName) return source.NewFuture(source.DecodeDCGMDeviceContainerUsageResult, ctx.QueryAtTime(queryDCGMContainerUsageAvg, end)) } func (pds *PrometheusMetricsQuerier) QueryDCGMContainerUsageMax(start, end time.Time) *source.Future[source.DCGMDeviceContainerUsageResult] { const queryName = "QueryDCGMContainerUsageMax" const queryFmtDCGMContainerUsageMax = `max(max_over_time(DCGM_FI_PROF_GR_ENGINE_ACTIVE{container!="", %s}[%s])) by (UUID, pod_uid, container, %s)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryDCGMContainerUsageMax := fmt.Sprintf(queryFmtDCGMContainerUsageMax, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryDCGMContainerUsageMax) ctx := pds.promContexts.NewNamedContext(ComputeCostDataContextName) return source.NewFuture(source.DecodeDCGMDeviceContainerUsageResult, ctx.QueryAtTime(queryDCGMContainerUsageMax, end)) } func (pds *PrometheusMetricsQuerier) QueryGPUInfo(start, end time.Time) *source.Future[source.GPUInfoResult] { const queryName = "QueryGPUInfo" const queryFmtGetGPUInfo = `avg(avg_over_time(DCGM_FI_DEV_DEC_UTIL{container!="",%s}[%s])) by (container, pod, namespace, device, modelName, UUID, pod_uid, %s)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryGetGPUInfo := fmt.Sprintf(queryFmtGetGPUInfo, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryGetGPUInfo) ctx := pds.promContexts.NewNamedContext(AllocationContextName) return source.NewFuture(source.DecodeGPUInfoResult, ctx.QueryAtTime(queryGetGPUInfo, end)) } func (pds *PrometheusMetricsQuerier) QueryNodeCPUPricePerHr(start, end time.Time) *source.Future[source.NodeCPUPricePerHrResult] { const queryName = "QueryNodeCPUPricePerHr" const queryFmtNodeCostPerCPUHr = `avg(avg_over_time(node_cpu_hourly_cost{%s}[%s])) by (node, uid, %s, instance_type, provider_id)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryNodeCostPerCPUHr := fmt.Sprintf(queryFmtNodeCostPerCPUHr, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNodeCostPerCPUHr) ctx := pds.promContexts.NewNamedContext(AllocationContextName) return source.NewFuture(source.DecodeNodeCPUPricePerHrResult, ctx.QueryAtTime(queryNodeCostPerCPUHr, end)) } func (pds *PrometheusMetricsQuerier) QueryNodeRAMPricePerGiBHr(start, end time.Time) *source.Future[source.NodeRAMPricePerGiBHrResult] { const queryName = "QueryNodeRAMPricePerGiBHr" const queryFmtNodeCostPerRAMGiBHr = `avg(avg_over_time(node_ram_hourly_cost{%s}[%s])) by (node, uid, %s, instance_type, provider_id)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryNodeCostPerRAMGiBHr := fmt.Sprintf(queryFmtNodeCostPerRAMGiBHr, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNodeCostPerRAMGiBHr) ctx := pds.promContexts.NewNamedContext(AllocationContextName) return source.NewFuture(source.DecodeNodeRAMPricePerGiBHrResult, ctx.QueryAtTime(queryNodeCostPerRAMGiBHr, end)) } func (pds *PrometheusMetricsQuerier) QueryNodeGPUPricePerHr(start, end time.Time) *source.Future[source.NodeGPUPricePerHrResult] { const queryName = "QueryNodeGPUPricePerHr" const queryFmtNodeCostPerGPUHr = `avg(avg_over_time(node_gpu_hourly_cost{%s}[%s])) by (node, uid, %s, instance_type, provider_id)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryNodeCostPerGPUHr := fmt.Sprintf(queryFmtNodeCostPerGPUHr, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNodeCostPerGPUHr) ctx := pds.promContexts.NewNamedContext(AllocationContextName) return source.NewFuture(source.DecodeNodeGPUPricePerHrResult, ctx.QueryAtTime(queryNodeCostPerGPUHr, end)) } func (pds *PrometheusMetricsQuerier) QueryNodeIsSpot(start, end time.Time) *source.Future[source.NodeIsSpotResult] { const queryName = "QueryNodeIsSpot" const queryFmtNodeIsSpot = `avg_over_time(kubecost_node_is_spot{%s}[%s])` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryNodeIsSpot := fmt.Sprintf(queryFmtNodeIsSpot, cfg.ClusterFilter, durStr) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNodeIsSpot) ctx := pds.promContexts.NewNamedContext(AllocationContextName) return source.NewFuture(source.DecodeNodeIsSpotResult, ctx.QueryAtTime(queryNodeIsSpot, end)) } func (pds *PrometheusMetricsQuerier) QueryPodPVCAllocation(start, end time.Time) *source.Future[source.PodPVCAllocationResult] { const queryName = "QueryPodPVCAllocation" const queryFmtPodPVCAllocation = `avg(avg_over_time(pod_pvc_allocation{%s}[%s])) by (persistentvolume, persistentvolumeclaim, pod, namespace, uid, %s)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryPodPVCAllocation := fmt.Sprintf(queryFmtPodPVCAllocation, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryPodPVCAllocation) ctx := pds.promContexts.NewNamedContext(AllocationContextName) return source.NewFuture(source.DecodePodPVCAllocationResult, ctx.QueryAtTime(queryPodPVCAllocation, end)) } func (pds *PrometheusMetricsQuerier) QueryPVCBytesRequested(start, end time.Time) *source.Future[source.PVCBytesRequestedResult] { const queryName = "QueryPVCBytesRequested" const queryFmtPVCBytesRequested = `avg(avg_over_time(kube_persistentvolumeclaim_resource_requests_storage_bytes{%s}[%s])) by (persistentvolumeclaim, namespace, uid, %s)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryPVCBytesRequested := fmt.Sprintf(queryFmtPVCBytesRequested, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryPVCBytesRequested) ctx := pds.promContexts.NewNamedContext(AllocationContextName) return source.NewFuture(source.DecodePVCBytesRequestedResult, ctx.QueryAtTime(queryPVCBytesRequested, end)) } func (pds *PrometheusMetricsQuerier) QueryPVBytes(start, end time.Time) *source.Future[source.PVBytesResult] { const queryName = "QueryPVBytes" const queryFmtPVBytes = `avg(avg_over_time(kube_persistentvolume_capacity_bytes{%s}[%s])) by (persistentvolume, uid, %s)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryPVBytes := fmt.Sprintf(queryFmtPVBytes, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryPVBytes) ctx := pds.promContexts.NewNamedContext(AllocationContextName) return source.NewFuture(source.DecodePVBytesResult, ctx.QueryAtTime(queryPVBytes, end)) } func (pds *PrometheusMetricsQuerier) QueryPVInfo(start, end time.Time) *source.Future[source.PVInfoResult] { const queryName = "QueryPVInfo" const queryFmtPVMeta = `avg(avg_over_time(kubecost_pv_info{%s}[%s])) by (%s, storageclass, persistentvolume, uid, provider_id)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryPVMeta := fmt.Sprintf(queryFmtPVMeta, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryPVMeta) ctx := pds.promContexts.NewNamedContext(AllocationContextName) return source.NewFuture(source.DecodePVInfoResult, ctx.QueryAtTime(queryPVMeta, end)) } func (pds *PrometheusMetricsQuerier) QueryPVUptime(start, end time.Time) *source.Future[source.UptimeResult] { const queryName = "QueryPVUptime" const queryFmtPVUptime = `avg(kubecost_pv_info{%s}) by (%s, uid)[%s:%dm]` cfg := pds.promConfig minsPerResolution := cfg.DataResolutionMinutes durStr := pds.durationStringFor(start, end, minsPerResolution, false) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryPVUptime := fmt.Sprintf(queryFmtPVUptime, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryPVUptime) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeUptimeResult, ctx.QueryAtTime(queryPVUptime, end)) } func (pds *PrometheusMetricsQuerier) QueryNetZoneGiB(start, end time.Time) *source.Future[source.NetZoneGiBResult] { const queryName = "QueryNetZoneGiB" 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` cfg := pds.promConfig minsPerResolution := cfg.DataResolutionMinutes durStr := pds.durationStringFor(start, end, minsPerResolution, true) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryNetZoneGiB := fmt.Sprintf(queryFmtNetZoneGiB, cfg.ClusterFilter, durStr, minsPerResolution, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNetZoneGiB) ctx := pds.promContexts.NewNamedContext(AllocationContextName) return source.NewFuture(source.DecodeNetZoneGiBResult, ctx.QueryAtTime(queryNetZoneGiB, end)) } func (pds *PrometheusMetricsQuerier) QueryNetZonePricePerGiB(start, end time.Time) *source.Future[source.NetZonePricePerGiBResult] { const queryName = "QueryNetZonePricePerGiB" const queryFmtNetZoneCostPerGiB = `avg(avg_over_time(kubecost_network_zone_egress_cost{%s}[%s])) by (%s)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryNetZoneCostPerGiB := fmt.Sprintf(queryFmtNetZoneCostPerGiB, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNetZoneCostPerGiB) ctx := pds.promContexts.NewNamedContext(AllocationContextName) return source.NewFuture(source.DecodeNetZonePricePerGiBResult, ctx.QueryAtTime(queryNetZoneCostPerGiB, end)) } func (pds *PrometheusMetricsQuerier) QueryNetRegionGiB(start, end time.Time) *source.Future[source.NetRegionGiBResult] { const queryName = "QueryNetRegionGiB" 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` cfg := pds.promConfig minsPerResolution := cfg.DataResolutionMinutes durStr := pds.durationStringFor(start, end, minsPerResolution, true) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryNetRegionGiB := fmt.Sprintf(queryFmtNetRegionGiB, cfg.ClusterFilter, durStr, minsPerResolution, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNetRegionGiB) ctx := pds.promContexts.NewNamedContext(AllocationContextName) return source.NewFuture(source.DecodeNetRegionGiBResult, ctx.QueryAtTime(queryNetRegionGiB, end)) } func (pds *PrometheusMetricsQuerier) QueryNetRegionPricePerGiB(start, end time.Time) *source.Future[source.NetRegionPricePerGiBResult] { const queryName = "QueryNetRegionPricePerGiB" const queryFmtNetRegionCostPerGiB = `avg(avg_over_time(kubecost_network_region_egress_cost{%s}[%s])) by (%s)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic("failed to parse duration string passed to QueryNetRegionPricePerGiB") } queryNetRegionCostPerGiB := fmt.Sprintf(queryFmtNetRegionCostPerGiB, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNetRegionCostPerGiB) ctx := pds.promContexts.NewNamedContext(AllocationContextName) return source.NewFuture(source.DecodeNetRegionPricePerGiBResult, ctx.QueryAtTime(queryNetRegionCostPerGiB, end)) } func (pds *PrometheusMetricsQuerier) QueryNetInternetGiB(start, end time.Time) *source.Future[source.NetInternetGiBResult] { const queryName = "QueryNetInternetGiB" const queryFmtNetInternetGiB = `sum(increase(kubecost_pod_network_egress_bytes_total{internet="true", %s}[%s:%dm])) by (pod_name, namespace, uid, %s) / 1024 / 1024 / 1024` cfg := pds.promConfig minsPerResolution := cfg.DataResolutionMinutes durStr := pds.durationStringFor(start, end, minsPerResolution, true) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryNetInternetGiB := fmt.Sprintf(queryFmtNetInternetGiB, cfg.ClusterFilter, durStr, minsPerResolution, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNetInternetGiB) ctx := pds.promContexts.NewNamedContext(AllocationContextName) return source.NewFuture(source.DecodeNetInternetGiBResult, ctx.QueryAtTime(queryNetInternetGiB, end)) } func (pds *PrometheusMetricsQuerier) QueryNetInternetPricePerGiB(start, end time.Time) *source.Future[source.NetInternetPricePerGiBResult] { const queryName = "QueryNetInternetPricePerGiB" const queryFmtNetInternetCostPerGiB = `avg(avg_over_time(kubecost_network_internet_egress_cost{%s}[%s])) by (%s)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryNetInternetCostPerGiB := fmt.Sprintf(queryFmtNetInternetCostPerGiB, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNetInternetCostPerGiB) ctx := pds.promContexts.NewNamedContext(AllocationContextName) return source.NewFuture(source.DecodeNetInternetPricePerGiBResult, ctx.QueryAtTime(queryNetInternetCostPerGiB, end)) } func (pds *PrometheusMetricsQuerier) QueryNetInternetServiceGiB(start, end time.Time) *source.Future[source.NetInternetServiceGiBResult] { const queryName = "QueryNetInternetServiceGiB" 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` cfg := pds.promConfig minsPerResolution := cfg.DataResolutionMinutes durStr := pds.durationStringFor(start, end, minsPerResolution, true) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryNetInternetGiB := fmt.Sprintf(queryFmtNetInternetGiB, cfg.ClusterFilter, durStr, minsPerResolution, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNetInternetGiB) ctx := pds.promContexts.NewNamedContext(NetworkInsightsContextName) return source.NewFuture(source.DecodeNetInternetServiceGiBResult, ctx.QueryAtTime(queryNetInternetGiB, end)) } func (pds *PrometheusMetricsQuerier) QueryNetNatGatewayPricePerGiB(start, end time.Time) *source.Future[source.NetNatGatewayPricePerGiBResult] { const queryName = "QueryNetNatGatewayPricePerGiB" const queryFmtNetNatGatewayPricePerGiB = `avg(avg_over_time(kubecost_network_nat_gateway_egress_cost{%s}[%s])) by (%s)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryNetNatGatewayPricePerGiB := fmt.Sprintf(queryFmtNetNatGatewayPricePerGiB, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNetNatGatewayPricePerGiB) ctx := pds.promContexts.NewNamedContext(AllocationContextName) return source.NewFuture(source.DecodeNetNatGatewayPricePerGiBResult, ctx.QueryAtTime(queryNetNatGatewayPricePerGiB, end)) } func (pds *PrometheusMetricsQuerier) QueryNetNatGatewayGiB(start, end time.Time) *source.Future[source.NetNatGatewayGiBResult] { const queryName = "QueryNetNatGatewayGiB" 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` cfg := pds.promConfig minsPerResolution := cfg.DataResolutionMinutes durStr := pds.durationStringFor(start, end, minsPerResolution, true) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryNetNatGatewayGiB := fmt.Sprintf(queryFmtNetNatGatewayGiB, cfg.ClusterFilter, durStr, minsPerResolution, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNetNatGatewayGiB) ctx := pds.promContexts.NewNamedContext(NetworkInsightsContextName) return source.NewFuture(source.DecodeNetNatGatewayGiBResult, ctx.QueryAtTime(queryNetNatGatewayGiB, end)) } func (pds *PrometheusMetricsQuerier) QueryNetTransferBytes(start, end time.Time) *source.Future[source.NetTransferBytesResult] { const queryName = "QueryNetTransferBytes" const queryFmtNetTransferBytes = `sum(increase(container_network_transmit_bytes_total{pod!="", %s}[%s:%dm])) by (pod_name, pod, namespace, uid, %s)` cfg := pds.promConfig minsPerResolution := cfg.DataResolutionMinutes durStr := pds.durationStringFor(start, end, minsPerResolution, true) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryNetTransferBytes := fmt.Sprintf(queryFmtNetTransferBytes, cfg.ClusterFilter, durStr, minsPerResolution, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNetTransferBytes) ctx := pds.promContexts.NewNamedContext(AllocationContextName) return source.NewFuture(source.DecodeNetTransferBytesResult, ctx.QueryAtTime(queryNetTransferBytes, end)) } func (pds *PrometheusMetricsQuerier) QueryNetZoneIngressGiB(start, end time.Time) *source.Future[source.NetZoneIngressGiBResult] { const queryName = "QueryNetZoneIngressGiB" 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` cfg := pds.promConfig minsPerResolution := cfg.DataResolutionMinutes durStr := pds.durationStringFor(start, end, minsPerResolution, true) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryNetZoneCostPerGiB := fmt.Sprintf(queryFmtIngNetZoneGiB, cfg.ClusterFilter, durStr, minsPerResolution, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNetZoneCostPerGiB) ctx := pds.promContexts.NewNamedContext(NetworkInsightsContextName) return source.NewFuture(source.DecodeNetZoneIngressGiBResult, ctx.QueryAtTime(queryNetZoneCostPerGiB, end)) } func (pds *PrometheusMetricsQuerier) QueryNetRegionIngressGiB(start, end time.Time) *source.Future[source.NetRegionIngressGiBResult] { const queryName = "QueryNetRegionIngressGiB" 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` cfg := pds.promConfig minsPerResolution := cfg.DataResolutionMinutes durStr := pds.durationStringFor(start, end, minsPerResolution, true) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryNetRegionIngGiB := fmt.Sprintf(queryFmtIngNetRegionGiB, cfg.ClusterFilter, durStr, minsPerResolution, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNetRegionIngGiB) ctx := pds.promContexts.NewNamedContext(NetworkInsightsContextName) return source.NewFuture(source.DecodeNetRegionIngressGiBResult, ctx.QueryAtTime(queryNetRegionIngGiB, end)) } func (pds *PrometheusMetricsQuerier) QueryNetInternetIngressGiB(start, end time.Time) *source.Future[source.NetInternetIngressGiBResult] { const queryName = "QueryNetInternetIngressGiB" const queryFmtNetIngInternetGiB = `sum(increase(kubecost_pod_network_ingress_bytes_total{internet="true", %s}[%s:%dm])) by (pod_name, namespace, uid, %s) / 1024 / 1024 / 1024` cfg := pds.promConfig minsPerResolution := cfg.DataResolutionMinutes durStr := pds.durationStringFor(start, end, minsPerResolution, true) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryNetIngInternetGiB := fmt.Sprintf(queryFmtNetIngInternetGiB, cfg.ClusterFilter, durStr, minsPerResolution, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNetIngInternetGiB) ctx := pds.promContexts.NewNamedContext(NetworkInsightsContextName) return source.NewFuture(source.DecodeNetInternetIngressGiBResult, ctx.QueryAtTime(queryNetIngInternetGiB, end)) } func (pds *PrometheusMetricsQuerier) QueryNetInternetServiceIngressGiB(start, end time.Time) *source.Future[source.NetInternetServiceIngressGiBResult] { const queryName = "QueryNetInternetServiceIngressGiB" 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` cfg := pds.promConfig minsPerResolution := cfg.DataResolutionMinutes durStr := pds.durationStringFor(start, end, minsPerResolution, true) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryNetIngInternetGiB := fmt.Sprintf(queryFmtIngNetInternetGiB, cfg.ClusterFilter, durStr, minsPerResolution, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNetIngInternetGiB) ctx := pds.promContexts.NewNamedContext(NetworkInsightsContextName) return source.NewFuture(source.DecodeNetInternetServiceIngressGiBResult, ctx.QueryAtTime(queryNetIngInternetGiB, end)) } func (pds *PrometheusMetricsQuerier) QueryNetNatGatewayIngressPricePerGiB(start, end time.Time) *source.Future[source.NetNatGatewayPricePerGiBResult] { const queryName = "QueryNetNatGatewayIngressPricePerGiB" const queryFmtNetNatGatewayIngressPricePerGiB = `avg(avg_over_time(kubecost_network_nat_gateway_ingress_cost{%s}[%s])) by (%s)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryNetNatGatewayIngressPricePerGiB := fmt.Sprintf(queryFmtNetNatGatewayIngressPricePerGiB, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNetNatGatewayIngressPricePerGiB) ctx := pds.promContexts.NewNamedContext(AllocationContextName) return source.NewFuture(source.DecodeNetNatGatewayPricePerGiBResult, ctx.QueryAtTime(queryNetNatGatewayIngressPricePerGiB, end)) } func (pds *PrometheusMetricsQuerier) QueryNetNatGatewayIngressGiB(start, end time.Time) *source.Future[source.NetNatGatewayIngressGiBResult] { const queryName = "QueryNetNatGatewayIngressGiB" 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` cfg := pds.promConfig minsPerResolution := cfg.DataResolutionMinutes durStr := pds.durationStringFor(start, end, minsPerResolution, true) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryNetNatGatewayIngressGiB := fmt.Sprintf(queryFmtNetNatGatewayIngressGiB, cfg.ClusterFilter, durStr, minsPerResolution, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNetNatGatewayIngressGiB) ctx := pds.promContexts.NewNamedContext(NetworkInsightsContextName) return source.NewFuture(source.DecodeNetNatGatewayIngressGiBResult, ctx.QueryAtTime(queryNetNatGatewayIngressGiB, end)) } func (pds *PrometheusMetricsQuerier) QueryNetReceiveBytes(start, end time.Time) *source.Future[source.NetReceiveBytesResult] { const queryName = "QueryNetReceiveBytes" const queryFmtNetReceiveBytes = `sum(increase(container_network_receive_bytes_total{pod!="", %s}[%s:%dm])) by (pod_name, pod, namespace, uid, %s)` cfg := pds.promConfig minsPerResolution := cfg.DataResolutionMinutes durStr := pds.durationStringFor(start, end, minsPerResolution, true) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryNetReceiveBytes := fmt.Sprintf(queryFmtNetReceiveBytes, cfg.ClusterFilter, durStr, minsPerResolution, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNetReceiveBytes) ctx := pds.promContexts.NewNamedContext(AllocationContextName) return source.NewFuture(source.DecodeNetReceiveBytesResult, ctx.QueryAtTime(queryNetReceiveBytes, end)) } func (pds *PrometheusMetricsQuerier) QueryNamespaceInfo(start, end time.Time) *source.Future[source.NamespaceInfoResult] { const queryName = "QueryNamespaceInfo" const queryFmtNamespaceInfo = `avg(avg_over_time(namespace_info{%s}[%s])) by (%s, uid, namespace)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryNamespaceInfo := fmt.Sprintf(queryFmtNamespaceInfo, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNamespaceInfo) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeNamespaceInfoResult, ctx.QueryAtTime(queryNamespaceInfo, end)) } // Note: namespace_info is not currently emitted func (pds *PrometheusMetricsQuerier) QueryNamespaceUptime(start, end time.Time) *source.Future[source.UptimeResult] { const queryName = "QueryNamespaceUptime" const queryFmtNamespaceUptime = `avg(namespace_info{%s}) by (%s, uid)[%s:%dm]` cfg := pds.promConfig minsPerResolution := cfg.DataResolutionMinutes durStr := pds.durationStringFor(start, end, minsPerResolution, false) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryNamespaceUptime := fmt.Sprintf(queryFmtNamespaceUptime, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryFmtNamespaceUptime) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeUptimeResult, ctx.QueryAtTime(queryNamespaceUptime, end)) } func (pds *PrometheusMetricsQuerier) QueryNamespaceLabels(start, end time.Time) *source.Future[source.NamespaceLabelsResult] { const queryName = "QueryNamespaceLabels" const queryFmtNamespaceLabels = `avg_over_time(kube_namespace_labels{%s}[%s])` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryNamespaceLabels := fmt.Sprintf(queryFmtNamespaceLabels, cfg.ClusterFilter, durStr) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNamespaceLabels) ctx := pds.promContexts.NewNamedContext(AllocationContextName) return source.NewFuture(source.DecodeNamespaceLabelsResult, ctx.QueryAtTime(queryNamespaceLabels, end)) } func (pds *PrometheusMetricsQuerier) QueryNamespaceAnnotations(start, end time.Time) *source.Future[source.NamespaceAnnotationsResult] { const queryName = "QueryNamespaceAnnotations" const queryFmtNamespaceAnnotations = `avg_over_time(kube_namespace_annotations{%s}[%s])` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryNamespaceAnnotations := fmt.Sprintf(queryFmtNamespaceAnnotations, cfg.ClusterFilter, durStr) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryNamespaceAnnotations) ctx := pds.promContexts.NewNamedContext(AllocationContextName) return source.NewFuture(source.DecodeNamespaceAnnotationsResult, ctx.QueryAtTime(queryNamespaceAnnotations, end)) } func (pds *PrometheusMetricsQuerier) QueryPodLabels(start, end time.Time) *source.Future[source.PodLabelsResult] { const queryName = "QueryPodLabels" const queryFmtPodLabels = `avg_over_time(kube_pod_labels{%s}[%s])` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryPodLabels := fmt.Sprintf(queryFmtPodLabels, cfg.ClusterFilter, durStr) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryPodLabels) ctx := pds.promContexts.NewNamedContext(AllocationContextName) return source.NewFuture(source.DecodePodLabelsResult, ctx.QueryAtTime(queryPodLabels, end)) } func (pds *PrometheusMetricsQuerier) QueryPodAnnotations(start, end time.Time) *source.Future[source.PodAnnotationsResult] { const queryName = "QueryPodAnnotations" const queryFmtPodAnnotations = `avg_over_time(kube_pod_annotations{%s}[%s])` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryPodAnnotations := fmt.Sprintf(queryFmtPodAnnotations, cfg.ClusterFilter, durStr) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryPodAnnotations) ctx := pds.promContexts.NewNamedContext(AllocationContextName) return source.NewFuture(source.DecodePodAnnotationsResult, ctx.QueryAtTime(queryPodAnnotations, end)) } func (pds *PrometheusMetricsQuerier) QueryServiceInfo(start, end time.Time) *source.Future[source.ServiceInfoResult] { const queryName = "QueryServiceInfo" const queryFmtServiceInfo = `avg(avg_over_time(service_selector_labels{%s}[%s])) by (%s, uid, namespace_uid, service, service_type)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryServiceInfo := fmt.Sprintf(queryFmtServiceInfo, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryServiceInfo) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeServiceInfoResult, ctx.QueryAtTime(queryServiceInfo, end)) } func (pds *PrometheusMetricsQuerier) QueryServiceUptime(start, end time.Time) *source.Future[source.UptimeResult] { const queryName = "QueryServiceUptime" const queryFmtServiceUptime = `avg(service_selector_labels{%s}) by (%s, uid)[%s:%dm]` cfg := pds.promConfig minsPerResolution := cfg.DataResolutionMinutes durStr := pds.durationStringFor(start, end, minsPerResolution, false) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryServiceUptime := fmt.Sprintf(queryFmtServiceUptime, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryServiceUptime) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeUptimeResult, ctx.QueryAtTime(queryServiceUptime, end)) } func (pds *PrometheusMetricsQuerier) QueryServiceSelectorLabels(start, end time.Time) *source.Future[source.ServiceLabelsResult] { const queryName = "QueryServiceSelectorLabels" const queryFmtServiceSelectorLabels = `avg_over_time(service_selector_labels{%s}[%s])` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryServiceSelectorLabels := fmt.Sprintf(queryFmtServiceSelectorLabels, cfg.ClusterFilter, durStr) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryServiceSelectorLabels) ctx := pds.promContexts.NewNamedContext(AllocationContextName) return source.NewFuture(source.DecodeServiceLabelsResult, ctx.QueryAtTime(queryServiceSelectorLabels, end)) } func (pds *PrometheusMetricsQuerier) QueryDeploymentInfo(start, end time.Time) *source.Future[source.DeploymentInfoResult] { const queryName = "QueryDeploymentInfo" const queryFmtDeploymentInfo = `avg(avg_over_time(deployment_info{%s}[%s])) by (%s, uid, namespace_uid, deployment)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryDeploymentInfo := fmt.Sprintf(queryFmtDeploymentInfo, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryDeploymentInfo) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeDeploymentInfoResult, ctx.QueryAtTime(queryDeploymentInfo, end)) } func (pds *PrometheusMetricsQuerier) QueryDeploymentUptime(start, end time.Time) *source.Future[source.UptimeResult] { const queryName = "QueryDeploymentUptime" const queryFmtDeploymentUptime = `avg(deployment_info{%s}) by (%s, uid)[%s:%dm]` cfg := pds.promConfig minsPerResolution := cfg.DataResolutionMinutes durStr := pds.durationStringFor(start, end, minsPerResolution, false) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryDeploymentUptime := fmt.Sprintf(queryFmtDeploymentUptime, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryDeploymentUptime) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeUptimeResult, ctx.QueryAtTime(queryDeploymentUptime, end)) } func (pds *PrometheusMetricsQuerier) QueryDeploymentLabels(start, end time.Time) *source.Future[source.LabelsResult] { const queryName = "QueryDeploymentLabels" const queryFmtDeploymentLabels = `avg_over_time(deployment_labels{%s}[%s])` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryDeploymentLabels := fmt.Sprintf(queryFmtDeploymentLabels, cfg.ClusterFilter, durStr) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryDeploymentLabels) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeLabelsResult, ctx.QueryAtTime(queryDeploymentLabels, end)) } func (pds *PrometheusMetricsQuerier) QueryDeploymentAnnotations(start, end time.Time) *source.Future[source.AnnotationsResult] { const queryName = "QueryDeploymentAnnotations" const queryFmtDeploymentAnnotations = `avg_over_time(deployment_annotations{%s}[%s])` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryDeploymentAnnotations := fmt.Sprintf(queryFmtDeploymentAnnotations, cfg.ClusterFilter, durStr) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryDeploymentAnnotations) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeAnnotationsResult, ctx.QueryAtTime(queryDeploymentAnnotations, end)) } func (pds *PrometheusMetricsQuerier) QueryDeploymentMatchLabels(start, end time.Time) *source.Future[source.DeploymentLabelsResult] { const queryName = "QueryDeploymentMatchLabels" const queryFmtDeploymentLabels = `avg_over_time(deployment_match_labels{%s}[%s])` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryDeploymentMatchLabels := fmt.Sprintf(queryFmtDeploymentLabels, cfg.ClusterFilter, durStr) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryDeploymentMatchLabels) ctx := pds.promContexts.NewNamedContext(AllocationContextName) return source.NewFuture(source.DecodeDeploymentLabelsResult, ctx.QueryAtTime(queryDeploymentMatchLabels, end)) } func (pds *PrometheusMetricsQuerier) QueryStatefulSetInfo(start, end time.Time) *source.Future[source.StatefulSetInfoResult] { const queryName = "QueryStatefulSetInfo" const queryFmtStatefulSetInfo = `avg(avg_over_time(statefulset_info{%s}[%s])) by (%s, uid, namespace_uid, statefulSet)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryStatefulSetInfo := fmt.Sprintf(queryFmtStatefulSetInfo, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryStatefulSetInfo) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeStatefulSetInfoResult, ctx.QueryAtTime(queryStatefulSetInfo, end)) } func (pds *PrometheusMetricsQuerier) QueryStatefulSetUptime(start, end time.Time) *source.Future[source.UptimeResult] { const queryName = "QueryStatefulSetUptime" const queryFmtStatefulSetUptime = `avg(statefulset_info{%s}) by (%s, uid)[%s:%dm]` cfg := pds.promConfig minsPerResolution := cfg.DataResolutionMinutes durStr := pds.durationStringFor(start, end, minsPerResolution, false) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryStatefulSetUptime := fmt.Sprintf(queryFmtStatefulSetUptime, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryStatefulSetUptime) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeUptimeResult, ctx.QueryAtTime(queryStatefulSetUptime, end)) } func (pds *PrometheusMetricsQuerier) QueryStatefulSetLabels(start, end time.Time) *source.Future[source.LabelsResult] { const queryName = "QueryStatefulSetLabels" const queryFmtStatefulSetLabels = `avg_over_time(statefulset_labels{%s}[%s])` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryStatefulSetLabels := fmt.Sprintf(queryFmtStatefulSetLabels, cfg.ClusterFilter, durStr) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryStatefulSetLabels) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeLabelsResult, ctx.QueryAtTime(queryStatefulSetLabels, end)) } func (pds *PrometheusMetricsQuerier) QueryStatefulSetAnnotations(start, end time.Time) *source.Future[source.AnnotationsResult] { const queryName = "QueryStatefulSetAnnotations" const queryFmtStatefulSetAnnotations = `avg_over_time(statefulset_annotations{%s}[%s])` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryStatefulSetAnnotations := fmt.Sprintf(queryFmtStatefulSetAnnotations, cfg.ClusterFilter, durStr) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryStatefulSetAnnotations) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeAnnotationsResult, ctx.QueryAtTime(queryStatefulSetAnnotations, end)) } func (pds *PrometheusMetricsQuerier) QueryStatefulSetMatchLabels(start, end time.Time) *source.Future[source.StatefulSetLabelsResult] { const queryName = "QueryStatefulSetMatchLabels" const queryFmtStatefulSetMatchLabels = `avg_over_time(statefulSet_match_labels{%s}[%s])` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryStatefulSetMatchLabels := fmt.Sprintf(queryFmtStatefulSetMatchLabels, cfg.ClusterFilter, durStr) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryStatefulSetMatchLabels) ctx := pds.promContexts.NewNamedContext(AllocationContextName) return source.NewFuture(source.DecodeStatefulSetLabelsResult, ctx.QueryAtTime(queryStatefulSetMatchLabels, end)) } func (pds *PrometheusMetricsQuerier) QueryDaemonSetInfo(start, end time.Time) *source.Future[source.DaemonSetInfoResult] { const queryName = "QueryDaemonSetInfo" const queryFmtDaemonSetInfo = `avg(avg_over_time(daemonset_info{%s}[%s])) by (%s, uid, namespace_uid, daemonset)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryDaemonSetInfo := fmt.Sprintf(queryFmtDaemonSetInfo, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryDaemonSetInfo) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeDaemonSetInfoResult, ctx.QueryAtTime(queryDaemonSetInfo, end)) } func (pds *PrometheusMetricsQuerier) QueryDaemonSetUptime(start, end time.Time) *source.Future[source.UptimeResult] { const queryName = "QueryDaemonSetUptime" const queryFmtDaemonSetUptime = `avg(daemonset_info{%s}) by (%s, uid)[%s:%dm]` cfg := pds.promConfig minsPerResolution := cfg.DataResolutionMinutes durStr := pds.durationStringFor(start, end, minsPerResolution, false) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryDaemonSetUptime := fmt.Sprintf(queryFmtDaemonSetUptime, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryDaemonSetUptime) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeUptimeResult, ctx.QueryAtTime(queryDaemonSetUptime, end)) } func (pds *PrometheusMetricsQuerier) QueryDaemonSetLabels(start, end time.Time) *source.Future[source.LabelsResult] { const queryName = "QueryDaemonSetLabels" const queryFmtDaemonSetLabels = `avg_over_time(daemonset_labels{%s}[%s])` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryDaemonSetLabels := fmt.Sprintf(queryFmtDaemonSetLabels, cfg.ClusterFilter, durStr) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryDaemonSetLabels) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeLabelsResult, ctx.QueryAtTime(queryDaemonSetLabels, end)) } func (pds *PrometheusMetricsQuerier) QueryDaemonSetAnnotations(start, end time.Time) *source.Future[source.AnnotationsResult] { const queryName = "QueryDaemonSetAnnotations" const queryFmtDaemonSetAnnotations = `avg_over_time(daemonset_annotations{%s}[%s])` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryDaemonSetAnnotations := fmt.Sprintf(queryFmtDaemonSetAnnotations, cfg.ClusterFilter, durStr) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryDaemonSetAnnotations) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeAnnotationsResult, ctx.QueryAtTime(queryDaemonSetAnnotations, end)) } func (pds *PrometheusMetricsQuerier) QueryJobInfo(start, end time.Time) *source.Future[source.JobInfoResult] { const queryName = "QueryJobInfo" const queryFmtJobInfo = `avg(avg_over_time(job_info{%s}[%s])) by (%s, uid, namespace_uid, job)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryJobInfo := fmt.Sprintf(queryFmtJobInfo, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryJobInfo) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeJobInfoResult, ctx.QueryAtTime(queryJobInfo, end)) } func (pds *PrometheusMetricsQuerier) QueryJobUptime(start, end time.Time) *source.Future[source.UptimeResult] { const queryName = "QueryJobUptime" const queryFmtJobUptime = `avg(job_info{%s}) by (%s, uid)[%s:%dm]` cfg := pds.promConfig minsPerResolution := cfg.DataResolutionMinutes durStr := pds.durationStringFor(start, end, minsPerResolution, false) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryJobUptime := fmt.Sprintf(queryFmtJobUptime, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryJobUptime) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeUptimeResult, ctx.QueryAtTime(queryJobUptime, end)) } func (pds *PrometheusMetricsQuerier) QueryJobLabels(start, end time.Time) *source.Future[source.LabelsResult] { const queryName = "QueryJobLabels" const queryFmtJobLabels = `avg_over_time(job_labels{%s}[%s])` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryJobLabels := fmt.Sprintf(queryFmtJobLabels, cfg.ClusterFilter, durStr) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryJobLabels) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeLabelsResult, ctx.QueryAtTime(queryJobLabels, end)) } func (pds *PrometheusMetricsQuerier) QueryJobAnnotations(start, end time.Time) *source.Future[source.AnnotationsResult] { const queryName = "QueryJobAnnotations" const queryFmtJobAnnotations = `avg_over_time(job_annotations{%s}[%s])` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryJobAnnotations := fmt.Sprintf(queryFmtJobAnnotations, cfg.ClusterFilter, durStr) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryJobAnnotations) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeAnnotationsResult, ctx.QueryAtTime(queryJobAnnotations, end)) } func (pds *PrometheusMetricsQuerier) QueryCronJobInfo(start, end time.Time) *source.Future[source.CronJobInfoResult] { const queryName = "QueryCronJobInfo" const queryFmtCronJobInfo = `avg(avg_over_time(cronjob_info{%s}[%s])) by (%s, uid, namespace_uid, cronjob)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryCronJobInfo := fmt.Sprintf(queryFmtCronJobInfo, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryCronJobInfo) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeCronJobInfoResult, ctx.QueryAtTime(queryCronJobInfo, end)) } func (pds *PrometheusMetricsQuerier) QueryCronJobUptime(start, end time.Time) *source.Future[source.UptimeResult] { const queryName = "QueryCronJobUptime" const queryFmtCronJobUptime = `avg(cronjob_info{%s}) by (%s, uid)[%s:%dm]` cfg := pds.promConfig minsPerResolution := cfg.DataResolutionMinutes durStr := pds.durationStringFor(start, end, minsPerResolution, false) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryCronJobUptime := fmt.Sprintf(queryFmtCronJobUptime, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryCronJobUptime) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeUptimeResult, ctx.QueryAtTime(queryCronJobUptime, end)) } func (pds *PrometheusMetricsQuerier) QueryCronJobLabels(start, end time.Time) *source.Future[source.LabelsResult] { const queryName = "QueryCronJobLabels" const queryFmtCronJobLabels = `avg_over_time(cronjob_labels{%s}[%s])` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryCronJobLabels := fmt.Sprintf(queryFmtCronJobLabels, cfg.ClusterFilter, durStr) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryCronJobLabels) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeLabelsResult, ctx.QueryAtTime(queryCronJobLabels, end)) } func (pds *PrometheusMetricsQuerier) QueryCronJobAnnotations(start, end time.Time) *source.Future[source.AnnotationsResult] { const queryName = "QueryCronJobAnnotations" const queryFmtCronJobAnnotations = `avg_over_time(cronjob_annotations{%s}[%s])` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryCronJobAnnotations := fmt.Sprintf(queryFmtCronJobAnnotations, cfg.ClusterFilter, durStr) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryCronJobAnnotations) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeAnnotationsResult, ctx.QueryAtTime(queryCronJobAnnotations, end)) } func (pds *PrometheusMetricsQuerier) QueryReplicaSetInfo(start, end time.Time) *source.Future[source.ReplicaSetInfoResult] { const queryName = "QueryReplicaSetInfo" const queryFmtReplicaSetInfo = `avg(avg_over_time(replicaset_info{%s}[%s])) by (%s, uid, namespace_uid, replicaset)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryReplicaSetInfo := fmt.Sprintf(queryFmtReplicaSetInfo, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryReplicaSetInfo) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeReplicaSetInfoResult, ctx.QueryAtTime(queryReplicaSetInfo, end)) } func (pds *PrometheusMetricsQuerier) QueryReplicaSetUptime(start, end time.Time) *source.Future[source.UptimeResult] { const queryName = "QueryReplicaSetUptime" const queryFmtReplicaSetUptime = `avg(replicaset_info{%s}) by (%s, uid)[%s:%dm]` cfg := pds.promConfig minsPerResolution := cfg.DataResolutionMinutes durStr := pds.durationStringFor(start, end, minsPerResolution, false) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryReplicaSetUptime := fmt.Sprintf(queryFmtReplicaSetUptime, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryReplicaSetUptime) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeUptimeResult, ctx.QueryAtTime(queryReplicaSetUptime, end)) } func (pds *PrometheusMetricsQuerier) QueryReplicaSetLabels(start, end time.Time) *source.Future[source.LabelsResult] { const queryName = "QueryReplicaSetLabels" const queryFmtReplicaSetLabels = `avg_over_time(replicaset_labels{%s}[%s])` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryReplicaSetLabels := fmt.Sprintf(queryFmtReplicaSetLabels, cfg.ClusterFilter, durStr) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryReplicaSetLabels) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeLabelsResult, ctx.QueryAtTime(queryReplicaSetLabels, end)) } func (pds *PrometheusMetricsQuerier) QueryReplicaSetAnnotations(start, end time.Time) *source.Future[source.AnnotationsResult] { const queryName = "QueryReplicaSetAnnotations" const queryFmtReplicaSetAnnotations = `avg_over_time(replicaset_annotations{%s}[%s])` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryReplicaSetAnnotations := fmt.Sprintf(queryFmtReplicaSetAnnotations, cfg.ClusterFilter, durStr) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryReplicaSetAnnotations) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeAnnotationsResult, ctx.QueryAtTime(queryReplicaSetAnnotations, end)) } func (pds *PrometheusMetricsQuerier) QueryReplicaSetOwners(start, end time.Time) *source.Future[source.OwnerResult] { const queryName = "QueryReplicaSetOwners" const queryFmtReplicaSetOwners = `avg(avg_over_time(kube_replicaset_owner{%s}[%s])) by (%s, uid, owner_uid, owner_kind)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryReplicaSetOwners := fmt.Sprintf(queryFmtReplicaSetOwners, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryReplicaSetOwners) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeOwnerResult, ctx.QueryAtTime(queryReplicaSetOwners, end)) } func (pds *PrometheusMetricsQuerier) QueryPodsWithDaemonSetOwner(start, end time.Time) *source.Future[source.PodsWithDaemonSetOwnerResult] { const queryName = "QueryPodsWithDaemonSetOwner" const queryFmtPodsWithDaemonSetOwner = `sum(avg_over_time(kube_pod_owner{owner_kind="DaemonSet", %s}[%s])) by (pod, owner_name, namespace, uid, %s)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryPodsWithDaemonSetOwner := fmt.Sprintf(queryFmtPodsWithDaemonSetOwner, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryPodsWithDaemonSetOwner) ctx := pds.promContexts.NewNamedContext(AllocationContextName) return source.NewFuture(source.DecodePodsWithDaemonSetOwnerResult, ctx.QueryAtTime(queryPodsWithDaemonSetOwner, end)) } func (pds *PrometheusMetricsQuerier) QueryPodsWithJobOwner(start, end time.Time) *source.Future[source.PodsWithJobOwnerResult] { const queryName = "QueryPodsWithJobOwner" const queryFmtPodsWithJobOwner = `sum(avg_over_time(kube_pod_owner{owner_kind="Job", %s}[%s])) by (pod, owner_name, namespace, uid, %s)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryPodsWithJobOwner := fmt.Sprintf(queryFmtPodsWithJobOwner, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryPodsWithJobOwner) ctx := pds.promContexts.NewNamedContext(AllocationContextName) return source.NewFuture(source.DecodePodsWithJobOwnerResult, ctx.QueryAtTime(queryPodsWithJobOwner, end)) } func (pds *PrometheusMetricsQuerier) QueryPodsWithReplicaSetOwner(start, end time.Time) *source.Future[source.PodsWithReplicaSetOwnerResult] { const queryName = "QueryPodsWithReplicaSetOwner" const queryFmtPodsWithReplicaSetOwner = `sum(avg_over_time(kube_pod_owner{owner_kind="ReplicaSet", %s}[%s])) by (pod, owner_name, namespace, uid, %s)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryPodsWithReplicaSetOwner := fmt.Sprintf(queryFmtPodsWithReplicaSetOwner, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryPodsWithReplicaSetOwner) ctx := pds.promContexts.NewNamedContext(AllocationContextName) return source.NewFuture(source.DecodePodsWithReplicaSetOwnerResult, ctx.QueryAtTime(queryPodsWithReplicaSetOwner, end)) } func (pds *PrometheusMetricsQuerier) QueryReplicaSetsWithoutOwners(start, end time.Time) *source.Future[source.ReplicaSetsWithoutOwnersResult] { const queryName = "QueryReplicaSetsWithoutOwners" const queryFmtReplicaSetsWithoutOwners = `avg(avg_over_time(kube_replicaset_owner{owner_kind="", owner_name="", %s}[%s])) by (replicaset, namespace, uid, %s)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryReplicaSetsWithoutOwners := fmt.Sprintf(queryFmtReplicaSetsWithoutOwners, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryReplicaSetsWithoutOwners) ctx := pds.promContexts.NewNamedContext(AllocationContextName) return source.NewFuture(source.DecodeReplicaSetsWithoutOwnersResult, ctx.QueryAtTime(queryReplicaSetsWithoutOwners, end)) } func (pds *PrometheusMetricsQuerier) QueryReplicaSetsWithRollout(start, end time.Time) *source.Future[source.ReplicaSetsWithRolloutResult] { const queryName = "QueryReplicaSetsWithRollout" const queryFmtReplicaSetsWithRolloutOwner = `avg(avg_over_time(kube_replicaset_owner{owner_kind="Rollout", %s}[%s])) by (replicaset, namespace, owner_kind, owner_name, uid, %s)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryReplicaSetsWithRolloutOwner := fmt.Sprintf(queryFmtReplicaSetsWithRolloutOwner, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryReplicaSetsWithRolloutOwner) ctx := pds.promContexts.NewNamedContext(AllocationContextName) return source.NewFuture(source.DecodeReplicaSetsWithRolloutResult, ctx.QueryAtTime(queryReplicaSetsWithRolloutOwner, end)) } // Note: The ResourceQuota metrics are _not_ emitted at the moment. Leaving the query implementations here in case we add metric emission later on. func (pds *PrometheusMetricsQuerier) QueryResourceQuotaInfo(start, end time.Time) *source.Future[source.ResourceQuotaInfoResult] { const queryName = "QueryResourceQuotaInfo" const queryFmtResourceQuotaInfo = `avg(avg_over_time(resourcequota_info{%s}[%s])) by (%s, uid, namespace_uid, resourcequota)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryResourceQuotaInfo := fmt.Sprintf(queryFmtResourceQuotaInfo, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryResourceQuotaInfo) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeResourceQuotaInfoResult, ctx.QueryAtTime(queryResourceQuotaInfo, end)) } func (pds *PrometheusMetricsQuerier) QueryResourceQuotaUptime(start, end time.Time) *source.Future[source.UptimeResult] { const queryName = "QueryResourceQuotaUptime" const queryFmtResourceQuotaUptime = `avg(resourcequota_info{%s}) by (%s, uid)[%s:%dm]` cfg := pds.promConfig minsPerResolution := cfg.DataResolutionMinutes durStr := pds.durationStringFor(start, end, minsPerResolution, false) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryResourceQuotaUptime := fmt.Sprintf(queryFmtResourceQuotaUptime, cfg.ClusterFilter, cfg.ClusterLabel, durStr, minsPerResolution) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryFmtResourceQuotaUptime) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeUptimeResult, ctx.QueryAtTime(queryResourceQuotaUptime, end)) } func (pds *PrometheusMetricsQuerier) QueryResourceQuotaSpecCPURequestAverage(start, end time.Time) *source.Future[source.ResourceResult] { const queryName = "QueryResourceQuotaSpecCPURequestAverage" const queryFmtResourceQuotaSpecCPURequests = `avg(avg_over_time(resourcequota_spec_resource_requests{resource="cpu",unit="core", %s}[%s])) by (resourcequota, namespace, uid, %s)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryResourceQuotaSpecCPURequests := fmt.Sprintf(queryFmtResourceQuotaSpecCPURequests, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryResourceQuotaSpecCPURequests) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeResourceResult, ctx.QueryAtTime(queryResourceQuotaSpecCPURequests, end)) } func (pds *PrometheusMetricsQuerier) QueryResourceQuotaSpecCPURequestMax(start, end time.Time) *source.Future[source.ResourceResult] { const queryName = "QueryResourceQuotaSpecCPURequestMax" const queryFmtResourceQuotaSpecCPURequests = `max(max_over_time(resourcequota_spec_resource_requests{resource="cpu",unit="core", %s}[%s])) by (resourcequota, namespace, uid, %s)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryResourceQuotaSpecCPURequests := fmt.Sprintf(queryFmtResourceQuotaSpecCPURequests, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryResourceQuotaSpecCPURequests) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeResourceResult, ctx.QueryAtTime(queryResourceQuotaSpecCPURequests, end)) } func (pds *PrometheusMetricsQuerier) QueryResourceQuotaSpecRAMRequestAverage(start, end time.Time) *source.Future[source.ResourceResult] { const queryName = "QueryResourceQuotaSpecRAMRequestAverage" const queryFmtResourceQuotaSpecRAMRequests = `avg(avg_over_time(resourcequota_spec_resource_requests{resource="memory",unit="byte", %s}[%s])) by (resourcequota, namespace, uid, %s)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryResourceQuotaSpecRAMRequests := fmt.Sprintf(queryFmtResourceQuotaSpecRAMRequests, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryResourceQuotaSpecRAMRequests) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeResourceResult, ctx.QueryAtTime(queryResourceQuotaSpecRAMRequests, end)) } func (pds *PrometheusMetricsQuerier) QueryResourceQuotaSpecRAMRequestMax(start, end time.Time) *source.Future[source.ResourceResult] { const queryName = "QueryResourceQuotaSpecRAMRequestMax" const queryFmtResourceQuotaSpecRAMRequests = `max(max_over_time(resourcequota_spec_resource_requests{resource="memory",unit="byte", %s}[%s])) by (resourcequota, namespace, uid, %s)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryResourceQuotaSpecRAMRequests := fmt.Sprintf(queryFmtResourceQuotaSpecRAMRequests, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryResourceQuotaSpecRAMRequests) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeResourceResult, ctx.QueryAtTime(queryResourceQuotaSpecRAMRequests, end)) } func (pds *PrometheusMetricsQuerier) QueryResourceQuotaSpecCPULimitAverage(start, end time.Time) *source.Future[source.ResourceResult] { const queryName = "QueryResourceQuotaSpecCPULimitAverage" const queryFmtResourceQuotaSpecCPULimits = `avg(avg_over_time(resourcequota_spec_resource_limits{resource="cpu",unit="core", %s}[%s])) by (resourcequota, namespace, uid, %s)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryResourceQuotaSpecCPULimits := fmt.Sprintf(queryFmtResourceQuotaSpecCPULimits, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryResourceQuotaSpecCPULimits) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeResourceResult, ctx.QueryAtTime(queryResourceQuotaSpecCPULimits, end)) } func (pds *PrometheusMetricsQuerier) QueryResourceQuotaSpecCPULimitMax(start, end time.Time) *source.Future[source.ResourceResult] { const queryName = "QueryResourceQuotaSpecCPULimitMax" const queryFmtResourceQuotaSpecCPULimits = `max(max_over_time(resourcequota_spec_resource_limits{resource="cpu",unit="core", %s}[%s])) by (resourcequota, namespace, uid, %s)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryResourceQuotaSpecCPULimits := fmt.Sprintf(queryFmtResourceQuotaSpecCPULimits, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryResourceQuotaSpecCPULimits) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeResourceResult, ctx.QueryAtTime(queryResourceQuotaSpecCPULimits, end)) } func (pds *PrometheusMetricsQuerier) QueryResourceQuotaSpecRAMLimitAverage(start, end time.Time) *source.Future[source.ResourceResult] { const queryName = "QueryResourceQuotaSpecRAMLimitAverage" const queryFmtResourceQuotaSpecRAMLimits = `avg(avg_over_time(resourcequota_spec_resource_limits{resource="memory",unit="byte", %s}[%s])) by (resourcequota, namespace, uid, %s)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryResourceQuotaSpecRAMLimits := fmt.Sprintf(queryFmtResourceQuotaSpecRAMLimits, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryResourceQuotaSpecRAMLimits) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeResourceResult, ctx.QueryAtTime(queryResourceQuotaSpecRAMLimits, end)) } func (pds *PrometheusMetricsQuerier) QueryResourceQuotaSpecRAMLimitMax(start, end time.Time) *source.Future[source.ResourceResult] { const queryName = "QueryResourceQuotaSpecRAMLimitMax" const queryFmtResourceQuotaSpecRAMLimits = `max(max_over_time(resourcequota_spec_resource_limits{resource="memory",unit="byte", %s}[%s])) by (resourcequota, namespace, uid, %s)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryResourceQuotaSpecRAMLimits := fmt.Sprintf(queryFmtResourceQuotaSpecRAMLimits, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryResourceQuotaSpecRAMLimits) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeResourceResult, ctx.QueryAtTime(queryResourceQuotaSpecRAMLimits, end)) } func (pds *PrometheusMetricsQuerier) QueryResourceQuotaStatusUsedCPURequestAverage(start, end time.Time) *source.Future[source.ResourceResult] { const queryName = "QueryResourceQuotaStatusUsedCPURequestAverage" const queryFmtResourceQuotaStatusUsedCPURequests = `avg(avg_over_time(resourcequota_status_used_resource_requests{resource="cpu",unit="core", %s}[%s])) by (resourcequota, namespace, uid, %s)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryResourceQuotaStatusUsedCPURequests := fmt.Sprintf(queryFmtResourceQuotaStatusUsedCPURequests, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryResourceQuotaStatusUsedCPURequests) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeResourceResult, ctx.QueryAtTime(queryResourceQuotaStatusUsedCPURequests, end)) } func (pds *PrometheusMetricsQuerier) QueryResourceQuotaStatusUsedCPURequestMax(start, end time.Time) *source.Future[source.ResourceResult] { const queryName = "QueryResourceQuotaStatusUsedCPURequestMax" const queryFmtResourceQuotaStatusUsedCPURequests = `max(max_over_time(resourcequota_status_used_resource_requests{resource="cpu",unit="core", %s}[%s])) by (resourcequota, namespace, uid, %s)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryResourceQuotaStatusUsedCPURequests := fmt.Sprintf(queryFmtResourceQuotaStatusUsedCPURequests, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryResourceQuotaStatusUsedCPURequests) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeResourceResult, ctx.QueryAtTime(queryResourceQuotaStatusUsedCPURequests, end)) } func (pds *PrometheusMetricsQuerier) QueryResourceQuotaStatusUsedRAMRequestAverage(start, end time.Time) *source.Future[source.ResourceResult] { const queryName = "QueryResourceQuotaStatusUsedRAMRequestAverage" const queryFmtResourceQuotaStatusUsedRAMRequests = `avg(avg_over_time(resourcequota_status_used_resource_requests{resource="memory",unit="byte", %s}[%s])) by (resourcequota, namespace, uid, %s)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryResourceQuotaStatusUsedRAMRequests := fmt.Sprintf(queryFmtResourceQuotaStatusUsedRAMRequests, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryResourceQuotaStatusUsedRAMRequests) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeResourceResult, ctx.QueryAtTime(queryResourceQuotaStatusUsedRAMRequests, end)) } func (pds *PrometheusMetricsQuerier) QueryResourceQuotaStatusUsedRAMRequestMax(start, end time.Time) *source.Future[source.ResourceResult] { const queryName = "QueryResourceQuotaStatusUsedRAMRequestMax" const queryFmtResourceQuotaStatusUsedRAMRequests = `max(max_over_time(resourcequota_status_used_resource_requests{resource="memory",unit="byte", %s}[%s])) by (resourcequota, namespace, uid, %s)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryResourceQuotaStatusUsedRAMRequests := fmt.Sprintf(queryFmtResourceQuotaStatusUsedRAMRequests, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryResourceQuotaStatusUsedRAMRequests) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeResourceResult, ctx.QueryAtTime(queryResourceQuotaStatusUsedRAMRequests, end)) } func (pds *PrometheusMetricsQuerier) QueryResourceQuotaStatusUsedCPULimitAverage(start, end time.Time) *source.Future[source.ResourceResult] { const queryName = "QueryResourceQuotaStatusUsedCPULimitAverage" const queryFmtResourceQuotaStatusUsedCPULimits = `avg(avg_over_time(resourcequota_status_used_resource_limits{resource="cpu",unit="core", %s}[%s])) by (resourcequota, namespace, uid, %s)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryResourceQuotaStatusUsedCPULimits := fmt.Sprintf(queryFmtResourceQuotaStatusUsedCPULimits, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryResourceQuotaStatusUsedCPULimits) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeResourceResult, ctx.QueryAtTime(queryResourceQuotaStatusUsedCPULimits, end)) } func (pds *PrometheusMetricsQuerier) QueryResourceQuotaStatusUsedCPULimitMax(start, end time.Time) *source.Future[source.ResourceResult] { const queryName = "QueryResourceQuotaStatusUsedCPULimitMax" const queryFmtResourceQuotaStatusUsedCPULimits = `max(max_over_time(resourcequota_status_used_resource_limits{resource="cpu",unit="core", %s}[%s])) by (resourcequota, namespace, uid, %s)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryResourceQuotaStatusUsedCPULimits := fmt.Sprintf(queryFmtResourceQuotaStatusUsedCPULimits, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryResourceQuotaStatusUsedCPULimits) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeResourceResult, ctx.QueryAtTime(queryResourceQuotaStatusUsedCPULimits, end)) } func (pds *PrometheusMetricsQuerier) QueryResourceQuotaStatusUsedRAMLimitAverage(start, end time.Time) *source.Future[source.ResourceResult] { const queryName = "QueryResourceQuotaStatusUsedRAMLimitAverage" const queryFmtResourceQuotaStatusUsedRAMLimits = `avg(avg_over_time(resourcequota_status_used_resource_limits{resource="memory",unit="byte", %s}[%s])) by (resourcequota, namespace, uid, %s)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryResourceQuotaStatusUsedRAMLimits := fmt.Sprintf(queryFmtResourceQuotaStatusUsedRAMLimits, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryResourceQuotaStatusUsedRAMLimits) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeResourceResult, ctx.QueryAtTime(queryResourceQuotaStatusUsedRAMLimits, end)) } func (pds *PrometheusMetricsQuerier) QueryResourceQuotaStatusUsedRAMLimitMax(start, end time.Time) *source.Future[source.ResourceResult] { const queryName = "QueryResourceQuotaStatusUsedRAMLimitMax" const queryFmtResourceQuotaStatusUsedRAMLimits = `max(max_over_time(resourcequota_status_used_resource_limits{resource="memory",unit="byte", %s}[%s])) by (resourcequota, namespace, uid, %s)` cfg := pds.promConfig durStr := timeutil.DurationString(end.Sub(start)) if durStr == "" { panic(fmt.Sprintf("failed to parse duration string passed to %s", queryName)) } queryResourceQuotaStatusUsedRAMLimits := fmt.Sprintf(queryFmtResourceQuotaStatusUsedRAMLimits, cfg.ClusterFilter, durStr, cfg.ClusterLabel) log.Debugf(PrometheusMetricsQueryLogFormat, queryName, end.Unix(), queryResourceQuotaStatusUsedRAMLimits) ctx := pds.promContexts.NewNamedContext(KubeModelContextName) return source.NewFuture(source.DecodeResourceResult, ctx.QueryAtTime(queryResourceQuotaStatusUsedRAMLimits, end)) } func (pds *PrometheusMetricsQuerier) QueryDataCoverage(limitDays int) (time.Time, time.Time, error) { const ( queryName = "QueryDataCoverage" queryFmtOldestSample = `min_over_time(timestamp(group(node_cpu_hourly_cost{%s}))[%s:%s])` queryFmtNewestSample = `max_over_time(timestamp(group(node_cpu_hourly_cost{%s}))[%s:%s])` ) cfg := pds.promConfig minutesPerDuration := 60 dur := time.Duration(limitDays) * timeutil.Day end := time.Now().UTC().Truncate(timeutil.Day).Add(timeutil.Day) start := end.Add(-dur) durStr := pds.durationStringFor(start, end, minutesPerDuration, false) ctx := pds.promContexts.NewNamedContext(AllocationContextName) queryOldest := fmt.Sprintf(queryFmtOldestSample, cfg.ClusterFilter, durStr, "1h") log.Debugf("[Prometheus][%s[Oldest]][At Time: %d]: %s", queryName, end.Unix(), queryOldest) resOldestFut := ctx.QueryAtTime(queryOldest, end) resOldest, err := resOldestFut.Await() if err != nil { return time.Time{}, time.Time{}, fmt.Errorf("querying oldest sample: %w", err) } if len(resOldest) == 0 || len(resOldest[0].Values) == 0 { // If node_cpu_hourly_cost metric is not available, fallback to a reasonable time range // This prevents CSV export from failing when the metric doesn't exist yet log.Warnf("QueryDataCoverage: node_cpu_hourly_cost metric not available, using fallback time range") // Use a reasonable fallback: start from 1 day ago to account for metric collection delay fallbackEnd := time.Now().UTC().Truncate(timeutil.Day) fallbackStart := fallbackEnd.AddDate(0, 0, -1) // 1 day ago return fallbackStart, fallbackEnd, nil } oldest := time.Unix(int64(resOldest[0].Values[0].Value), 0) queryNewest := fmt.Sprintf(queryFmtNewestSample, cfg.ClusterFilter, durStr, "1h") log.Debugf("[Prometheus][%s[Newest]][At Time: %d]: %s", queryName, end.Unix(), queryNewest) resNewestFut := ctx.QueryAtTime(queryNewest, end) resNewest, err := resNewestFut.Await() if err != nil { return time.Time{}, time.Time{}, fmt.Errorf("querying newest sample: %w", err) } if len(resNewest) == 0 || len(resNewest[0].Values) == 0 { // If newest query fails but oldest succeeded, use oldest as both start and end // This allows CSV export to proceed with at least some time range log.Warnf("QueryDataCoverage: newest sample query returned no results, using oldest timestamp") return oldest, oldest, nil } newest := time.Unix(int64(resNewest[0].Values[0].Value), 0) return oldest, newest, nil } // durationStringFor simplifies the determination of query duration based on the version of prom and if the function // in the query needs all data points in the vector it is provided or if it will extrapolate its own. Functions // that extrapolate will add on another resolution if given a duration that is one resolution longer than the intended // duration. func (pds *PrometheusMetricsQuerier) durationStringFor(start, end time.Time, minsPerResolution int, extrapolated bool) string { dur := end.Sub(start) // If using a version of Prometheus where the resolution needs duration offset, // we need to apply that here. // // E.g. avg(node_total_hourly_cost{}) by (node, provider_id)[60m:5m] with // time=01:00:00 will return, for a node running the entire time, 12 // timestamps where the first is 00:05:00 and the last is 01:00:00. // However, OpenCost expects for there to be 13 timestamps where the first // begins at 00:00:00. To achieve this, we must modify our query to // avg(node_total_hourly_cost{}) by (node, provider_id)[65m:5m] if pds.promConfig.IsOffsetResolution && !extrapolated { // increase the query time by the resolution dur = dur + (time.Duration(minsPerResolution) * time.Minute) } return timeutil.DurationString(dur) }