From 1ba6c8c8187363560251876347c693291a405173 Mon Sep 17 00:00:00 2001 From: Petru Anica-Popa Date: Thu, 29 Oct 2026 16:28:32 +0000 Subject: [PATCH 1/2] fix(otelmetrics): return complete query results and honor Retry-After --- test/otel/efa/dedup_test.go | 2 +- test/otel/efa/efa_test.go | 2 +- test/otel/efa/node_color_test.go | 2 +- test/otel/gpu/dcgm_test.go | 2 +- test/otel/gpu/dedup_test.go | 2 +- test/otel/gpu/node_color_test.go | 2 +- test/otel/neuron/multi_device_test.go | 4 +- test/otel/neuron/neuron_test.go | 2 +- test/otel/neuron/node_color_test.go | 4 +- test/otel/performance/performance_test.go | 2 +- test/otel/standard/cadvisor_test.go | 4 +- test/otel/standard/dedup_test.go | 10 +- test/otel/standard/ksm_test.go | 12 +- test/otel/standard/kubeletstats_test.go | 2 +- test/otel/standard/labels_common_test.go | 4 +- test/otel/standard/node_exporter_test.go | 2 +- util/otelmetrics/client.go | 94 ++- util/otelmetrics/client_test.go | 58 ++ util/otelmetrics/complete.go | 370 ++++++++++++ util/otelmetrics/complete_test.go | 670 ++++++++++++++++++++++ util/otelmetrics/query_cache.go | 33 +- 21 files changed, 1225 insertions(+), 58 deletions(-) create mode 100644 util/otelmetrics/client_test.go create mode 100644 util/otelmetrics/complete.go create mode 100644 util/otelmetrics/complete_test.go diff --git a/test/otel/efa/dedup_test.go b/test/otel/efa/dedup_test.go index 4d8b7e73c..51b968bd4 100644 --- a/test/otel/efa/dedup_test.go +++ b/test/otel/efa/dedup_test.go @@ -48,7 +48,7 @@ func TestEFANoDuplicateSeries(t *testing.T) { `efa_rx_bytes{"@resource.k8s.cluster.name"="%s","@resource.k8s.node.name"="%s","aws.efa.device"="%s"}`, escaped, pinNode, pinDevice, ) - pinned, err := client.Query(ctx, promql) + pinned, err := client.QueryComplete(ctx, promql) require.NoError(t, err, "querying pinned efa_rx_bytes") require.NotEmpty(t, pinned, "pinned efa_rx_bytes returned 0 results") require.Equal(t, 1, len(pinned), diff --git a/test/otel/efa/efa_test.go b/test/otel/efa/efa_test.go index 2d80ff1ef..0f2993192 100644 --- a/test/otel/efa/efa_test.go +++ b/test/otel/efa/efa_test.go @@ -406,7 +406,7 @@ func TestEFANodeGroupCoverage(t *testing.T) { promql := fmt.Sprintf( `efa_rx_bytes{"@resource.k8s.cluster.name"="%s","@resource.host.type"="%s"}`, escaped, "c5n.9xlarge") - results, err := client.Query(context.Background(), promql) + results, err := client.QueryComplete(context.Background(), promql) require.NoError(t, err, "querying efa_rx_bytes on c5n.9xlarge") require.True(t, len(results) > 0, "EFA metrics missing from c5n.9xlarge nodes — awsefareceiver not collecting?") diff --git a/test/otel/efa/node_color_test.go b/test/otel/efa/node_color_test.go index ab80aca2d..29dd6bf4c 100644 --- a/test/otel/efa/node_color_test.go +++ b/test/otel/efa/node_color_test.go @@ -31,7 +31,7 @@ func TestEFAHasYellowNodeColor(t *testing.T) { promql := fmt.Sprintf( `%s{"@resource.k8s.cluster.name"="%s","@resource.%s"="yellow"}`, metricName, escaped, nodeColorLabel) - results, err := client.Query(context.Background(), promql) + results, err := client.QueryComplete(context.Background(), promql) require.NoError(t, err, "querying %s with node-color=yellow filter", metricName) require.True(t, len(results) > 0, "%s: no results from c5n.9xlarge nodes (expected node-color=yellow)", metricName) diff --git a/test/otel/gpu/dcgm_test.go b/test/otel/gpu/dcgm_test.go index 4a096ab5b..6b02f6651 100644 --- a/test/otel/gpu/dcgm_test.go +++ b/test/otel/gpu/dcgm_test.go @@ -422,7 +422,7 @@ func TestDCGMNodeGroupCoverage(t *testing.T) { promql := fmt.Sprintf( `DCGM_FI_DEV_GPU_UTIL{"@resource.k8s.cluster.name"="%s","@resource.host.type"="%s"}`, escaped, ng.InstanceType) - results, err := client.Query(context.Background(), promql) + results, err := client.QueryComplete(context.Background(), promql) require.NoError(t, err, "querying DCGM_FI_DEV_GPU_UTIL on %s", ng.Description) require.True(t, len(results) > 0, "DCGM missing from %s nodes (%s) — dcgm-exporter not running?", diff --git a/test/otel/gpu/dedup_test.go b/test/otel/gpu/dedup_test.go index abc5d860e..7a796bf73 100644 --- a/test/otel/gpu/dedup_test.go +++ b/test/otel/gpu/dedup_test.go @@ -50,7 +50,7 @@ func TestDCGMNoDuplicateSeries(t *testing.T) { `DCGM_FI_DEV_GPU_UTIL{"@resource.k8s.cluster.name"="%s","@resource.k8s.node.name"="%s",UUID="%s"}`, escaped, pinNode, pinUUID, ) - pinned, err := client.Query(ctx, promql) + pinned, err := client.QueryComplete(ctx, promql) require.NoError(t, err, "querying pinned DCGM_FI_DEV_GPU_UTIL") require.NotEmpty(t, pinned, "pinned DCGM_FI_DEV_GPU_UTIL returned 0 results") require.Equal(t, 1, len(pinned), diff --git a/test/otel/gpu/node_color_test.go b/test/otel/gpu/node_color_test.go index c9d1be06e..3c5f44057 100644 --- a/test/otel/gpu/node_color_test.go +++ b/test/otel/gpu/node_color_test.go @@ -35,7 +35,7 @@ func TestDCGMHasGreenNodeColor(t *testing.T) { promql := fmt.Sprintf( `%s{"@resource.k8s.cluster.name"="%s","@resource.%s"="green"}`, metricName, escaped, nodeColorLabel) - results, err := client.Query(context.Background(), promql) + results, err := client.QueryComplete(context.Background(), promql) require.NoError(t, err, "querying %s with node-color=green filter", metricName) require.True(t, len(results) > 0, "%s: no results from GPU nodes (expected node-color=green)", metricName) diff --git a/test/otel/neuron/multi_device_test.go b/test/otel/neuron/multi_device_test.go index 9b2e96dc2..c82698b32 100644 --- a/test/otel/neuron/multi_device_test.go +++ b/test/otel/neuron/multi_device_test.go @@ -273,7 +273,7 @@ func TestMultiNeuronCorrelatedCoreCount(t *testing.T) { promql := fmt.Sprintf( `neuroncore_utilization_ratio{"@resource.k8s.cluster.name"="%s","@resource.k8s.pod.name"=~"neuron-burn-multi.*"}`, escaped) - burnResults, err := client.Query(ctx, promql) + burnResults, err := client.QueryComplete(ctx, promql) require.NoError(t, err, "querying neuroncore_utilization_ratio for neuron-burn-multi") require.Equal(t, 2, len(burnResults), fmt.Sprintf("Expected 2 cores correlated to neuron-burn-multi (1 whole device × 2 cores), got %d", len(burnResults))) @@ -289,7 +289,7 @@ func TestMultiNeuronBurnDevicePodLabels(t *testing.T) { promql := fmt.Sprintf( `neuroncore_utilization_ratio{"@resource.k8s.cluster.name"="%s","@resource.k8s.pod.name"=~"neuron-burn-multi.*"}`, escaped) - burnResults, err := client.Query(ctx, promql) + burnResults, err := client.QueryComplete(ctx, promql) require.NoError(t, err, "querying neuroncore_utilization_ratio for neuron-burn-multi") require.Equal(t, 2, len(burnResults), fmt.Sprintf("Expected 2 cores correlated to neuron-burn-multi, got %d", len(burnResults))) diff --git a/test/otel/neuron/neuron_test.go b/test/otel/neuron/neuron_test.go index abe2e578e..9882d12b8 100644 --- a/test/otel/neuron/neuron_test.go +++ b/test/otel/neuron/neuron_test.go @@ -528,7 +528,7 @@ func TestNeuronNodeGroupCoverage(t *testing.T) { promql := fmt.Sprintf( `neuroncore_utilization_ratio{"@resource.k8s.cluster.name"="%s","@resource.host.type"="%s"}`, escaped, ng.InstanceType) - results, err := client.Query(context.Background(), promql) + results, err := client.QueryComplete(context.Background(), promql) require.NoError(t, err, "querying neuroncore_utilization_ratio on %s", ng.Description) require.True(t, len(results) > 0, "Neuron metrics missing from %s nodes (%s) — neuron-monitor not running?", diff --git a/test/otel/neuron/node_color_test.go b/test/otel/neuron/node_color_test.go index e033946e4..7d24db650 100644 --- a/test/otel/neuron/node_color_test.go +++ b/test/otel/neuron/node_color_test.go @@ -33,7 +33,7 @@ func TestNeuronHasRedNodeColor(t *testing.T) { promql := fmt.Sprintf( `%s{"@resource.k8s.cluster.name"="%s","@resource.%s"="red"}`, metricName, escaped, nodeColorLabel) - results, err := client.Query(context.Background(), promql) + results, err := client.QueryComplete(context.Background(), promql) require.NoError(t, err, "querying %s with node-color=red filter", metricName) require.True(t, len(results) > 0, "%s: no results from inf2.xlarge nodes (expected node-color=red)", metricName) @@ -55,7 +55,7 @@ func TestNeuronHasPurpleNodeColor(t *testing.T) { promql := fmt.Sprintf( `%s{"@resource.k8s.cluster.name"="%s","@resource.%s"="purple"}`, md.Name, escaped, nodeColorLabel) - results, err := client.Query(context.Background(), promql) + results, err := client.QueryComplete(context.Background(), promql) require.NoError(t, err, "querying %s with node-color=purple filter", md.Name) require.True(t, len(results) > 0, "%s: no results from inf2.24xlarge node (expected node-color=purple)", md.Name) diff --git a/test/otel/performance/performance_test.go b/test/otel/performance/performance_test.go index 3c1fa7be8..bd8798a3c 100644 --- a/test/otel/performance/performance_test.go +++ b/test/otel/performance/performance_test.go @@ -62,7 +62,7 @@ func getNodeAllocatable(t *testing.T, metricName string, resource string) (float // and skew the percent-of-node denominator the thresholds are calibrated to. query := fmt.Sprintf(`%s{"@resource.k8s.cluster.name"="%s", resource="%s"}`, metricName, otelmetrics.EscapePromQLValue(cfg.ClusterName), resource) - results, err := client.Query(context.Background(), query) + results, err := client.QueryComplete(context.Background(), query) require.NoError(t, err, "failed to query node allocatable for resource=%s", resource) require.NotEmpty(t, results, "no kube_node_status_allocatable data for resource=%s — are KSM metrics being collected?", resource) var sum float64 diff --git a/test/otel/standard/cadvisor_test.go b/test/otel/standard/cadvisor_test.go index 8e1a8ccdb..7d1d67ea1 100644 --- a/test/otel/standard/cadvisor_test.go +++ b/test/otel/standard/cadvisor_test.go @@ -138,7 +138,7 @@ func TestCadvisorNginxWorkloadLabels(t *testing.T) { t.Run(name, func(t *testing.T) { promql := fmt.Sprintf(`%s{"@resource.k8s.cluster.name"="%s","@resource.k8s.pod.name"=~"nginx-test.*"}`, name, cfg.ClusterName) - nginx, err := client.Query(context.Background(), promql) + nginx, err := client.QueryComplete(context.Background(), promql) require.NoError(t, err, "querying %s for nginx-test", name) require.True(t, len(nginx) > 0, "No %s from nginx-test pods", name) for _, r := range nginx { @@ -237,7 +237,7 @@ func TestCadvisorNodeGroupCoverage(t *testing.T) { promql := fmt.Sprintf( `container_memory_working_set_bytes{"@resource.k8s.cluster.name"="%s","@resource.host.type"="%s"}`, otelmetrics.EscapePromQLValue(cfg.ClusterName), ng.InstanceType) - results, err := client.Query(context.Background(), promql) + results, err := client.QueryComplete(context.Background(), promql) require.NoError(t, err, "querying cadvisor on %s", ng.Description) require.True(t, len(results) > 0, "cadvisor missing from %s (%s)", ng.Description, ng.InstanceType) }) diff --git a/test/otel/standard/dedup_test.go b/test/otel/standard/dedup_test.go index e5c95006d..62a7b69a5 100644 --- a/test/otel/standard/dedup_test.go +++ b/test/otel/standard/dedup_test.go @@ -29,7 +29,7 @@ func TestNodeExporterNoDuplicateSeries(t *testing.T) { `node_cpu_seconds_total{"@resource.k8s.cluster.name"="%s","@resource.host.type"="%s"}`, escaped, clusterHostTypes[0], ) - results, err := client.Query(ctx, promql) + results, err := client.QueryComplete(ctx, promql) require.NoError(t, err, "querying node_cpu_seconds_total") require.NotEmpty(t, results, "node_cpu_seconds_total not available") @@ -44,7 +44,7 @@ func TestNodeExporterNoDuplicateSeries(t *testing.T) { `node_cpu_seconds_total{"@resource.k8s.cluster.name"="%s","@resource.k8s.node.name"="%s",cpu="%s",mode="%s"}`, escaped, pinNode, pinCPU, pinMode, ) - pinned, err := client.Query(ctx, promql) + pinned, err := client.QueryComplete(ctx, promql) require.NoError(t, err, "querying pinned node_cpu_seconds_total") require.NotEmpty(t, pinned, "pinned node_cpu_seconds_total returned 0 results") require.Equal(t, 1, len(pinned), @@ -62,7 +62,7 @@ func TestCadvisorNoDuplicateSeries(t *testing.T) { `container_memory_working_set_bytes{"@resource.k8s.cluster.name"="%s","@resource.k8s.pod.name"=~"nginx-test.*"}`, escaped, ) - results, err := client.Query(ctx, promql) + results, err := client.QueryComplete(ctx, promql) require.NoError(t, err, "querying container_memory_working_set_bytes for nginx-test") require.NotEmpty(t, results, "container_memory_working_set_bytes not available for nginx-test pods") @@ -84,7 +84,7 @@ func TestCadvisorNoDuplicateSeries(t *testing.T) { `container_memory_working_set_bytes{"@resource.k8s.cluster.name"="%s","@resource.k8s.node.name"="%s","@resource.k8s.pod.name"="%s","@resource.k8s.container.name"="%s"}`, escaped, pinNode, pinPod, pinContainer, ) - pinned, err := client.Query(ctx, promql) + pinned, err := client.QueryComplete(ctx, promql) require.NoError(t, err, "querying pinned container_memory_working_set_bytes") require.NotEmpty(t, pinned, "pinned container_memory_working_set_bytes returned 0 results") require.Equal(t, 1, len(pinned), @@ -128,7 +128,7 @@ func TestKubeletstatsNoDuplicateSeries(t *testing.T) { `{"__name__"="k8s.node.cpu.usage","@resource.k8s.cluster.name"="%s","@resource.k8s.node.name"="%s"}`, escaped, pinNode, ) - pinned, err := client.Query(ctx, promql) + pinned, err := client.QueryComplete(ctx, promql) require.NoError(t, err, "querying pinned k8s.node.cpu.usage") require.NotEmpty(t, pinned, "pinned k8s.node.cpu.usage returned 0 results") require.Equal(t, 1, len(pinned), diff --git a/test/otel/standard/ksm_test.go b/test/otel/standard/ksm_test.go index d2ad15d88..45270585a 100644 --- a/test/otel/standard/ksm_test.go +++ b/test/otel/standard/ksm_test.go @@ -544,7 +544,7 @@ func TestKSM_PodBucket_NginxDeploymentName(t *testing.T) { ctx := context.Background() promql := fmt.Sprintf(`%s{"@resource.k8s.cluster.name"="%s","@resource.k8s.pod.name"=~"nginx-test.*"}`, metricName, cfg.ClusterName) - results, err := client.Query(ctx, promql) + results, err := client.QueryComplete(ctx, promql) require.NoError(t, err, "querying %s for nginx-test", metricName) require.True(t, len(results) > 0, "No %s results from nginx-test pods", metricName) for _, r := range results { @@ -643,7 +643,7 @@ func TestKSM_PodBucket_StatefulSetPodName(t *testing.T) { t.Parallel() promql := fmt.Sprintf(`%s{"@resource.k8s.cluster.name"="%s","@resource.k8s.pod.name"=~"ksm-test-statefulset.*"}`, metricName, cfg.ClusterName) - results, err := client.Query(context.Background(), promql) + results, err := client.QueryComplete(context.Background(), promql) require.NoError(t, err, "querying %s for ksm-test-statefulset", metricName) require.True(t, len(results) > 0, "No %s results from ksm-test-statefulset pods", metricName) for _, r := range results { @@ -665,7 +665,7 @@ func TestKSM_PodBucket_JobPodName(t *testing.T) { t.Parallel() promql := fmt.Sprintf(`%s{"@resource.k8s.cluster.name"="%s","@resource.k8s.pod.name"=~"ksm-test-job.*"}`, metricName, cfg.ClusterName) - results, err := client.Query(context.Background(), promql) + results, err := client.QueryComplete(context.Background(), promql) require.NoError(t, err, "querying %s for ksm-test-job", metricName) require.True(t, len(results) > 0, "No %s results from ksm-test-job pods", metricName) for _, r := range results { @@ -687,7 +687,7 @@ func TestKSM_PodBucket_CronJobPodName(t *testing.T) { t.Parallel() promql := fmt.Sprintf(`%s{"@resource.k8s.cluster.name"="%s","@resource.k8s.pod.name"=~"ksm-test-cronjob.*"}`, metricName, cfg.ClusterName) - results, err := client.Query(context.Background(), promql) + results, err := client.QueryComplete(context.Background(), promql) require.NoError(t, err, "querying %s for ksm-test-cronjob", metricName) if len(results) == 0 { t.Skipf("No %s results from ksm-test-cronjob pods (may have been cleaned up)", metricName) @@ -714,7 +714,7 @@ func TestKSM_PodBucket_ReplicaSetPodName(t *testing.T) { t.Parallel() promql := fmt.Sprintf(`%s{"@resource.k8s.cluster.name"="%s","@resource.k8s.pod.name"=~"ksm-test-replicaset.*"}`, metricName, cfg.ClusterName) - results, err := client.Query(context.Background(), promql) + results, err := client.QueryComplete(context.Background(), promql) require.NoError(t, err, "querying %s for ksm-test-replicaset", metricName) require.True(t, len(results) > 0, "No %s results from ksm-test-replicaset pods", metricName) for _, r := range results { @@ -1008,7 +1008,7 @@ func TestKSM_ContainerBucket_NginxDeploymentName(t *testing.T) { ctx := context.Background() promql := fmt.Sprintf(`%s{"@resource.k8s.cluster.name"="%s","@resource.k8s.pod.name"=~"nginx-test.*"}`, metricName, cfg.ClusterName) - results, err := client.Query(ctx, promql) + results, err := client.QueryComplete(ctx, promql) require.NoError(t, err, "querying %s for nginx-test", metricName) require.True(t, len(results) > 0, "No %s results from nginx-test pods", metricName) for _, r := range results { diff --git a/test/otel/standard/kubeletstats_test.go b/test/otel/standard/kubeletstats_test.go index 27a40860b..993a13543 100644 --- a/test/otel/standard/kubeletstats_test.go +++ b/test/otel/standard/kubeletstats_test.go @@ -247,7 +247,7 @@ func TestKubeletstatsNodeGroupCoverage(t *testing.T) { promql := fmt.Sprintf( `{"__name__"="k8s.node.cpu.usage","@resource.k8s.cluster.name"="%s","@resource.host.type"="%s"}`, otelmetrics.EscapePromQLValue(cfg.ClusterName), ng.InstanceType) - results, err := client.Query(context.Background(), promql) + results, err := client.QueryComplete(context.Background(), promql) require.NoError(t, err, "querying k8s.node.cpu.usage on %s", ng.Description) require.True(t, len(results) > 0, "kubeletstats missing from %s nodes (%s) — kubeletstatsreceiver not scraping?", diff --git a/test/otel/standard/labels_common_test.go b/test/otel/standard/labels_common_test.go index 8834acf3c..79ced4255 100644 --- a/test/otel/standard/labels_common_test.go +++ b/test/otel/standard/labels_common_test.go @@ -503,7 +503,7 @@ func TestWorkloadLabels(t *testing.T) { t.Run(metricName+"/cadvisor_nginx_workload", func(t *testing.T) { promql := fmt.Sprintf(`%s{"@resource.k8s.cluster.name"="%s","@resource.k8s.pod.name"=~"nginx-test.*"}`, metricName, cfg.ClusterName) - results, err := client.Query(context.Background(), promql) + results, err := client.QueryComplete(context.Background(), promql) require.NoError(t, err, "querying %s for nginx-test", metricName) require.True(t, len(results) > 0, "No %s results from nginx-test pods", metricName) for _, r := range results { @@ -607,7 +607,7 @@ func TestCustomNodeLabels(t *testing.T) { t.Run(metricName+"/has_blue", func(t *testing.T) { promql := fmt.Sprintf(`%s{"@resource.k8s.cluster.name"="%s","@resource.%s"="blue"}`, metricName, cfg.ClusterName, nodeColorLabel) - results, err := client.Query(context.Background(), promql) + results, err := client.QueryComplete(context.Background(), promql) require.NoError(t, err, "querying %s with blue filter", metricName) require.True(t, len(results) > 0, "%s no results from standard nodes (color=blue)", metricName) diff --git a/test/otel/standard/node_exporter_test.go b/test/otel/standard/node_exporter_test.go index 69db04781..105e519fc 100644 --- a/test/otel/standard/node_exporter_test.go +++ b/test/otel/standard/node_exporter_test.go @@ -137,7 +137,7 @@ func TestNodeExporterNodeGroupCoverage(t *testing.T) { promql := fmt.Sprintf( `node_load1{"@resource.k8s.cluster.name"="%s","@resource.host.type"="%s"}`, otelmetrics.EscapePromQLValue(cfg.ClusterName), ng.InstanceType) - results, err := client.Query(context.Background(), promql) + results, err := client.QueryComplete(context.Background(), promql) require.NoError(t, err, "querying node_load1 on %s", ng.Description) require.True(t, len(results) > 0, "node_exporter missing from %s nodes (%s) — DaemonSet not scheduling?", diff --git a/util/otelmetrics/client.go b/util/otelmetrics/client.go index c9315830b..6d308bdbe 100644 --- a/util/otelmetrics/client.go +++ b/util/otelmetrics/client.go @@ -12,6 +12,7 @@ import ( "net/http" "net/url" "strconv" + "strings" "time" "github.com/aws/aws-sdk-go-v2/aws" @@ -39,11 +40,13 @@ type OtelMetricsClient struct { region string signingService string maxRetries int + seriesLimit int // endpoint's per-query series cap; 0 means defaultSeriesLimit } type promqlResponse struct { - Status string `json:"status"` - Data promqlData `json:"data"` + Status string `json:"status"` + Data promqlData `json:"data"` + Warnings []string `json:"warnings"` } type promqlData struct { @@ -74,18 +77,43 @@ func NewClient(ctx context.Context, config TestConfig) (*OtelMetricsClient, erro }, nil } -// Query executes a PromQL instant query and returns parsed results. +// Query executes a single PromQL instant query and returns parsed results. The endpoint caps +// results at 500 series, often without saying so; use QueryComplete when every series matters. func (c *OtelMetricsClient) Query(ctx context.Context, promql string) ([]MetricResult, error) { + results, truncated, err := c.queryAt(ctx, promql, "") + if truncated { + log.Printf("WARNING: result truncated by the endpoint series limit query=%s", promql) + } + return results, err +} + +// queryAt runs one instant query, evaluated at evalTime when non-empty. It also returns +// whether the endpoint warned of truncation. +func (c *OtelMetricsClient) queryAt(ctx context.Context, promql, evalTime string) ([]MetricResult, bool, error) { params := url.Values{"query": {promql}} + if evalTime != "" { + params.Set("time", evalTime) + } log.Printf("querying promql=%s", promql) raw, err := c.requestWithRetry(ctx, c.queryURL, params) if err != nil { - return nil, err + return nil, false, err } results := c.parseResponse(raw) log.Printf("query returned count=%d", len(results)) - return results, nil + return results, hasTruncationWarning(raw.Warnings), nil +} + +// hasTruncationWarning reports the endpoint's "results truncated due to series limit" warning. +// Aggregations carry it reliably; plain selectors only sometimes, so it is never proof of completeness. +func hasTruncationWarning(warnings []string) bool { + for _, w := range warnings { + if strings.Contains(strings.ToLower(w), "truncated") { + return true + } + } + return false } // MetricNames returns the distinct metric names present for the given PromQL matcher over [start, end]. @@ -143,8 +171,13 @@ func (c *OtelMetricsClient) requestRawWithRetry(ctx context.Context, baseURL str return nil, fmt.Errorf("signing request: %w", err) } + release, err := acquireRequestSlot(ctx) + if err != nil { + return nil, err + } resp, err := c.httpClient.Do(req) if err != nil { + release() lastErr = err log.Printf("request failed attempt=%d error=%v", attempt+1, err) if attempt < c.maxRetries-1 { @@ -155,6 +188,7 @@ func (c *OtelMetricsClient) requestRawWithRetry(ctx context.Context, baseURL str body, readErr := io.ReadAll(resp.Body) resp.Body.Close() + release() if readErr != nil { return nil, fmt.Errorf("reading response: %w", readErr) } @@ -164,7 +198,7 @@ func (c *OtelMetricsClient) requestRawWithRetry(ctx context.Context, baseURL str lastErr = fmt.Errorf("server error %d: %s", resp.StatusCode, truncate(string(body), 200)) log.Printf("server error, retrying status=%d attempt=%d", resp.StatusCode, attempt+1) if attempt < c.maxRetries-1 { - time.Sleep(time.Duration(1<= 0 { + d = time.Duration(secs) * time.Second + } else if at, err := http.ParseTime(retryAfter); err == nil { + d = at.Sub(now) + } + if d < 0 { + d = 0 + } + if d > maxRetryAfter { + d = maxRetryAfter + } + return d +} + +type requestSlotsKey struct{} + +// withRequestSlots makes every request issued under ctx hold one slot of sem while it is in flight. +func withRequestSlots(ctx context.Context, sem chan struct{}) context.Context { + return context.WithValue(ctx, requestSlotsKey{}, sem) +} + +func acquireRequestSlot(ctx context.Context) (release func(), err error) { + sem, _ := ctx.Value(requestSlotsKey{}).(chan struct{}) + if sem == nil { + return func() {}, nil + } + select { + case sem <- struct{}{}: + return func() { <-sem }, nil + case <-ctx.Done(): + return nil, ctx.Err() + } +} + func (c *OtelMetricsClient) parseResponse(response *promqlResponse) []MetricResult { var results []MetricResult for _, series := range response.Data.Result { @@ -242,8 +316,9 @@ type RangeResult struct { } type promqlRangeResponse struct { - Status string `json:"status"` - Data promqlRangeData `json:"data"` + Status string `json:"status"` + Data promqlRangeData `json:"data"` + Warnings []string `json:"warnings"` } type promqlRangeData struct { @@ -275,6 +350,9 @@ func (c *OtelMetricsClient) QueryRange(ctx context.Context, promql string, start if err := json.Unmarshal(rawBytes, &rangeResp); err != nil { return nil, fmt.Errorf("parsing range response: %w", err) } + if hasTruncationWarning(rangeResp.Warnings) { + log.Printf("WARNING: range result truncated by the endpoint series limit query=%s", promql) + } var results []RangeResult for _, series := range rangeResp.Data.Result { diff --git a/util/otelmetrics/client_test.go b/util/otelmetrics/client_test.go new file mode 100644 index 000000000..811141381 --- /dev/null +++ b/util/otelmetrics/client_test.go @@ -0,0 +1,58 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: MIT + +package otelmetrics + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestRetryDelay(t *testing.T) { + now := time.Date(2026, 10, 15, 12, 0, 0, 0, time.UTC) + for _, tc := range []struct { + header string + attempt int + want time.Duration + }{ + {"", 0, time.Second}, + {"", 2, 4 * time.Second}, + {"not-a-delay", 1, 2 * time.Second}, + {"0", 2, 0}, + {"5", 0, 5 * time.Second}, + {"120", 0, maxRetryAfter}, + {now.Add(3 * time.Second).Format(http.TimeFormat), 0, 3 * time.Second}, + {now.Add(-time.Minute).Format(http.TimeFormat), 0, 0}, + } { + assert.Equal(t, tc.want, retryDelay(tc.header, tc.attempt, now), "Retry-After %q attempt %d", tc.header, tc.attempt) + } +} + +func TestQuery_ThrottledRequestWaitsForRetryAfter(t *testing.T) { + var calls int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if atomic.AddInt32(&calls, 1) == 1 { + w.Header().Set("Retry-After", "0") + http.Error(w, "throttled", http.StatusTooManyRequests) + return + } + fmt.Fprint(w, `{"status":"success","data":{"resultType":"vector","result":[]}}`) + })) + defer srv.Close() + c := newTestClient(srv.URL) + c.maxRetries = 2 + + start := time.Now() + _, err := c.Query(context.Background(), "up") + require.NoError(t, err) + assert.EqualValues(t, 2, atomic.LoadInt32(&calls)) + assert.Less(t, time.Since(start), 500*time.Millisecond, "Retry-After: 0 replaces the 1 s backoff") +} diff --git a/util/otelmetrics/complete.go b/util/otelmetrics/complete.go new file mode 100644 index 000000000..1fae4ccec --- /dev/null +++ b/util/otelmetrics/complete.go @@ -0,0 +1,370 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: MIT + +package otelmetrics + +import ( + "context" + "errors" + "fmt" + "log" + "regexp" + "sort" + "strconv" + "strings" + "sync" + "time" +) + +// ErrTruncated reports that the endpoint dropped series to stay under its per-query +// series limit and the complete result could not be recovered. +var ErrTruncated = errors.New("query result truncated by the endpoint series limit") + +const ( + // defaultSeriesLimit is the endpoint's per-query cap ("series limit (500)" in its warnings). + defaultSeriesLimit = 500 + // maxMatcherValueLen is the endpoint's limit on one matcher value, which bounds a shard's regex. + maxMatcherValueLen = 1000 + sampleSize = 50 + maxShardDepth = 4 + maxSplitCandidates = 3 + shardConcurrency = 4 + nodeNameLabel = "@resource.k8s.node.name" + // evalTimeLag keeps the evaluation time that the client picks behind the endpoint's clock + // and its newest samples. + evalTimeLag = 5 * time.Second +) + +var metricNameRE = regexp.MustCompile(`^[a-zA-Z_:][a-zA-Z0-9_:]*$`) + +// QueryComplete returns every series matched by an instant vector selector. The endpoint caps +// selector results and does not reliably say so, so the result is compared with count() at the +// same evaluation time and, only when short, re-fetched as disjoint shards (by node first). +// The client picks that time, so the selector and its count run at once. +// Anything that is not a plain vector selector is passed to Query unchanged. +func (c *OtelMetricsClient) QueryComplete(ctx context.Context, selector string) ([]MetricResult, error) { + if !isVectorSelector(selector) { + return c.Query(ctx, selector) + } + evalTime := evalTimeAt(time.Now()) + var before int + var countErr error + counted := make(chan struct{}) + go func() { + defer close(counted) + before, countErr = c.countAt(ctx, selector, evalTime) + }() + results, _, err := c.queryAt(ctx, selector, evalTime) + <-counted + if err != nil { + return nil, err + } + if countErr != nil { + return nil, countErr + } + if len(results) >= before { + return results, nil + } + log.Printf("result truncated: got %d of %d series, re-fetching in shards query=%s", len(results), before, selector) + + all, err := c.split(ctx, selector, evalTime, before, results, nil, 1) + if err != nil { + return nil, err + } + after, err := c.countAt(ctx, selector, evalTime) + if err != nil { + return nil, err + } + lo, hi := before, after + if lo > hi { + lo, hi = hi, lo + } + if len(all) < lo || len(all) > hi { + return nil, fmt.Errorf("%w: shards returned %d series but count() was %d before and %d after (query %s)", + ErrTruncated, len(all), before, after, selector) + } + log.Printf("sharded fetch complete: %d series query=%s", len(all), selector) + return all, nil +} + +type shardGroup struct { + value string // "" when the label is absent + count int +} + +type shard struct { + matcher string + want int + exact bool // the shard pins the label to one value (or to absent) +} + +// evalTimeAt is the evaluation time, in Unix seconds, that QueryComplete pins its queries to. +func evalTimeAt(now time.Time) string { + return strconv.FormatInt(now.Add(-evalTimeLag).Unix(), 10) +} + +// countAt returns count(selector) at evalTime. +func (c *OtelMetricsClient) countAt(ctx context.Context, selector, evalTime string) (int, error) { + groups, err := c.countBy(ctx, selector, "", evalTime) + if err != nil { + return 0, err + } + return groupTotal(groups), nil +} + +func groupTotal(groups []shardGroup) int { + total := 0 + for _, g := range groups { + total += g.count + } + return total +} + +// countBy returns the series count per value of label (or the total when label is ""). +// It fails with ErrTruncated when the label has too many values to list in one response. +func (c *OtelMetricsClient) countBy(ctx context.Context, selector, label, evalTime string) ([]shardGroup, error) { + promql := "count(" + selector + ")" + if label != "" { + promql = fmt.Sprintf("count by (%s)(%s)", quoteLabel(label), selector) + } + results, truncated, err := c.queryAt(ctx, promql, evalTime) + if err != nil { + return nil, err + } + if truncated { + return nil, fmt.Errorf("%w: too many groups (query %s)", ErrTruncated, promql) + } + groups := make([]shardGroup, 0, len(results)) + for _, r := range results { + g := shardGroup{count: int(r.Value)} + if label != "" { + g.value = r.Labels.AllLabels()[label] + } + groups = append(groups, g) + } + return groups, nil +} + +// split re-fetches a truncated selector of want series as disjoint shards over the first label that divides it. +func (c *OtelMetricsClient) split(ctx context.Context, selector, evalTime string, want int, sample []MetricResult, pinned map[string]bool, depth int) ([]MetricResult, error) { + if depth > maxShardDepth { + return nil, fmt.Errorf("%w: still truncated after %d shard levels (query %s)", ErrTruncated, maxShardDepth, selector) + } + if len(sample) == 0 { + // Stale series filled the cap; topk evaluates fully, so it yields live series to inspect. + sample, _, _ = c.queryAt(ctx, fmt.Sprintf("topk(%d,%s)", sampleSize, selector), evalTime) + } + for _, label := range splitCandidates(sample, pinned) { + groups, err := c.countBy(ctx, selector, label, evalTime) + if errors.Is(err, ErrTruncated) { + continue + } + if err != nil { + return nil, err + } + if len(groups) < 2 || groupTotal(groups) < want { + continue // label does not divide the set, or its groups were cut short + } + shards := packShards(label, groups, c.shardTarget(depth)) + log.Printf("fetching %d shards by %s depth=%d query=%s", len(shards), label, depth, selector) + return c.fetchShards(ctx, selector, evalTime, label, shards, pinned, depth) + } + return nil, fmt.Errorf("%w: no label divides the result further (query %s)", ErrTruncated, selector) +} + +func (c *OtelMetricsClient) fetchShards(ctx context.Context, selector, evalTime, label string, shards []shard, pinned map[string]bool, depth int) ([]MetricResult, error) { + type outcome struct { + results []MetricResult + err error + } + outcomes := make([]outcome, len(shards)) + sem := make(chan struct{}, shardConcurrency) + var wg sync.WaitGroup + for i := range shards { + wg.Add(1) + go func(i int) { + defer wg.Done() + sem <- struct{}{} + defer func() { <-sem }() + sh := shards[i] + sub := withMatcher(selector, sh.matcher) + results, _, err := c.queryAt(ctx, sub, evalTime) + if err == nil && len(results) < sh.want { + subPinned := pinned + if sh.exact { + subPinned = withPinned(pinned, label) + } + results, err = c.split(ctx, sub, evalTime, sh.want, results, subPinned, depth+1) + } + outcomes[i] = outcome{results: results, err: err} + }(i) + } + wg.Wait() + + var all []MetricResult + for _, o := range outcomes { + if o.err != nil { + return nil, o.err + } + all = append(all, o.results...) + } + return all, nil +} + +// shardTarget is the series budget per shard: 80% of the cap, since stale series also count +// against it, halved at each level so a re-split always yields smaller shards. +func (c *OtelMetricsClient) shardTarget(depth int) int { + limit := c.seriesLimit + if limit <= 0 { + limit = defaultSeriesLimit + } + return (limit * 4 / 5) >> (depth - 1) +} + +// splitCandidates orders the labels worth splitting on: node name first, then the sample's +// labels by number of distinct values. Labels already pinned to one value are skipped. +func splitCandidates(sample []MetricResult, pinned map[string]bool) []string { + distinct := make(map[string]map[string]struct{}) + for _, r := range sample { + for k, v := range r.Labels.AllLabels() { + if distinct[k] == nil { + distinct[k] = make(map[string]struct{}) + } + distinct[k][v] = struct{}{} + } + } + var out []string + if !pinned[nodeNameLabel] && (len(sample) == 0 || distinct[nodeNameLabel] != nil) { + out = append(out, nodeNameLabel) + } + var keys []string + for k, values := range distinct { + if k != nodeNameLabel && !pinned[k] && len(values) > 1 { + keys = append(keys, k) + } + } + sort.Slice(keys, func(i, j int) bool { + if ni, nj := len(distinct[keys[i]]), len(distinct[keys[j]]); ni != nj { + return ni > nj + } + return keys[i] < keys[j] + }) + out = append(out, keys...) + if len(out) > maxSplitCandidates { + out = out[:maxSplitCandidates] + } + return out +} + +// packShards bins label values into shards of at most target series. The absent value gets +// its own !~".+" shard because ="" does not match a missing label on this endpoint. +func packShards(label string, groups []shardGroup, target int) []shard { + if target < 1 { + target = 1 + } + sorted := append([]shardGroup(nil), groups...) + sort.Slice(sorted, func(i, j int) bool { + if sorted[i].count != sorted[j].count { + return sorted[i].count > sorted[j].count + } + return sorted[i].value < sorted[j].value + }) + + var out []shard + type bin struct { + values []string + want, chars int + } + var bins []bin + for _, g := range sorted { + if g.value == "" { + out = append(out, shard{matcher: quoteLabel(label) + `!~".+"`, want: g.count, exact: true}) + continue + } + chars := len(regexp.QuoteMeta(g.value)) + placed := false + for i := range bins { + if bins[i].want+g.count <= target && bins[i].chars+1+chars <= maxMatcherValueLen { + bins[i].values = append(bins[i].values, g.value) + bins[i].want += g.count + bins[i].chars += 1 + chars + placed = true + break + } + } + if !placed { + bins = append(bins, bin{values: []string{g.value}, want: g.count, chars: chars}) + } + } + for _, b := range bins { + out = append(out, shard{matcher: valueMatcher(label, b.values), want: b.want, exact: len(b.values) == 1}) + } + return out +} + +func valueMatcher(label string, values []string) string { + if len(values) == 1 { + return fmt.Sprintf(`%s="%s"`, quoteLabel(label), promqlEscaper.Replace(values[0])) + } + alts := make([]string, len(values)) + for i, v := range values { + alts[i] = regexp.QuoteMeta(v) + } + sort.Strings(alts) + return fmt.Sprintf(`%s=~"%s"`, quoteLabel(label), promqlEscaper.Replace(strings.Join(alts, "|"))) +} + +func quoteLabel(label string) string { + return `"` + promqlEscaper.Replace(label) + `"` +} + +func withPinned(pinned map[string]bool, label string) map[string]bool { + out := make(map[string]bool, len(pinned)+1) + for k := range pinned { + out[k] = true + } + out[label] = true + return out +} + +// withMatcher adds one label matcher to a vector selector. +func withMatcher(selector, matcher string) string { + s := strings.TrimSpace(selector) + if !strings.HasSuffix(s, "}") { + return s + "{" + matcher + "}" + } + body := strings.TrimSpace(s[:len(s)-1]) + if strings.HasSuffix(body, "{") || strings.HasSuffix(body, ",") { + return body + matcher + "}" + } + return body + "," + matcher + "}" +} + +// isVectorSelector reports whether s is a single instant vector selector (name, name{...} +// or {...}) with no range, offset, function or operator around it. +func isVectorSelector(s string) bool { + s = strings.TrimSpace(s) + open := strings.IndexByte(s, '{') + if open < 0 { + return metricNameRE.MatchString(s) + } + if name := strings.TrimSpace(s[:open]); name != "" && !metricNameRE.MatchString(name) { + return false + } + var quote byte + for i := open + 1; i < len(s); i++ { + ch := s[i] + switch { + case quote != 0 && ch == '\\' && quote != '`': + i++ + case quote != 0 && ch == quote: + quote = 0 + case quote != 0: + case ch == '"' || ch == '\'' || ch == '`': + quote = ch + case ch == '}': + return i == len(s)-1 + } + } + return false +} diff --git a/util/otelmetrics/complete_test.go b/util/otelmetrics/complete_test.go new file mode 100644 index 000000000..816710928 --- /dev/null +++ b/util/otelmetrics/complete_test.go @@ -0,0 +1,670 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: MIT + +package otelmetrics + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "net/http/httptest" + "regexp" + "sort" + "strconv" + "strings" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const fakeEvalTime = "1700000000.123" + +type fakeSeries struct { + labels map[string]string + stale bool // occupies the cap like a recently ended series, but has no sample +} + +type fakeRequest struct { + query, time string +} + +// fakePromQL mimics the endpoint: plain selectors are capped at limit series (stale ones +// included), aggregations warn when capped, and `=""` never matches an absent label. +type fakePromQL struct { + t *testing.T + series []fakeSeries + limit int + countOffset int // added to count() without by, to simulate an inconsistent endpoint + brokenNotRegex bool // make !~ match everything, so shards overlap + mu sync.Mutex + requests []fakeRequest +} + +func (f *fakePromQL) handler(w http.ResponseWriter, r *http.Request) { + q := r.URL.Query() + query, ts := q.Get("query"), q.Get("time") + f.mu.Lock() + f.requests = append(f.requests, fakeRequest{query: query, time: ts}) + f.mu.Unlock() + if ts == "" { + ts = fakeEvalTime + } + + type result struct { + Metric map[string]string `json:"metric"` + Value []json.RawMessage `json:"value"` + } + resp := struct { + Status string `json:"status"` + Warnings []string `json:"warnings,omitempty"` + Data struct { + ResultType string `json:"resultType"` + Result []result `json:"result"` + } `json:"data"` + }{Status: "success"} + resp.Data.ResultType = "vector" + sample := func(metric map[string]string, v int) result { + return result{Metric: metric, Value: []json.RawMessage{json.RawMessage(ts), json.RawMessage(strconv.Quote(strconv.Itoa(v)))}} + } + + switch { + case strings.HasPrefix(query, "topk("): + m := regexp.MustCompile(`^topk\((\d+),(.*)\)$`).FindStringSubmatch(query) + if m == nil { + f.badRequest(w, "unexpected topk query %q", query) + return + } + k, _ := strconv.Atoi(m[1]) + for _, s := range f.match(m[2]) { + if !s.stale && len(resp.Data.Result) < k { + resp.Data.Result = append(resp.Data.Result, sample(s.labels, 1)) + } + } + case strings.HasPrefix(query, "count by ("): + label, inner, ok := parseCountBy(query) + if !ok { + f.badRequest(w, "unexpected count by query %q", query) + return + } + groups := map[string]int{} + for _, s := range f.match(inner) { + if !s.stale { + groups[s.labels[label]]++ + } + } + keys := make([]string, 0, len(groups)) + for k := range groups { + keys = append(keys, k) + } + sort.Strings(keys) + if len(keys) > f.limit { + keys = keys[:f.limit] + resp.Warnings = []string{fmt.Sprintf("results truncated due to series limit (%d).", f.limit)} + } + for _, k := range keys { + m := map[string]string{} + if k != "" { + m[label] = k + } + resp.Data.Result = append(resp.Data.Result, sample(m, groups[k])) + } + case strings.HasPrefix(query, "count("): + n := 0 + for _, s := range f.match(query[len("count(") : len(query)-1]) { + if !s.stale { + n++ + } + } + if n > 0 { + resp.Data.Result = append(resp.Data.Result, sample(map[string]string{}, n+f.countOffset)) + } + default: + matched := f.match(query) + capped := len(matched) > f.limit + if capped { + matched = matched[:f.limit] + } + for _, s := range matched { + if !s.stale { + resp.Data.Result = append(resp.Data.Result, sample(s.labels, 1)) + } + } + // The endpoint warns only when the capped output is full, so stale series hide it. + if capped && len(resp.Data.Result) == f.limit { + resp.Warnings = []string{fmt.Sprintf("results truncated due to series limit (%d).", f.limit)} + } + } + if err := json.NewEncoder(w).Encode(resp); err != nil { + f.t.Errorf("encoding response: %v", err) + } +} + +// badRequest reports a query the fake cannot parse; the handler runs off the test goroutine. +func (f *fakePromQL) badRequest(w http.ResponseWriter, format string, args ...interface{}) { + f.t.Errorf(format, args...) + http.Error(w, fmt.Sprintf(format, args...), http.StatusBadRequest) +} + +func (f *fakePromQL) match(selector string) []fakeSeries { + matchers, ok := parseFakeSelector(selector) + if !ok { + f.t.Errorf("unparseable selector %q", selector) + return nil + } + var out []fakeSeries + for _, s := range f.series { + ok := true + for _, m := range matchers { + v, present := s.labels[m.key] + switch m.op { + case "=": + ok = present && v == m.val + case "=~": + ok = regexp.MustCompile("^(?:" + m.val + ")$").MatchString(v) + case "!~": + ok = f.brokenNotRegex || !regexp.MustCompile("^(?:"+m.val+")$").MatchString(v) + default: + f.t.Errorf("unsupported matcher op %q", m.op) + ok = false + } + if !ok { + break + } + } + if ok { + out = append(out, s) + } + } + return out +} + +func (f *fakePromQL) queries() []fakeRequest { + f.mu.Lock() + defer f.mu.Unlock() + return append([]fakeRequest(nil), f.requests...) +} + +type fakeMatcher struct{ key, op, val string } + +func parseCountBy(q string) (string, string, bool) { + m := regexp.MustCompile(`^count by \("((?:[^"\\]|\\.)*)"\)\((.*)\)$`).FindStringSubmatch(q) + if m == nil { + return "", "", false + } + return unescape(m[1]), m[2], true +} + +func parseFakeSelector(s string) ([]fakeMatcher, bool) { + open := strings.IndexByte(s, '{') + if open < 0 { + return []fakeMatcher{{"__name__", "=", s}}, true + } + var out []fakeMatcher + if open > 0 { + out = append(out, fakeMatcher{"__name__", "=", s[:open]}) + } + re := regexp.MustCompile(`^\s*(?:"((?:[^"\\]|\\.)*)"|([a-zA-Z_][a-zA-Z0-9_]*))\s*(=~|!~|=)\s*"((?:[^"\\]|\\.)*)"\s*,?`) + body := s[open+1 : len(s)-1] + for strings.TrimSpace(body) != "" { + m := re.FindStringSubmatch(body) + if m == nil { + return nil, false + } + key := m[2] + if m[1] != "" { + key = unescape(m[1]) + } + out = append(out, fakeMatcher{key, m[3], unescape(m[4])}) + body = body[len(m[0]):] + } + return out, true +} + +func unescape(s string) string { + return strings.NewReplacer(`\\`, `\`, `\"`, `"`).Replace(s) +} + +func newFake(t *testing.T, limit int, series []fakeSeries) (*fakePromQL, *OtelMetricsClient) { + f := &fakePromQL{t: t, series: series, limit: limit} + srv := httptest.NewServer(http.HandlerFunc(f.handler)) + t.Cleanup(srv.Close) + c := newTestClient(srv.URL) + c.seriesLimit = limit + return f, c +} + +// podSeries builds kube_pod_info-like series: perNode pods on each node, plus extra. +func podSeries(nodes, perNode int) []fakeSeries { + var out []fakeSeries + for n := 0; n < nodes; n++ { + for p := 0; p < perNode; p++ { + out = append(out, fakeSeries{labels: map[string]string{ + "__name__": "kube_pod_info", + "@resource.k8s.cluster.name": "c1", + "@resource.k8s.node.name": fmt.Sprintf("ip-10-0-%d-1.ec2.internal", n), + "@resource.k8s.pod.name": fmt.Sprintf("pod-%d-%d", n, p), + "@resource.k8s.namespace.name": "default", + "@instrumentation.@name": "ksm", + "@aws.account": "123456789012", + "__type__": "Gauge", + }}) + } + } + return out +} + +const podSelector = `kube_pod_info{"@resource.k8s.cluster.name"="c1"}` + +func podNames(results []MetricResult) map[string]bool { + out := map[string]bool{} + for _, r := range results { + out[r.Labels.Resource["k8s.pod.name"]] = true + } + return out +} + +func countQueries(reqs []fakeRequest, pred func(string) bool) int { + n := 0 + for _, r := range reqs { + if pred(r.query) { + n++ + } + } + return n +} + +func isPlain(q string) bool { return !strings.HasPrefix(q, "count") } + +// assertOneEvalTime checks that the client pinned every request to the same evaluation time. +func assertOneEvalTime(t *testing.T, reqs []fakeRequest) { + t.Helper() + require.NotEmpty(t, reqs) + require.NotEmpty(t, reqs[0].time, "the client must pick the evaluation time: %s", reqs[0].query) + for _, r := range reqs { + assert.Equal(t, reqs[0].time, r.time, "every query is pinned to one evaluation time: %s", r.query) + } +} + +func TestQueryComplete_NotTruncatedIsOneDataQuery(t *testing.T) { + f, c := newFake(t, 10, podSeries(2, 3)) + + results, err := c.QueryComplete(context.Background(), podSelector) + require.NoError(t, err) + assert.Len(t, results, 6) + + reqs := f.queries() + require.Len(t, reqs, 2, "expected the selector plus one count(): %+v", reqs) + assert.ElementsMatch(t, []string{podSelector, "count(" + podSelector + ")"}, []string{reqs[0].query, reqs[1].query}) + assertOneEvalTime(t, reqs) +} + +func TestQueryComplete_SelectorAndCountRunTogether(t *testing.T) { + f := &fakePromQL{t: t, series: podSeries(2, 3), limit: 10} + var mu sync.Mutex + arrived := 0 + both := make(chan struct{}) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + if arrived++; arrived == 2 { + close(both) + } + mu.Unlock() + // Answer only once both requests are in flight, which a client that waits for one cannot do. + select { + case <-both: + case <-time.After(2 * time.Second): + t.Errorf("request %q was not sent together with the other one", r.URL.Query().Get("query")) + } + f.handler(w, r) + })) + t.Cleanup(srv.Close) + c := newTestClient(srv.URL) + c.seriesLimit = 10 + + results, err := c.QueryComplete(context.Background(), podSelector) + require.NoError(t, err) + assert.Len(t, results, 6) + assertOneEvalTime(t, f.queries()) +} + +func TestEvalTimeAt(t *testing.T) { + assert.Equal(t, "1700000005", evalTimeAt(time.Unix(1700000010, 900e6)), "Unix seconds, evalTimeLag in the past") +} + +func TestQueryComplete_TruncatedShardsByNode(t *testing.T) { + series := podSeries(6, 5) + series[0].stale, series[7].stale = true, true // stale series still use up the cap + f, c := newFake(t, 10, series) + + results, err := c.QueryComplete(context.Background(), podSelector) + require.NoError(t, err) + assert.Len(t, results, 28) + assert.Len(t, podNames(results), 28, "every live pod exactly once") + + reqs := f.queries() + assert.Equal(t, 1, countQueries(reqs, func(q string) bool { + return q == `count by ("@resource.k8s.node.name")(`+podSelector+")" + })) + shards := countQueries(reqs, func(q string) bool { return isPlain(q) && strings.Contains(q, `"@resource.k8s.node.name"`) }) + assert.GreaterOrEqual(t, shards, 4, "28 series under a cap of 10 need at least 4 shards") + assert.Equal(t, 2, countQueries(reqs, func(q string) bool { return q == "count("+podSelector+")" }), "count() before and after") + assertOneEvalTime(t, reqs) +} + +func TestQueryComplete_AbsentNodeShardUsesNotRegex(t *testing.T) { + series := podSeries(3, 4) + for i := 0; i < 3; i++ { + series = append(series, fakeSeries{labels: map[string]string{ + "__name__": "kube_pod_info", + "@resource.k8s.cluster.name": "c1", + "@resource.k8s.pod.name": fmt.Sprintf("pending-%d", i), + }}) + } + f, c := newFake(t, 5, series) + + results, err := c.QueryComplete(context.Background(), podSelector) + require.NoError(t, err) + assert.Len(t, results, 15) + names := podNames(results) + for i := 0; i < 3; i++ { + assert.True(t, names[fmt.Sprintf("pending-%d", i)], "series without a node must be returned") + } + + reqs := f.queries() + assert.Equal(t, 1, countQueries(reqs, func(q string) bool { + return strings.Contains(q, `"@resource.k8s.node.name"!~".+"`) + })) + assert.Zero(t, countQueries(reqs, func(q string) bool { + return strings.Contains(q, `"@resource.k8s.node.name"=""`) + }), `="" does not match an absent label on this endpoint`) +} + +func TestQueryComplete_SingleNodeOverLimitSplitsFurther(t *testing.T) { + var series []fakeSeries + for cpu := 0; cpu < 8; cpu++ { + for _, mode := range []string{"idle", "user", "system"} { + series = append(series, fakeSeries{labels: map[string]string{ + "__name__": "node_cpu_seconds_total", + "@resource.k8s.cluster.name": "c1", + "@resource.k8s.node.name": "big-node", + "cpu": strconv.Itoa(cpu), + "mode": mode, + }}) + } + } + f, c := newFake(t, 10, series) + + results, err := c.QueryComplete(context.Background(), `node_cpu_seconds_total{"@resource.k8s.cluster.name"="c1"}`) + require.NoError(t, err) + assert.Len(t, results, 24) + assert.Positive(t, countQueries(f.queries(), func(q string) bool { return strings.HasPrefix(q, `count by ("cpu")`) }), + "one node holds every series, so the split must fall through to the cpu label") +} + +func TestQueryComplete_SkipsLabelWithTooManyValues(t *testing.T) { + var series []fakeSeries + for i := 0; i < 20; i++ { + series = append(series, fakeSeries{labels: map[string]string{ + "__name__": "m", + "id": strconv.Itoa(i), + "zone": fmt.Sprintf("z%d", i%3), + }}) + } + f, c := newFake(t, 10, series) + + results, err := c.QueryComplete(context.Background(), `m{}`) + require.NoError(t, err) + assert.Len(t, results, 20) + reqs := f.queries() + assert.Equal(t, 1, countQueries(reqs, func(q string) bool { return strings.HasPrefix(q, `count by ("id")`) })) + assert.Equal(t, 3, countQueries(reqs, func(q string) bool { return isPlain(q) && strings.Contains(q, `"zone"="z`) })) +} + +func TestQueryComplete_InconsistentCountIsError(t *testing.T) { + f, c := newFake(t, 10, podSeries(6, 5)) + f.countOffset = 3 // count() claims 33, but every split only accounts for 30 + + _, err := c.QueryComplete(context.Background(), podSelector) + require.Error(t, err) + assert.True(t, errors.Is(err, ErrTruncated), "got %v", err) +} + +func TestQueryComplete_ShardTotalMismatchIsError(t *testing.T) { + series := podSeries(3, 4) + for i := 0; i < 3; i++ { + series = append(series, fakeSeries{labels: map[string]string{ + "__name__": "kube_pod_info", + "@resource.k8s.cluster.name": "c1", + "@resource.k8s.pod.name": fmt.Sprintf("pending-%d", i), + }}) + } + f, c := newFake(t, 5, series) + f.brokenNotRegex = true // the absent-node shard overlaps the node shards + + _, err := c.QueryComplete(context.Background(), podSelector) + require.Error(t, err) + assert.True(t, errors.Is(err, ErrTruncated), "got %v", err) + assert.Contains(t, err.Error(), "shards returned 17 series but count() was 15 before and 15 after") +} + +func TestQueryComplete_UnsplittableIsError(t *testing.T) { + var series []fakeSeries + for i := 0; i < 20; i++ { + series = append(series, fakeSeries{labels: map[string]string{"__name__": "m", "id": strconv.Itoa(i)}}) + } + _, c := newFake(t, 10, series) + // id has 20 values, so its count by is capped too; nothing else varies. + _, err := c.QueryComplete(context.Background(), `m{}`) + require.Error(t, err) + assert.True(t, errors.Is(err, ErrTruncated), "got %v", err) +} + +func TestQueryComplete_EmptyResultWithLiveSeries(t *testing.T) { + series := podSeries(2, 3) + stale := make([]fakeSeries, 0, 5) + for i := 0; i < 5; i++ { + s := podSeries(1, 1)[0] + s.labels["@resource.k8s.pod.name"] = fmt.Sprintf("done-%d", i) + s.stale = true + stale = append(stale, s) + } + f, c := newFake(t, 5, append(stale, series...)) + + results, err := c.QueryComplete(context.Background(), podSelector) + require.NoError(t, err) + assert.Len(t, results, 6, "the first query sees only stale series, yet six are live") + assertOneEvalTime(t, f.queries()) +} + +func TestQueryComplete_NonSelectorPassesThrough(t *testing.T) { + f, c := newFake(t, 10, podSeries(1, 2)) + _, err := c.QueryComplete(context.Background(), "count("+podSelector+")") + require.NoError(t, err) + assert.Len(t, f.queries(), 1) +} + +func TestQueryComplete_WarnedSelectorStillShards(t *testing.T) { + f, c := newFake(t, 10, podSeries(4, 5)) + + results, err := c.QueryComplete(context.Background(), podSelector) + require.NoError(t, err) + assert.Len(t, results, 20) + assert.Positive(t, countQueries(f.queries(), func(q string) bool { return strings.HasPrefix(q, "count by (") })) +} + +func TestQuery_TruncationWarningKeepsResults(t *testing.T) { + _, c := newFake(t, 3, podSeries(2, 3)) + results, err := c.Query(context.Background(), podSelector) + require.NoError(t, err, "Query stays a single best-effort request") + assert.Len(t, results, 3) + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + fmt.Fprint(w, `{"status":"success","warnings":["results truncated due to series limit (500)."],"data":{"resultType":"matrix","result":[{"metric":{"a":"b"},"values":[[1,"1"]]}]}}`) + })) + defer srv.Close() + rr, err := newTestClient(srv.URL).QueryRange(context.Background(), "up", time.Unix(0, 0), time.Unix(60, 0), time.Minute) + require.NoError(t, err) + assert.Len(t, rr, 1) +} + +func TestQueryCache_GetReturnsEverySeriesWhenTruncated(t *testing.T) { + f, c := newFake(t, 10, podSeries(5, 7)) + qc := NewQueryCache(c, "c1") + + results, err := qc.Get(context.Background(), "kube_pod_info") + require.NoError(t, err) + assert.Len(t, results, 35) + + n := len(f.queries()) + again, err := qc.Get(context.Background(), "kube_pod_info") + require.NoError(t, err) + assert.Len(t, again, 35) + assert.Len(t, f.queries(), n, "a cached metric must not be re-queried") +} + +func TestQueryCache_TruncatedHostTypeFailsFetch(t *testing.T) { + series := podSeries(1, 3) + for i := range series { + series[i].labels["@resource.host.type"] = "a" + } + for i := 0; i < 20; i++ { // host type b cannot be split, so its shard fetch fails + series = append(series, fakeSeries{labels: map[string]string{ + "__name__": "kube_pod_info", "@resource.k8s.cluster.name": "c1", + "@resource.host.type": "b", "id": strconv.Itoa(i), + }}) + } + _, c := newFake(t, 10, series) + qc := NewQueryCache(c, "c1", WithHostTypes([]string{"a", "b"})) + + results, err := qc.Get(context.Background(), "kube_pod_info") + require.Error(t, err, "the 3 series of host type a are not the whole result") + assert.True(t, errors.Is(err, ErrTruncated), "got %v", err) + assert.Empty(t, results) +} + +func TestQueryCache_ShardRequestsStayWithinConcurrencyLimit(t *testing.T) { + var series []fakeSeries + for _, ht := range []string{"a", "b"} { + for _, s := range podSeries(6, 5) { + s.labels["@resource.host.type"] = ht + s.labels["@resource.k8s.node.name"] += "-" + ht + series = append(series, s) + } + } + f := &fakePromQL{t: t, series: series, limit: 10} + var inflight, peak int + var mu sync.Mutex + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + inflight++ + if inflight > peak { + peak = inflight + } + mu.Unlock() + time.Sleep(5 * time.Millisecond) + f.handler(w, r) + mu.Lock() + inflight-- + mu.Unlock() + })) + t.Cleanup(srv.Close) + c := newTestClient(srv.URL) + c.seriesLimit = 10 + qc := NewQueryCache(c, "c1", WithHostTypes([]string{"a", "b"}), WithMaxConcurrency(2)) + + results, err := qc.Get(context.Background(), "kube_pod_info") + require.NoError(t, err) + assert.Len(t, results, 60) + assert.LessOrEqual(t, peak, 2, "shard re-fetches must share the cache's concurrency limit") +} + +func TestIsVectorSelector(t *testing.T) { + for s, want := range map[string]bool{ + `up`: true, + `kube_pod_info{"a"="b"}`: true, + `{"__name__"="k8s.pod.cpu.usage","a"="}"}`: true, + ` m{a=~"x|y", b!~".+"} `: true, + `k8s.pod.cpu.usage`: false, + `count(m{a="b"})`: false, + `m{a="b"}[5m]`: false, + `m{a="b"} or n{c="d"}`: false, + `rate(m[5m])`: false, + `m{a="b"} offset 5m`: false, + } { + assert.Equal(t, want, isVectorSelector(s), s) + } +} + +func TestWithMatcher(t *testing.T) { + assert.Equal(t, `m{x="1"}`, withMatcher("m", `x="1"`)) + assert.Equal(t, `m{x="1"}`, withMatcher("m{}", `x="1"`)) + assert.Equal(t, `m{a="b",x="1"}`, withMatcher(`m{a="b"}`, `x="1"`)) + assert.Equal(t, `{"__name__"="a.b",x="1"}`, withMatcher(`{"__name__"="a.b",}`, `x="1"`)) +} + +func TestPackShards(t *testing.T) { + groups := []shardGroup{{"a.b", 30}, {"c", 25}, {"", 4}, {"d", 10}, {"e", 50}} + shards := packShards("@resource.k8s.node.name", groups, 40) + + total := 0 + for _, s := range shards { + total += s.want + assert.LessOrEqual(t, s.want, 50) + } + assert.Equal(t, 119, total) + assert.Contains(t, shards, shard{matcher: `"@resource.k8s.node.name"!~".+"`, want: 4, exact: true}) + assert.Contains(t, shards, shard{matcher: `"@resource.k8s.node.name"="e"`, want: 50, exact: true}) + assert.Contains(t, shards, shard{matcher: `"@resource.k8s.node.name"=~"a\\.b|d"`, want: 40, exact: false}) + + var many []shardGroup + for i := 0; i < 200; i++ { + many = append(many, shardGroup{fmt.Sprintf("ip-10-0-%d-%d.ec2.internal", i/50, i), 1}) + } + for _, s := range packShards("@resource.k8s.node.name", many, 400) { + value := s.matcher[strings.Index(s.matcher, `"=~"`)+4 : len(s.matcher)-1] + assert.LessOrEqual(t, len(unescape(value)), maxMatcherValueLen, "the endpoint rejects longer matcher values") + } +} + +func TestQueryCache_ParallelGetWithFilterStaysWithinConcurrencyLimit(t *testing.T) { + f := &fakePromQL{t: t, series: podSeries(6, 5), limit: 10} + var inflight, peak int + var mu sync.Mutex + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + inflight++ + if inflight > peak { + peak = inflight + } + mu.Unlock() + time.Sleep(5 * time.Millisecond) + f.handler(w, r) + mu.Lock() + inflight-- + mu.Unlock() + })) + t.Cleanup(srv.Close) + c := newTestClient(srv.URL) + c.seriesLimit = 10 + qc := NewQueryCache(c, "c1", WithMaxConcurrency(2)) + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + var wg sync.WaitGroup + for i := 0; i < 4; i++ { // more callers than slots, each needing several shard requests + wg.Add(1) + go func() { + defer wg.Done() + results, err := qc.GetWithFilter(ctx, "kube_pod_info", map[string]string{"@resource.k8s.namespace.name": "default"}) + assert.NoError(t, err, "a call that held a slot while its shards waited for one would deadlock here") + assert.Len(t, results, 30) + }() + } + wg.Wait() + assert.LessOrEqual(t, peak, 2, "shard re-fetches must share the cache's concurrency limit") +} diff --git a/util/otelmetrics/query_cache.go b/util/otelmetrics/query_cache.go index 51aef0f4f..3c39c75a3 100644 --- a/util/otelmetrics/query_cache.go +++ b/util/otelmetrics/query_cache.go @@ -32,6 +32,8 @@ type cacheEntry struct { // QueryCache provides session-scoped caching of PromQL queries. // Each unique metric name is queried exactly once; subsequent calls return cached data. // Concurrent requests for the same metric are deduplicated via singleflight. +// Every fetch goes through QueryComplete, so results are never silently truncated, and the +// concurrency limit bounds HTTP requests, so shard re-fetches stay inside it. type QueryCache struct { mu sync.RWMutex filtered map[string]cacheEntry @@ -142,6 +144,7 @@ func (qc *QueryCache) Get(ctx context.Context, metricName string) ([]MetricResul // fetchFiltered performs the actual HTTP query for a filtered metric. func (qc *QueryCache) fetchFiltered(ctx context.Context, metricName string) cacheEntry { + ctx = withRequestSlots(ctx, qc.sem) escaped := promqlEscaper.Replace(qc.cluster) sel := promqlMetricSelector(metricName) @@ -163,7 +166,7 @@ func (qc *QueryCache) fetchFiltered(ctx context.Context, metricName string) cach if clusterScopedOnly { promql := fmt.Sprintf(`%s"@resource.k8s.cluster.name"="%s","@resource.host.type"=""}`, sel, escaped) - r, err := qc.client.Query(ctx, promql) + r, err := qc.client.QueryComplete(ctx, promql) if err != nil { firstErr = err } else { @@ -181,15 +184,8 @@ func (qc *QueryCache) fetchFiltered(ctx context.Context, metricName string) cach wg.Add(1) go func(hostType string) { defer wg.Done() - select { - case qc.sem <- struct{}{}: - defer func() { <-qc.sem }() - case <-ctx.Done(): - ch <- queryResult{err: ctx.Err()} - return - } promql := fmt.Sprintf(`%s"@resource.k8s.cluster.name"="%s","@resource.host.type"="%s"}`, sel, escaped, hostType) - r, err := qc.client.Query(ctx, promql) + r, err := qc.client.QueryComplete(ctx, promql) ch <- queryResult{results: r, err: err} }(ht) } @@ -207,12 +203,13 @@ func (qc *QueryCache) fetchFiltered(ctx context.Context, metricName string) cach } } else { promql := fmt.Sprintf(`%s"@resource.k8s.cluster.name"="%s"}`, sel, escaped) - r, err := qc.client.Query(ctx, promql) + r, err := qc.client.QueryComplete(ctx, promql) results = r firstErr = err } - if len(results) == 0 && firstErr != nil { + // One failed host type makes the union incomplete, so it fails the whole fetch. + if firstErr != nil { slog.Debug("query failed", "metric", metricName, "error", firstErr) return cacheEntry{err: firstErr} } @@ -220,16 +217,10 @@ func (qc *QueryCache) fetchFiltered(ctx context.Context, metricName string) cach } // GetWithFilter returns results with additional PromQL label filters. Not cached. -// Shares the cache's concurrency limit, so a suite of parallel filtered queries -// doesn't burst past the endpoint's rate limit. +// Each HTTP request holds one slot of the cache's concurrency limit, so a suite of +// parallel filtered queries, and the shard re-fetches of a truncated one, stay inside it. func (qc *QueryCache) GetWithFilter(ctx context.Context, metricName string, extraFilters map[string]string) ([]MetricResult, error) { - select { - case qc.sem <- struct{}{}: - defer func() { <-qc.sem }() - case <-ctx.Done(): - return nil, ctx.Err() - } - return qc.client.Query(ctx, promqlMetricSelector(metricName)+qc.filterMatchers(extraFilters)+"}") + return qc.client.QueryComplete(withRequestSlots(ctx, qc.sem), promqlMetricSelector(metricName)+qc.filterMatchers(extraFilters)+"}") } // filterMatchers renders the cluster matcher plus extraFilters as PromQL label @@ -267,7 +258,7 @@ func (qc *QueryCache) GetUnfiltered(ctx context.Context, metricName string) ([]M } qc.mu.Unlock() - results, queryErr := qc.client.Query(ctx, metricName) + results, queryErr := qc.client.QueryComplete(withRequestSlots(ctx, qc.sem), metricName) entry := cacheEntry{results: results, err: queryErr} // Do not cache empty successful results; a transient miss would poison From 469d0d38069b30a03cbc06b59805bf31e6de3d92 Mon Sep 17 00:00:00 2001 From: Petru Anica-Popa Date: Thu, 29 Oct 2026 16:28:32 +0000 Subject: [PATCH 2/2] test(otel): check node and pod labels, the attribute schema and KSM zones against the cluster --- test/otel/efa/attribute_schema_test.go | 54 ++++ test/otel/gpu/attribute_schema_test.go | 54 ++++ test/otel/neuron/attribute_schema_test.go | 57 ++++ test/otel/standard/k8s_helpers_test.go | 64 +++-- .../standard/ksm_availability_zone_test.go | 94 ++++++ test/otel/standard/ksm_test.go | 15 +- test/otel/standard/labels_reverse_test.go | 267 ++++++++++++++++++ test/otel/standard/schema_ksm_test.go | 140 +++++++++ test/otel/standard/schema_node_test.go | 94 ++++++ test/otel/standard/schema_test.go | 132 +++++++++ util/otelmetrics/attribute_schema.go | 196 +++++++++++++ util/otelmetrics/attribute_schema_test.go | 125 ++++++++ util/otelmetrics/label_reconcile.go | 250 ++++++++++++++++ util/otelmetrics/label_reconcile_test.go | 204 +++++++++++++ util/otelmetrics/schema_keys.go | 90 ++++++ 15 files changed, 1796 insertions(+), 40 deletions(-) create mode 100644 test/otel/efa/attribute_schema_test.go create mode 100644 test/otel/gpu/attribute_schema_test.go create mode 100644 test/otel/neuron/attribute_schema_test.go create mode 100644 test/otel/standard/ksm_availability_zone_test.go create mode 100644 test/otel/standard/labels_reverse_test.go create mode 100644 test/otel/standard/schema_ksm_test.go create mode 100644 test/otel/standard/schema_node_test.go create mode 100644 test/otel/standard/schema_test.go create mode 100644 util/otelmetrics/attribute_schema.go create mode 100644 util/otelmetrics/attribute_schema_test.go create mode 100644 util/otelmetrics/label_reconcile.go create mode 100644 util/otelmetrics/label_reconcile_test.go create mode 100644 util/otelmetrics/schema_keys.go diff --git a/test/otel/efa/attribute_schema_test.go b/test/otel/efa/attribute_schema_test.go new file mode 100644 index 000000000..a9bb496b3 --- /dev/null +++ b/test/otel/efa/attribute_schema_test.go @@ -0,0 +1,54 @@ +//go:build integration + +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: MIT + +package efa + +import ( + "context" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/aws/amazon-cloudwatch-agent-test/util/otelmetrics" +) + +// --------------------------------------------------------------------------- +// TestAttributeSchema — the closed set of resource and instrumentation-scope +// attributes on EFA series. Every series must have every required key and +// nothing undeclared, expected keys must be on at least one series, and keys set +// together must arrive together. Node and pod labels are open by construction +// (k8sattributes extracts "(.*)"). Datapoint scope is not covered. +// --------------------------------------------------------------------------- + +// awsdevicepodcorrelation then efa_promote; only on devices a pod holds. +var efaPodKeys = []string{"k8s.pod.name", "k8s.namespace.name", "k8s.container.name"} + +// awsefareceiver sets no resource attributes, so unlike the scraped pipelines +// there are no server.* or service.* keys. This suite runs no workload that holds +// a device, so the pod keys are optional rather than expected. +var efaResourceSchema = otelmetrics.KeySchema{ + Required: otelmetrics.ConcatKeys(otelmetrics.CloudKeys, otelmetrics.HostKeys, []string{ + "k8s.cluster.name", // set_cluster_name; protected + "k8s.node.name", // set_node_name, promote_node_name; protected + "k8s.node.uid", // k8sattributes/cw_k8s_ci_v0_node + }), + Optional: otelmetrics.ConcatKeys(efaPodKeys, []string{"k8s.pod.uid"}, otelmetrics.OwnerKindKeys, otelmetrics.WorkloadKeys), + OpenPrefixes: []string{"k8s.node.label.", "k8s.pod.label."}, + Implies: otelmetrics.MergeImplies(otelmetrics.WorkloadImplies(), otelmetrics.Together(efaPodKeys...)), +} + +func TestAttributeSchema(t *testing.T) { + t.Parallel() + tally, err := otelmetrics.TallySchema(context.Background(), queryCache, efaMetricNamesList, otelmetrics.GroupSchema{ + Resource: efaResourceSchema, + Instrumentation: otelmetrics.KeySchema{Required: otelmetrics.ScopeKeys}, // set_scope_efa + }, nil) + require.NoError(t, err, "no EFA nodes?") + t.Logf("%d series across %d metrics", tally.Series, len(efaMetricNamesList)) + require.NotZero(t, tally.Series, "no series checked") + for _, p := range tally.Problems() { + t.Errorf("%s: change the schema if intended", p) + } +} diff --git a/test/otel/gpu/attribute_schema_test.go b/test/otel/gpu/attribute_schema_test.go new file mode 100644 index 000000000..3b7caa7bc --- /dev/null +++ b/test/otel/gpu/attribute_schema_test.go @@ -0,0 +1,54 @@ +//go:build integration + +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: MIT + +package gpu + +import ( + "context" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/aws/amazon-cloudwatch-agent-test/util/otelmetrics" +) + +// --------------------------------------------------------------------------- +// TestAttributeSchema — the closed set of resource and instrumentation-scope +// attributes on DCGM series. Every series must have every required key and +// nothing undeclared, expected keys must be on at least one series, and keys set +// together must arrive together. Node and pod labels are open by construction +// (k8sattributes extracts "(.*)"). Datapoint scope is not covered. +// --------------------------------------------------------------------------- + +// dcgm_promote; dcgm-exporter maps a GPU to a pod only while the pod holds it. +var dcgmPodKeys = []string{"k8s.pod.name", "k8s.namespace.name", "k8s.container.name"} + +var dcgmResourceSchema = otelmetrics.KeySchema{ + Required: otelmetrics.ConcatKeys(otelmetrics.PrometheusReceiverKeys, otelmetrics.CloudKeys, otelmetrics.HostKeys, []string{ + "k8s.cluster.name", // set_cluster_name; protected + "k8s.node.name", // set_node_name, promote_node_name; protected + "k8s.node.uid", // k8sattributes/cw_k8s_ci_v0_node + }), + // The multi-gpu-burn Deployment holds GPUs; TestDCGMPodName requires it. + Expected: otelmetrics.ConcatKeys(dcgmPodKeys, []string{"k8s.deployment.name", "k8s.replicaset.name"}, otelmetrics.WorkloadKeys), + Optional: otelmetrics.ConcatKeys([]string{"k8s.pod.uid"}, // k8sattributes/cw_k8s_ci_v0_pod + otelmetrics.Without(otelmetrics.OwnerKindKeys, "k8s.deployment.name", "k8s.replicaset.name")), + OpenPrefixes: []string{"k8s.node.label.", "k8s.pod.label."}, + Implies: otelmetrics.MergeImplies(otelmetrics.WorkloadImplies(), otelmetrics.Together(dcgmPodKeys...)), +} + +func TestAttributeSchema(t *testing.T) { + t.Parallel() + tally, err := otelmetrics.TallySchema(context.Background(), queryCache, dcgmMetricNamesList, otelmetrics.GroupSchema{ + Resource: dcgmResourceSchema, + Instrumentation: otelmetrics.KeySchema{Required: otelmetrics.ScopeKeys}, // set_scope_dcgm + }, nil) + require.NoError(t, err, "no GPU nodes?") + t.Logf("%d series across %d metrics", tally.Series, len(dcgmMetricNamesList)) + require.NotZero(t, tally.Series, "no series checked") + for _, p := range tally.Problems() { + t.Errorf("%s: change the schema if intended", p) + } +} diff --git a/test/otel/neuron/attribute_schema_test.go b/test/otel/neuron/attribute_schema_test.go new file mode 100644 index 000000000..a96678965 --- /dev/null +++ b/test/otel/neuron/attribute_schema_test.go @@ -0,0 +1,57 @@ +//go:build integration + +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: MIT + +package neuron + +import ( + "context" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/aws/amazon-cloudwatch-agent-test/util/otelmetrics" +) + +// --------------------------------------------------------------------------- +// TestAttributeSchema — the closed set of resource and instrumentation-scope +// attributes on Neuron series. Every series must have every required key and +// nothing undeclared, expected keys must be on at least one series, and keys set +// together must arrive together. Node and pod labels are open by construction +// (k8sattributes extracts "(.*)"). Datapoint scope is not covered. +// --------------------------------------------------------------------------- + +// awsdevicepodcorrelation then groupbyattrs; only on cores or devices a pod holds. +var neuronPodKeys = []string{"k8s.pod.name", "k8s.namespace.name", "k8s.container.name"} + +var neuronResourceSchema = otelmetrics.KeySchema{ + Required: otelmetrics.ConcatKeys(otelmetrics.PrometheusReceiverKeys, otelmetrics.CloudKeys, otelmetrics.HostKeys, []string{ + "k8s.cluster.name", // set_cluster_name; protected + "k8s.node.name", // set_node_name, promote_node_name; protected + "k8s.node.uid", // k8sattributes/cw_k8s_ci_v0_node + }), + // The neuron-burn Deployments hold cores and devices and run a Neuron runtime. + Expected: otelmetrics.ConcatKeys(neuronPodKeys, []string{ + "aws.neuron.runtime.tag", // neuron_promote; zero-filled series of an idle device have no runtime_tag + "k8s.deployment.name", "k8s.replicaset.name", + }, otelmetrics.WorkloadKeys), + Optional: otelmetrics.ConcatKeys([]string{"k8s.pod.uid"}, // k8sattributes/cw_k8s_ci_v0_pod + otelmetrics.Without(otelmetrics.OwnerKindKeys, "k8s.deployment.name", "k8s.replicaset.name")), + OpenPrefixes: []string{"k8s.node.label.", "k8s.pod.label."}, + Implies: otelmetrics.MergeImplies(otelmetrics.WorkloadImplies(), otelmetrics.Together(neuronPodKeys...)), +} + +func TestAttributeSchema(t *testing.T) { + t.Parallel() + tally, err := otelmetrics.TallySchema(context.Background(), queryCache, neuronMetricNamesList, otelmetrics.GroupSchema{ + Resource: neuronResourceSchema, + Instrumentation: otelmetrics.KeySchema{Required: otelmetrics.ScopeKeys}, // set_scope_neuron_monitor + }, nil) + require.NoError(t, err, "no Neuron nodes?") + t.Logf("%d series across %d metrics", tally.Series, len(neuronMetricNamesList)) + require.NotZero(t, tally.Series, "no series checked") + for _, p := range tally.Problems() { + t.Errorf("%s: change the schema if intended", p) + } +} diff --git a/test/otel/standard/k8s_helpers_test.go b/test/otel/standard/k8s_helpers_test.go index 3269ea071..805ba15a5 100644 --- a/test/otel/standard/k8s_helpers_test.go +++ b/test/otel/standard/k8s_helpers_test.go @@ -8,16 +8,16 @@ package standard import ( "context" "fmt" - "os" - "path/filepath" "strings" "sync" "testing" "time" + "github.com/stretchr/testify/require" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/client-go/kubernetes" + "k8s.io/client-go/rest" "k8s.io/client-go/tools/clientcmd" ) @@ -45,22 +45,46 @@ func getGroundTruth(t *testing.T) *k8sGroundTruth { return groundTruth } -func buildGroundTruth() (*k8sGroundTruth, error) { - kubeconfig := os.Getenv("KUBECONFIG") - if kubeconfig == "" { - kubeconfig = filepath.Join(os.Getenv("HOME"), ".kube", "config") - } +var ( + k8sClient *kubernetes.Clientset + k8sClientOnce sync.Once + k8sClientErr error +) - restConfig, err := clientcmd.BuildConfigFromFlags("", kubeconfig) - if err != nil { - return nil, fmt.Errorf("building kubeconfig: %w", err) - } +// sharedK8sClientset builds one clientset from the default kubeconfig loading rules +// ($KUBECONFIG, then ~/.kube/config), falling back to the in-cluster configuration. +func sharedK8sClientset() (*kubernetes.Clientset, error) { + k8sClientOnce.Do(func() { + restConfig, err := clientcmd.NewNonInteractiveDeferredLoadingClientConfig( + clientcmd.NewDefaultClientConfigLoadingRules(), nil).ClientConfig() + if err != nil { + if restConfig, err = rest.InClusterConfig(); err != nil { + k8sClientErr = fmt.Errorf("no kubeconfig or in-cluster config: %w", err) + return + } + } + k8sClient, k8sClientErr = kubernetes.NewForConfig(restConfig) + }) + return k8sClient, k8sClientErr +} - clientset, err := kubernetes.NewForConfig(restConfig) +func k8sClientset(t *testing.T) *kubernetes.Clientset { + t.Helper() + clientset, err := sharedK8sClientset() + require.NoError(t, err, "creating K8s clientset") + return clientset +} + +func buildGroundTruth() (*k8sGroundTruth, error) { + clientset, err := sharedK8sClientset() if err != nil { - return nil, fmt.Errorf("creating K8s clientset: %w", err) + return nil, err } + return snapshotGroundTruth(clientset) +} +// snapshotGroundTruth lists every node and pod now. +func snapshotGroundTruth(clientset kubernetes.Interface) (*k8sGroundTruth, error) { ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() @@ -116,19 +140,7 @@ func imageTagFromPod(t *testing.T, gt *k8sGroundTruth, namespace, labelKey, labe // k8sServerVersion returns the Kubernetes API server version string. func k8sServerVersion(t *testing.T) string { t.Helper() - kubeconfig := os.Getenv("KUBECONFIG") - if kubeconfig == "" { - kubeconfig = filepath.Join(os.Getenv("HOME"), ".kube", "config") - } - restConfig, err := clientcmd.BuildConfigFromFlags("", kubeconfig) - if err != nil { - t.Fatalf("building kubeconfig: %v", err) - } - clientset, err := kubernetes.NewForConfig(restConfig) - if err != nil { - t.Fatalf("creating K8s clientset: %v", err) - } - sv, err := clientset.Discovery().ServerVersion() + sv, err := k8sClientset(t).Discovery().ServerVersion() if err != nil { t.Fatalf("getting K8s server version: %v", err) } diff --git a/test/otel/standard/ksm_availability_zone_test.go b/test/otel/standard/ksm_availability_zone_test.go new file mode 100644 index 000000000..f02636a86 --- /dev/null +++ b/test/otel/standard/ksm_availability_zone_test.go @@ -0,0 +1,94 @@ +//go:build integration + +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: MIT + +package standard + +import ( + "testing" + + "github.com/stretchr/testify/require" + + "github.com/aws/amazon-cloudwatch-agent-test/util/otelmetrics" +) + +// --------------------------------------------------------------------------- +// TestKSMAvailabilityZoneIsTheObjectsZone — the cluster scraper runs on one +// node, so its own AZ is wrong for most kube-state-metrics series. On every KSM +// series, cloud.availability_zone must be the zone of the node the series names: +// nodemetadataenricher sets it, or else the node's topology.kubernetes.io/zone label +// does, and a pod's own zone label replaces it. A series whose node k8sattributes +// knows must carry it. A series that names no node, such as a Deployment's, must not. +// The pod zone label itself is denied, so the API is the reference for it. +// Series are read cluster-wide, not through queryCache.Get, whose host.type="" +// filter on workload metrics would hide a series that gained the scraper's host keys. +// --------------------------------------------------------------------------- + +const zoneLabel = "topology.kubernetes.io/zone" + +func TestKSMAvailabilityZoneIsTheObjectsZone(t *testing.T) { + gt := freshGroundTruth(t) + buckets := map[string][]string{ + "node": ksmNodeBucket, "pod": ksmPodBucket, "container": ksmContainerBucket, + "object": otelmetrics.ConcatKeys(ksmWorkloadBucket.Deployment, ksmWorkloadBucket.DaemonSet, + ksmWorkloadBucket.StatefulSet, ksmWorkloadBucket.Job, ksmWorkloadBucket.CronJob, ksmClusterBucket), + } + for _, bucket := range []string{"node", "pod", "container", "object"} { + bucket := bucket + t.Run(bucket, func(t *testing.T) { + var series, withZone, fromPodLabel int + wrong, missing, absent, unexpected := map[string]int{}, map[string]int{}, map[string]int{}, map[string]int{} + for _, metricName := range buckets[bucket] { + for _, r := range queryAllSeries(t, metricName) { + series++ + res := r.Labels.Resource + az, hasZone := res[ksmZoneKey] + if hasZone { + withZone++ + } + nodeName := res["k8s.node.name"] + if nodeName == "" { + if hasZone { + unexpected[metricName]++ + } + continue + } + node, known := gt.nodes[nodeName] + if !known { + continue // the node left between the API read and the scrape + } + if hasZone && az != node.Labels[zoneLabel] { + wrong[metricName+": "+az+" on a node in "+node.Labels[zoneLabel]]++ + } + pod, ok := gt.pods[res["k8s.namespace.name"]+"/"+res["k8s.pod.name"]] + if want := pod.Labels[zoneLabel]; ok && want != "" && pod.Spec.NodeName == nodeName { + fromPodLabel++ + if az != want { + missing[metricName]++ + } + } else if !hasZone && node.Labels[zoneLabel] != "" && res["k8s.node.uid"] != "" { + absent[metricName]++ // transform/cw_k8s_ci_v0_ksm_zone takes the node's zone label + } + } + } + t.Logf("%d series, %d with an AZ, %d of a pod with a zone label", series, withZone, fromPodLabel) + require.NotZero(t, series, "no %s series", bucket) + for _, k := range otelmetrics.SortedKeys(wrong) { + t.Errorf("%s: wrong AZ on %d series", k, wrong[k]) + } + for _, k := range otelmetrics.SortedKeys(missing) { + t.Errorf("%s: %d series of a pod with a zone label lack that AZ", k, missing[k]) + } + for _, k := range otelmetrics.SortedKeys(absent) { + t.Errorf("%s: %d series of a known node with a zone label carry no AZ", k, absent[k]) + } + for _, k := range otelmetrics.SortedKeys(unexpected) { + t.Errorf("%s: %d series name no node but carry an AZ, the cluster scraper's", k, unexpected[k]) + } + if bucket == "node" || bucket == "pod" { + require.NotZero(t, withZone, "no %s series carries an AZ", bucket) + } + }) + } +} diff --git a/test/otel/standard/ksm_test.go b/test/otel/standard/ksm_test.go index 45270585a..598ce8ac1 100644 --- a/test/otel/standard/ksm_test.go +++ b/test/otel/standard/ksm_test.go @@ -13,9 +13,6 @@ import ( "github.com/stretchr/testify/require" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - "k8s.io/client-go/kubernetes" - "k8s.io/client-go/rest" - "k8s.io/client-go/tools/clientcmd" "github.com/aws/amazon-cloudwatch-agent-test/util/otelmetrics" ) @@ -1571,18 +1568,8 @@ func TestKSM_LeaseExistence(t *testing.T) { t.Parallel() gt := getGroundTruth(t) - loadingRules := clientcmd.NewDefaultClientConfigLoadingRules() - kubeConfig := clientcmd.NewNonInteractiveDeferredLoadingClientConfig(loadingRules, nil) - restConfig, err := kubeConfig.ClientConfig() - if err != nil { - restConfig, err = rest.InClusterConfig() - require.NoError(t, err, "no kubeconfig or in-cluster config") - } - clientset, err := kubernetes.NewForConfig(restConfig) - require.NoError(t, err, "creating K8s clientset") - ctx := context.Background() - leases, err := clientset.CoordinationV1().Leases("amazon-cloudwatch").List(ctx, metav1.ListOptions{}) + leases, err := k8sClientset(t).CoordinationV1().Leases("amazon-cloudwatch").List(ctx, metav1.ListOptions{}) require.NoError(t, err, "listing Leases") leaseByNode := make(map[string]bool) diff --git a/test/otel/standard/labels_reverse_test.go b/test/otel/standard/labels_reverse_test.go new file mode 100644 index 000000000..c0cb74ed9 --- /dev/null +++ b/test/otel/standard/labels_reverse_test.go @@ -0,0 +1,267 @@ +//go:build integration + +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: MIT + +package standard + +import ( + "context" + "fmt" + "sort" + "testing" + "time" + + "github.com/stretchr/testify/require" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + "github.com/aws/amazon-cloudwatch-agent-test/util/otelmetrics" +) + +// --------------------------------------------------------------------------- +// Reverse label reconciliation. TestNodeLabelsMatchKubernetesAPI checks that +// every k8s.node.label.* on a metric exists on the node; it cannot notice a +// label that never arrived. These tests start from the Kubernetes API instead: +// every node and pod label the running denylist does not match must arrive with +// the same value, and every label it matches must not. The expectation comes +// from the cluster and the live configuration; the one maintained list is the +// never-denied attributes, which stops an over-broad denylist entry passing. +// +// Labels reach metrics on two paths, each with its own k8sattributes extraction +// and denylist: the cluster scraper's kube-state-metrics pipeline, which reports +// every node and pod, and the node agent's kubeletstats pipeline, which reports +// the nodes and pods where the agent runs. Both are queried cluster-wide, so the +// clusterHostTypes pin does not limit them. Every object a path reports must +// have a series, or an object that lost all of its series would pass: every +// node and pod on the cluster scraper path, and on the node agent path every +// node with a ready agent pod and every running pod on such a node. Each pass +// reads the API, then every series at one evaluation time, so the two sides are +// seconds apart. An object fails only if it fails two passes recheckDelay apart, +// because a label change or a new object reaches the API before the next +// collection is ingested. +// --------------------------------------------------------------------------- + +const ( + agentNamespace = "amazon-cloudwatch" + nodeAgentConfigMap = "cloudwatch-agent" + nodeAgentDaemonSet = "cloudwatch-agent" + clusterScraperConfigMap = "cloudwatch-agent-cluster-scraper" + otelConfigKey = "cwagentotelconfig.yaml" + + // recheckDelay covers one collection interval of either agent plus ingestion. + recheckDelay = 90 * time.Second +) + +// agentDenylist reads the attribute-limit configuration an agent is running. +func agentDenylist(t *testing.T, configMap string) otelmetrics.AttributeDenylist { + t.Helper() + cm, err := k8sClientset(t).CoreV1().ConfigMaps(agentNamespace).Get( + context.Background(), configMap, metav1.GetOptions{}) + require.NoError(t, err, "reading ConfigMap %s/%s", agentNamespace, configMap) + raw, ok := cm.Data[otelConfigKey] + require.True(t, ok, "ConfigMap %s has no %s", configMap, otelConfigKey) + d, err := otelmetrics.ParseAttributeDenylist(raw) + require.NoError(t, err, "ConfigMap %s", configMap) + return d +} + +func TestDenylistSparesNeverDeniedAttributes(t *testing.T) { + for _, configMap := range []string{nodeAgentConfigMap, clusterScraperConfigMap} { + t.Run(configMap, func(t *testing.T) { + d := agentDenylist(t, configMap) + for _, v := range d.NeverDeniedViolations() { + t.Errorf("%s: %s", d.Processor, v) + } + }) + } +} + +// reconcilePass reads both sides once and reconciles them. +type reconcilePass func(t *testing.T) otelmetrics.ObjectReconciliation + +// failedObjects returns the objects that failed a pass: a series that did not +// reconcile, or a required object with no series. +func failedObjects(r otelmetrics.ObjectReconciliation) map[string]bool { + failed := make(map[string]bool, len(r.Failed)+len(r.NoSeries)) + for key := range r.Failed { + failed[key] = true + } + for _, key := range r.NoSeries { + failed[key] = true + } + return failed +} + +// reconcileTwice runs pass and, if any object failed, runs it again after +// recheckDelay. Only objects that fail both passes are reported. +func reconcileTwice(t *testing.T, kind string, pass reconcilePass) { + t.Helper() + first := pass(t) + t.Logf("reconciled %d %ss, %d labels, %d series skipped at the attribute cap, %d failed, %d of %d required with no series", + first.Objects, kind, first.Labels, first.AtCap, len(first.Failed), len(first.NoSeries), first.Required) + require.NotZero(t, first.Required, "the Kubernetes API listed no %s that must have a series", kind) + require.NotZero(t, first.Objects, "no series matched a %s from the Kubernetes API", kind) + require.NotZero(t, first.Labels, "%ss matched but carried no checkable label", kind) + failedBefore := failedObjects(first) + if len(failedBefore) == 0 { + return + } + t.Logf("re-reading the API and the metrics in %s to separate label churn and new objects from a real drop", recheckDelay) + time.Sleep(recheckDelay) + second := pass(t) + keys := make([]string, 0, len(failedBefore)) + for key := range failedObjects(second) { + if failedBefore[key] { + keys = append(keys, key) + } + } + sort.Strings(keys) + for _, key := range keys { + diff, hasSeries := second.Failed[key] + if !hasSeries { + t.Errorf("%s %s: no series", kind, key) + continue + } + t.Errorf("%s %s: missing %v, value mismatch %v, denied but present %v", + kind, key, diff.Missing, diff.Mismatched, diff.Leaked) + } +} + +// agentNodes returns the listed nodes where a node agent pod is ready. +func agentNodes(gt *k8sGroundTruth) map[string]bool { + nodes := map[string]bool{} + for _, p := range gt.pods { + if p.Namespace != agentNamespace || p.DeletionTimestamp != nil || !ownedByDaemonSet(p, nodeAgentDaemonSet) { + continue + } + if _, listed := gt.nodes[p.Spec.NodeName]; !listed { + continue // it left between the two reads + } + for _, c := range p.Status.Conditions { + if c.Type == corev1.PodReady && c.Status == corev1.ConditionTrue { + nodes[p.Spec.NodeName] = true + } + } + } + return nodes +} + +func ownedByDaemonSet(p corev1.Pod, name string) bool { + for _, o := range p.OwnerReferences { + if o.Kind == "DaemonSet" && o.Name == name { + return true + } + } + return false +} + +// everyNode and everyPod are the objects that must have a series on the cluster +// scraper path, because kube-state-metrics reports every object the API lists. +func everyNode(gt *k8sGroundTruth) []string { + keys := make([]string, 0, len(gt.nodes)) + for name := range gt.nodes { + keys = append(keys, name) + } + return keys +} + +func everyPod(gt *k8sGroundTruth) []string { + keys := make([]string, 0, len(gt.pods)) + for key := range gt.pods { + keys = append(keys, key) + } + return keys +} + +// nodesWithAgent and runningPodsWithAgent are the objects that must have a series +// on the node agent path, which reports its own node and the pods running there. +func nodesWithAgent(gt *k8sGroundTruth) []string { + var keys []string + for name := range agentNodes(gt) { + keys = append(keys, name) + } + return keys +} + +func runningPodsWithAgent(gt *k8sGroundTruth) []string { + agents := agentNodes(gt) + var keys []string + for key, p := range gt.pods { + // A terminating pod can stop reporting before the API drops it. + if agents[p.Spec.NodeName] && p.Status.Phase == corev1.PodRunning && p.DeletionTimestamp == nil { + keys = append(keys, key) + } + } + return keys +} + +// freshGroundTruth lists the nodes and pods now, unlike the session-wide getGroundTruth. +func freshGroundTruth(t *testing.T) *k8sGroundTruth { + t.Helper() + gt, err := snapshotGroundTruth(k8sClientset(t)) + require.NoError(t, err, "listing nodes and pods") + return gt +} + +// queryAllSeries returns every series of a metric in this cluster, on every node, +// evaluated at one time. +func queryAllSeries(t *testing.T, metric string) []otelmetrics.MetricResult { + t.Helper() + results, err := client.QueryComplete(context.Background(), fmt.Sprintf( + `{"__name__"="%s","@resource.k8s.cluster.name"="%s"}`, + otelmetrics.EscapePromQLValue(metric), otelmetrics.EscapePromQLValue(cfg.ClusterName))) + require.NoError(t, err, "querying %s", metric) + require.NotEmpty(t, results, "%s not available", metric) + return results +} + +// labelPath is one metric that carries an object's labels, one series per object, +// the agent whose denylist the series went through, and the objects that must +// have a series. +type labelPath struct { + agent, metric, configMap string + required func(*k8sGroundTruth) []string +} + +func TestNodeLabelsReachMetrics(t *testing.T) { + for _, p := range []labelPath{ + {"cluster-scraper", "kube_node_info", clusterScraperConfigMap, everyNode}, + {"node-agent", "k8s.node.cpu.usage", nodeAgentConfigMap, nodesWithAgent}, + } { + p := p + t.Run(p.agent, func(t *testing.T) { + d := agentDenylist(t, p.configMap) + reconcileTwice(t, "node", func(t *testing.T) otelmetrics.ObjectReconciliation { + gt := freshGroundTruth(t) + return otelmetrics.ReconcileObjects(queryAllSeries(t, p.metric), + func(l otelmetrics.MetricLabels) (string, map[string]string, bool) { + name := l.Resource["k8s.node.name"] + node, ok := gt.nodes[name] // absent if it joined or left between the two reads + return name, node.Labels, ok + }, p.required(gt), "k8s.node.label.", d) + }) + }) + } +} + +func TestPodLabelsReachMetrics(t *testing.T) { + for _, p := range []labelPath{ + {"cluster-scraper", "kube_pod_info", clusterScraperConfigMap, everyPod}, + {"node-agent", "k8s.pod.cpu.usage", nodeAgentConfigMap, runningPodsWithAgent}, + } { + p := p + t.Run(p.agent, func(t *testing.T) { + d := agentDenylist(t, p.configMap) + reconcileTwice(t, "pod", func(t *testing.T) otelmetrics.ObjectReconciliation { + gt := freshGroundTruth(t) + return otelmetrics.ReconcileObjects(queryAllSeries(t, p.metric), + func(l otelmetrics.MetricLabels) (string, map[string]string, bool) { + key := l.Resource["k8s.namespace.name"] + "/" + l.Resource["k8s.pod.name"] + pod, ok := gt.pods[key] // absent if it was created or deleted between the two reads + return key, pod.Labels, ok + }, p.required(gt), "k8s.pod.label.", d) + }) + }) + } +} diff --git a/test/otel/standard/schema_ksm_test.go b/test/otel/standard/schema_ksm_test.go new file mode 100644 index 000000000..14297708b --- /dev/null +++ b/test/otel/standard/schema_ksm_test.go @@ -0,0 +1,140 @@ +//go:build integration + +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: MIT + +package standard + +import "github.com/aws/amazon-cloudwatch-agent-test/util/otelmetrics" + +// --------------------------------------------------------------------------- +// kube-state-metrics rows of TestAttributeSchema, one per ksm_test.go bucket and +// workload kind. Every key comes from the cluster scraper's kube_state_metrics +// pipeline. The raw labels groupbyattrs lifts (pod, namespace, node, uid, ...) +// are deleted from the resource after ksm_promote copies them back, so no row +// allows them. EKS only: this suite does not run on AKS. +// --------------------------------------------------------------------------- + +// On every KSM series. +var ksmCommonKeys = otelmetrics.ConcatKeys(otelmetrics.PrometheusReceiverKeys, + otelmetrics.Without(otelmetrics.CloudKeys, ksmZoneKey), []string{ + "k8s.cluster.name", // set_cluster_name; protected + }) + +// The zone of the node a series names: nodemetadataenricher sets it with host.id, and +// transform/cw_k8s_ci_v0_ksm_zone takes it from the node's zone label otherwise. A pod's +// zone label replaces it (transform/cw_k8s_ci_v0_pod_topology). A series that names no +// node has none. TestKSMAvailabilityZoneIsTheObjectsZone checks the value. +const ksmZoneKey = "cloud.availability_zone" + +// A series whose node k8sattributes knows has the node's labels, so it carries the AZ. +var ksmZoneImplies = map[string][]string{"k8s.node.uid": {ksmZoneKey}} + +// ksm_promote from namespace, pod and uid; protected. +var ksmPodKeys = []string{"k8s.namespace.name", "k8s.pod.name", "k8s.pod.uid"} + +// nodemetadataenricher sets the host keys and the AZ together, only while the node's +// Lease is fresh, that is while its DaemonSet agent runs, so some series carry them. +// TestKSM_NodeBucket_HasHostAttributes requires them. +var ksmHostImplies = otelmetrics.MergeImplies(otelmetrics.Together(otelmetrics.HostKeys...), + map[string][]string{"host.id": {ksmZoneKey}}) + +// The chart installs a DaemonSet (the agent) and Deployments (the operator and the +// cluster scraper), so these owner keys are on some pod series of any cluster. +// Jobs, CronJobs and StatefulSets depend on the cluster's workloads. +var ksmOwnerExpected = otelmetrics.ConcatKeys( + []string{"k8s.daemonset.name", "k8s.deployment.name", "k8s.replicaset.name"}, otelmetrics.WorkloadKeys) +var ksmOwnerOptional = []string{"k8s.cronjob.name", "k8s.job.name", "k8s.statefulset.name"} + +// One row per workload kind rather than one with the object-name keys optional, +// so a series that loses its own k8s..name or gains another kind's fails. +func ksmWorkloadSchema(name string, metrics []string, objectKey string) groupSchema { + return groupSchema{ + name: name, + metrics: func() []string { return metrics }, + schema: otelmetrics.GroupSchema{ + Resource: otelmetrics.KeySchema{ + Required: otelmetrics.ConcatKeys(ksmCommonKeys, []string{ + "k8s.namespace.name", // ksm_promote; protected + objectKey, // ksm_promote; protected + "k8s.workload.name", // set_workload; protected + "k8s.workload.type", // set_workload; protected + }), + }, + Instrumentation: scopeInstrumentation, + }, + } +} + +func init() { + attributeSchemas = append(attributeSchemas, + groupSchema{ + name: "ksm-node", + metrics: func() []string { return ksmNodeBucket }, + schema: otelmetrics.GroupSchema{ + Resource: otelmetrics.KeySchema{ + Required: otelmetrics.ConcatKeys(ksmCommonKeys, []string{ + "k8s.node.name", // ksm_promote; protected + "k8s.node.uid", // k8sattributes/cw_k8s_ci_v0_node, set by its node-label watch + }), + Expected: append([]string{ksmZoneKey}, otelmetrics.HostKeys...), + OpenPrefixes: []string{"k8s.node.label."}, + Implies: otelmetrics.MergeImplies(ksmHostImplies, ksmZoneImplies), + }, + Instrumentation: scopeInstrumentation, + }, + }, + groupSchema{ + name: "ksm-pod", + metrics: func() []string { return ksmPodBucket }, + schema: otelmetrics.GroupSchema{ + Resource: otelmetrics.KeySchema{ + Required: otelmetrics.ConcatKeys(ksmCommonKeys, ksmPodKeys), + // k8s.node.name from ksm_promote or k8sattributes/cw_k8s_ci_v0_pod, then k8s.node.uid + // from k8sattributes/cw_k8s_ci_v0_node; both only on scheduled pods. + Expected: otelmetrics.ConcatKeys([]string{"k8s.node.name", "k8s.node.uid", ksmZoneKey}, otelmetrics.HostKeys, ksmOwnerExpected), + Optional: ksmOwnerOptional, + OpenPrefixes: []string{"k8s.node.label.", "k8s.pod.label."}, + Implies: otelmetrics.MergeImplies(ksmHostImplies, ksmZoneImplies, otelmetrics.WorkloadImplies(), + otelmetrics.Together("k8s.node.name", "k8s.node.uid")), + }, + Instrumentation: scopeInstrumentation, + }, + }, + groupSchema{ + name: "ksm-container", + metrics: func() []string { return ksmContainerBucket }, + schema: otelmetrics.GroupSchema{ + Resource: otelmetrics.KeySchema{ + Required: otelmetrics.ConcatKeys(ksmCommonKeys, ksmPodKeys, []string{ + "k8s.container.name", // ksm_promote; protected + "k8s.node.name", // k8sattributes/cw_k8s_ci_v0_pod; container status implies a bound pod + "k8s.node.uid", // k8sattributes/cw_k8s_ci_v0_node + }), + Expected: otelmetrics.ConcatKeys([]string{ksmZoneKey}, otelmetrics.HostKeys, ksmOwnerExpected), + Optional: ksmOwnerOptional, + OpenPrefixes: []string{"k8s.node.label.", "k8s.pod.label."}, + Implies: otelmetrics.MergeImplies(ksmHostImplies, ksmZoneImplies, otelmetrics.WorkloadImplies()), + }, + Instrumentation: scopeInstrumentation, + }, + }, + ksmWorkloadSchema("ksm-deployment", ksmWorkloadBucket.Deployment, "k8s.deployment.name"), + ksmWorkloadSchema("ksm-daemonset", ksmWorkloadBucket.DaemonSet, "k8s.daemonset.name"), + ksmWorkloadSchema("ksm-statefulset", ksmWorkloadBucket.StatefulSet, "k8s.statefulset.name"), + ksmWorkloadSchema("ksm-job", ksmWorkloadBucket.Job, "k8s.job.name"), + ksmWorkloadSchema("ksm-cronjob", ksmWorkloadBucket.CronJob, "k8s.cronjob.name"), + groupSchema{ + name: "ksm-cluster", + metrics: func() []string { return ksmClusterBucket }, + schema: otelmetrics.GroupSchema{ + Resource: otelmetrics.KeySchema{ + Required: otelmetrics.ConcatKeys(ksmCommonKeys, []string{ + "k8s.namespace.name", // ksm_promote; protected + }), + }, + Instrumentation: scopeInstrumentation, + }, + }, + ) +} diff --git a/test/otel/standard/schema_node_test.go b/test/otel/standard/schema_node_test.go new file mode 100644 index 000000000..2fa8d954f --- /dev/null +++ b/test/otel/standard/schema_node_test.go @@ -0,0 +1,94 @@ +//go:build integration + +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: MIT + +package standard + +import "github.com/aws/amazon-cloudwatch-agent-test/util/otelmetrics" + +// Rows for the node agent's cadvisor and kubeletstats pipelines, one per series +// scope because each scope carries a different set of resource keys. +func init() { + labels := []string{"k8s.node.label.", "k8s.pod.label."} + // The agent DaemonSet runs a pod on every node, so its owner keys are on some series of + // every pod-level row. The other kinds depend on the cluster's workloads. + ownerExpected := otelmetrics.ConcatKeys([]string{"k8s.daemonset.name"}, otelmetrics.WorkloadKeys) + ownerOptional := otelmetrics.Without(otelmetrics.OwnerKindKeys, "k8s.daemonset.name") + podSchema := func(required []string) otelmetrics.KeySchema { + return otelmetrics.KeySchema{ + Required: required, + Expected: ownerExpected, + Optional: ownerOptional, + OpenPrefixes: labels, + Implies: otelmetrics.WorkloadImplies(), + } + } + + cadvisorKeys := otelmetrics.ConcatKeys(otelmetrics.PrometheusReceiverKeys, otelmetrics.CloudKeys, otelmetrics.HostKeys, []string{ + "k8s.cluster.name", // set_cluster_name; protected + "k8s.namespace.name", // cadvisor_promote; protected + "k8s.node.name", // set_node_name, promote_node_name; protected + "k8s.node.uid", // k8sattributes/cw_k8s_ci_v0_node + "k8s.pod.name", // cadvisor_promote; protected + "k8s.pod.uid", // k8sattributes/cw_k8s_ci_v0_pod; protected + }) + kubeletstatsPodKeys := otelmetrics.ConcatKeys(otelmetrics.CloudKeys, otelmetrics.HostKeys, []string{ + "k8s.cluster.name", // set_cluster_name; protected + "k8s.namespace.name", // kubeletstats/cw_k8s_ci_v0; protected + "k8s.node.name", // k8sattributes/cw_k8s_ci_v0_pod; protected + "k8s.node.uid", // k8sattributes/cw_k8s_ci_v0_node + "k8s.pod.name", // kubeletstats/cw_k8s_ci_v0; protected + "k8s.pod.uid", // kubeletstats/cw_k8s_ci_v0, k8sattributes/cw_k8s_ci_v0_pod; protected + }) + + attributeSchemas = append(attributeSchemas, + groupSchema{ + name: "cadvisor-container", + metrics: func() []string { return metricNamesByScope(cadvisorMetrics, otelmetrics.ScopeContainer) }, + schema: otelmetrics.GroupSchema{ + Resource: podSchema(otelmetrics.ConcatKeys(cadvisorKeys, []string{ + "k8s.container.name", // cadvisor_promote; cadvisor_rollup drops series without it; protected + })), + Instrumentation: scopeInstrumentation, + }, + }, + groupSchema{ + // container_network_* is sandbox-only and arrives without `container`, so + // cadvisor_promote never sets k8s.container.name. + name: "cadvisor-pod", + metrics: func() []string { return metricNamesByScope(cadvisorMetrics, otelmetrics.ScopePod) }, + schema: otelmetrics.GroupSchema{Resource: podSchema(cadvisorKeys), Instrumentation: scopeInstrumentation}, + }, + groupSchema{ + name: "kubeletstats-node", + metrics: func() []string { return metricNames(kubeletstatsNodeMetrics) }, + schema: otelmetrics.GroupSchema{ + Resource: otelmetrics.KeySchema{ + Required: otelmetrics.ConcatKeys(otelmetrics.CloudKeys, otelmetrics.HostKeys, []string{ + "k8s.cluster.name", // set_cluster_name; protected + "k8s.node.name", // kubeletstats/cw_k8s_ci_v0; protected + "k8s.node.uid", // k8sattributes/cw_k8s_ci_v0_node + }), + OpenPrefixes: []string{"k8s.node.label."}, + }, + Instrumentation: scopeInstrumentation, + }, + }, + groupSchema{ + name: "kubeletstats-pod", + metrics: func() []string { return metricNames(kubeletstatsPodMetrics) }, + schema: otelmetrics.GroupSchema{Resource: podSchema(kubeletstatsPodKeys), Instrumentation: scopeInstrumentation}, + }, + groupSchema{ + name: "kubeletstats-container", + metrics: func() []string { return metricNames(kubeletstatsContainerMetrics) }, + schema: otelmetrics.GroupSchema{ + Resource: podSchema(otelmetrics.ConcatKeys(kubeletstatsPodKeys, []string{ + "k8s.container.name", // kubeletstats/cw_k8s_ci_v0; protected + })), + Instrumentation: scopeInstrumentation, + }, + }, + ) +} diff --git a/test/otel/standard/schema_test.go b/test/otel/standard/schema_test.go new file mode 100644 index 000000000..d5600be2d --- /dev/null +++ b/test/otel/standard/schema_test.go @@ -0,0 +1,132 @@ +//go:build integration + +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: MIT + +package standard + +import ( + "context" + "strings" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/aws/amazon-cloudwatch-agent-test/util/otelmetrics" +) + +// --------------------------------------------------------------------------- +// TestAttributeSchema — the closed set of resource and instrumentation-scope +// attributes each metric group carries. Every series must have every required +// key and nothing undeclared, so an attribute removed by mistake fails here +// even though no test names it, and so does one that appears unnoticed. Node and +// pod labels are open by construction (k8sattributes extracts "(.*)") and are +// covered by TestNodeLabelsReachMetrics and TestPodLabelsReachMetrics instead. +// +// Every key carries the chart component that sets it. Required keys are on every +// series, expected keys on at least one series of the row, and an implication +// (otelmetrics.KeySchema.Implies) ties keys that are set together, so removing a +// key family from all series fails even where each series may lack it. +// Datapoint scope is not covered: its keys come from the scraped exporter and +// move with its version. Every series in the cluster is read, so the +// clusterHostTypes pin does not limit the check. +// +// An intentional attribute change edits the row in the same change. To add a +// key, list it as optional, merge the chart change, then make it required. Keys +// marked "protected" are on the attribute-limit processor's never-drop list. +// --------------------------------------------------------------------------- + +type groupSchema struct { + name string + metrics func() []string + schema otelmetrics.GroupSchema +} + +var scopeInstrumentation = otelmetrics.KeySchema{Required: otelmetrics.ScopeKeys} + +var attributeSchemas = []groupSchema{ + { + name: "apiserver", + metrics: func() []string { return metricNames(controlPlaneMetrics) }, + schema: otelmetrics.GroupSchema{ + Resource: otelmetrics.KeySchema{Required: otelmetrics.ConcatKeys(otelmetrics.PrometheusReceiverKeys, otelmetrics.CloudKeys, []string{ + "k8s.cluster.name", // set_cluster_name; protected + "k8s.component.name", // set_component, promote_component; protected + "k8s.namespace.name", // Prometheus receiver, from Kubernetes service discovery; protected + })}, + Instrumentation: scopeInstrumentation, + }, + }, + { + name: "node-exporter", + metrics: func() []string { return metricNames(nodeExporterMetrics) }, + schema: otelmetrics.GroupSchema{ + Resource: otelmetrics.KeySchema{ + Required: otelmetrics.ConcatKeys(otelmetrics.PrometheusReceiverKeys, otelmetrics.CloudKeys, otelmetrics.HostKeys, []string{ + "k8s.cluster.name", // set_cluster_name; protected + "k8s.node.name", // set_node_name, promote_node_name; protected + "k8s.node.uid", // k8sattributes/cw_k8s_ci_v0_node + }), + OpenPrefixes: []string{"k8s.node.label."}, + }, + Instrumentation: scopeInstrumentation, + }, + }, +} + +func TestAttributeSchema(t *testing.T) { + for _, g := range attributeSchemas { + g := g + t.Run(g.name, func(t *testing.T) { + deleted := deletedNodes(t) + tally, err := otelmetrics.TallySchema(context.Background(), queryCache, g.metrics(), g.schema, deleted.forgive) + require.NoError(t, err) + t.Logf("%d series across %d metrics; %d series of deleted nodes forgiven a missing k8s.node.uid", + tally.Series, len(g.metrics()), deleted.forgiven) + require.NotZero(t, tally.Series, "no series checked") + for _, p := range tally.Problems() { + t.Errorf("%s: add or change the %q row if intended", p, g.name) + } + }) + } +} + +// deletedNodeForgiveness forgives the one problem a deleted node causes: +// k8sattributes forgets the node, so k8s.node.uid is no longer set, while the +// node's last series stay inside the query lookback window. The API is read only +// when a series lacks k8s.node.uid. +type deletedNodeForgiveness struct { + t *testing.T + live map[string]bool + forgiven int +} + +func deletedNodes(t *testing.T) *deletedNodeForgiveness { + return &deletedNodeForgiveness{t: t} +} + +func (d *deletedNodeForgiveness) forgive(l otelmetrics.MetricLabels, c otelmetrics.SeriesCheck) otelmetrics.SeriesCheck { + node := l.Resource["k8s.node.name"] + if _, hasUID := l.Resource["k8s.node.uid"]; hasUID || node == "" { + return c + } + if d.live == nil { + d.live = map[string]bool{} + for name := range freshGroundTruth(d.t).nodes { + d.live[name] = true + } + } + if d.live[node] { + return c + } + d.forgiven++ + c.Missing = otelmetrics.Without(c.Missing, "k8s.node.uid") + var broken []string + for _, b := range c.Broken { + if !strings.HasSuffix(b, " without k8s.node.uid") { + broken = append(broken, b) + } + } + c.Broken = broken + return c +} diff --git a/util/otelmetrics/attribute_schema.go b/util/otelmetrics/attribute_schema.go new file mode 100644 index 000000000..ceb1ae442 --- /dev/null +++ b/util/otelmetrics/attribute_schema.go @@ -0,0 +1,196 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: MIT + +package otelmetrics + +import ( + "fmt" + "sort" + "strings" +) + +// KeySchema is the closed set of attribute keys one scope of a metric group +// carries. Required keys must be on every series. Expected keys may be absent +// from a series but must be on at least one series of the group, so a change +// that removes them everywhere fails. Optional keys may be absent everywhere. +// Keys under an open prefix (for example "k8s.node.label.") are allowed without +// being listed, because their names come from the cluster rather than the pipeline. +type KeySchema struct { + Required []string + Expected []string + Optional []string + OpenPrefixes []string + // Implies maps a key to keys that every series carrying it must also carry, + // so losing part of a family that is set together fails. + Implies map[string][]string +} + +// SeriesCheck is the result of checking one scope of one series. +type SeriesCheck struct { + Missing []string // required keys absent + Unexpected []string // keys the schema does not allow + Broken []string // "a without b" for each implication that does not hold +} + +// OK reports whether the series matched the schema. +func (c SeriesCheck) OK() bool { + return len(c.Missing) == 0 && len(c.Unexpected) == 0 && len(c.Broken) == 0 +} + +// Check checks one series. Keys starting with "@" are structural (the scope name +// and version) rather than attributes and are ignored. +func (s KeySchema) Check(observed map[string]string) SeriesCheck { + var c SeriesCheck + allowed := make(map[string]struct{}, len(s.Required)+len(s.Expected)+len(s.Optional)) + for _, k := range s.Required { + allowed[k] = struct{}{} + if _, ok := observed[k]; !ok { + c.Missing = append(c.Missing, k) + } + } + for _, k := range concatStrings(s.Expected, s.Optional) { + allowed[k] = struct{}{} + } + for k := range observed { + if strings.HasPrefix(k, "@") { + continue + } + if _, ok := allowed[k]; ok || hasAnyPrefix(k, s.OpenPrefixes) { + continue + } + c.Unexpected = append(c.Unexpected, k) + } + for k, implied := range s.Implies { + if _, ok := observed[k]; !ok { + continue + } + for _, want := range implied { + if _, ok := observed[want]; !ok { + c.Broken = append(c.Broken, k+" without "+want) + } + } + } + sort.Strings(c.Missing) + sort.Strings(c.Unexpected) + sort.Strings(c.Broken) + return c +} + +// Together returns implications under which the keys are all present or all absent. +func Together(keys ...string) map[string][]string { + out := make(map[string][]string, len(keys)) + for _, k := range keys { + for _, other := range keys { + if other != k { + out[k] = append(out[k], other) + } + } + } + return out +} + +// MergeImplies combines implication maps. +func MergeImplies(maps ...map[string][]string) map[string][]string { + out := map[string][]string{} + for _, m := range maps { + for k, v := range m { + out[k] = append(out[k], v...) + } + } + return out +} + +// GroupSchema is the attribute schema of one metric group. Datapoint scope is not +// covered: its keys come from the scraped exporter and move with its version. +type GroupSchema struct { + Resource KeySchema + Instrumentation KeySchema +} + +// SchemaTally counts schema problems over the series of one metric group. +type SchemaTally struct { + schema GroupSchema + Series int + counts map[string]int // series count per problem + seen map[string]bool // "@resource." and "@instrumentation." seen on a recorded series +} + +func NewSchemaTally(schema GroupSchema) *SchemaTally { + return &SchemaTally{schema: schema, counts: map[string]int{}, seen: map[string]bool{}} +} + +// Check checks one series without recording it. +func (t *SchemaTally) Check(l MetricLabels) (resource, instrumentation SeriesCheck) { + return t.schema.Resource.Check(l.Resource), t.schema.Instrumentation.Check(l.Instrumentation) +} + +// Record adds one series and the result of Check for it. +func (t *SchemaTally) Record(l MetricLabels, resource, instrumentation SeriesCheck) { + t.Series++ + for scope, c := range map[string]SeriesCheck{"@resource.": resource, "@instrumentation.": instrumentation} { + for _, k := range c.Missing { + t.counts["missing required "+scope+k]++ + } + for _, k := range c.Unexpected { + t.counts["undeclared "+scope+k]++ + } + for _, b := range c.Broken { + t.counts[scope+strings.Replace(b, " without ", " without "+scope, 1)]++ + } + } + for k := range l.Resource { + t.seen["@resource."+k] = true + } + for k := range l.Instrumentation { + t.seen["@instrumentation."+k] = true + } +} + +// Problems describes every problem with the number of series it affects, then +// every Expected key that no recorded series carried. It is empty when the group passed. +func (t *SchemaTally) Problems() []string { + var out []string + for _, p := range SortedKeys(t.counts) { + out = append(out, fmt.Sprintf("%s on %d of %d series", p, t.counts[p], t.Series)) + } + var unseen []string + for scope, s := range map[string]KeySchema{"@resource.": t.schema.Resource, "@instrumentation.": t.schema.Instrumentation} { + for _, k := range s.Expected { + if !t.seen[scope+k] { + unseen = append(unseen, scope+k) + } + } + } + sort.Strings(unseen) + for _, k := range unseen { + out = append(out, fmt.Sprintf("expected %s on at least one of %d series, found on none", k, t.Series)) + } + return out +} + +// SortedKeys returns the keys of m in order. +func SortedKeys[V any](m map[string]V) []string { + out := make([]string, 0, len(m)) + for k := range m { + out = append(out, k) + } + sort.Strings(out) + return out +} + +func concatStrings(groups ...[]string) []string { + var out []string + for _, g := range groups { + out = append(out, g...) + } + return out +} + +func hasAnyPrefix(s string, prefixes []string) bool { + for _, p := range prefixes { + if strings.HasPrefix(s, p) { + return true + } + } + return false +} diff --git a/util/otelmetrics/attribute_schema_test.go b/util/otelmetrics/attribute_schema_test.go new file mode 100644 index 000000000..fd2997286 --- /dev/null +++ b/util/otelmetrics/attribute_schema_test.go @@ -0,0 +1,125 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: MIT + +package otelmetrics + +import ( + "context" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestKeySchemaCheck(t *testing.T) { + s := KeySchema{ + Required: []string{"k8s.cluster.name", "server.port"}, + Optional: []string{"k8s.node.uid"}, + OpenPrefixes: []string{"k8s.node.label."}, + } + + t.Run("exact set passes", func(t *testing.T) { + require.True(t, s.Check(map[string]string{"k8s.cluster.name": "c", "server.port": "9100"}).OK()) + }) + + t.Run("optional and open-prefix keys are allowed", func(t *testing.T) { + require.True(t, s.Check(map[string]string{ + "k8s.cluster.name": "c", "server.port": "9100", + "k8s.node.uid": "u", "k8s.node.label.kubernetes.io/os": "linux", + }).OK()) + }) + + t.Run("a required key deleted by mistake is missing", func(t *testing.T) { + c := s.Check(map[string]string{"k8s.cluster.name": "c", "net.host.port": "9100"}) + require.Equal(t, []string{"server.port"}, c.Missing) + }) + + t.Run("an undeclared key is unexpected", func(t *testing.T) { + c := s.Check(map[string]string{"k8s.cluster.name": "c", "server.port": "9100", "net.host.port": "9100"}) + require.Equal(t, []string{"net.host.port"}, c.Unexpected) + }) + + t.Run("a key under another prefix is not open", func(t *testing.T) { + c := s.Check(map[string]string{"k8s.cluster.name": "c", "server.port": "9100", "k8s.pod.label.app": "x"}) + require.Equal(t, []string{"k8s.pod.label.app"}, c.Unexpected) + }) + + t.Run("scope name and version are not attributes", func(t *testing.T) { + require.True(t, s.Check(map[string]string{ + "k8s.cluster.name": "c", "server.port": "9100", "@name": "scope", "@version": "1", + }).OK()) + }) +} + +func TestKeySchemaImplies(t *testing.T) { + s := KeySchema{ + Optional: []string{"k8s.deployment.name", "k8s.workload.name", "k8s.workload.type"}, + Implies: MergeImplies( + map[string][]string{"k8s.deployment.name": {"k8s.workload.name", "k8s.workload.type"}}, + Together("k8s.workload.name", "k8s.workload.type"), + ), + } + require.True(t, s.Check(map[string]string{}).OK(), "a bare pod has none of them") + require.True(t, s.Check(map[string]string{ + "k8s.deployment.name": "d", "k8s.workload.name": "d", "k8s.workload.type": "Deployment", + }).OK()) + require.Equal(t, []string{"k8s.deployment.name without k8s.workload.name", "k8s.deployment.name without k8s.workload.type"}, + s.Check(map[string]string{"k8s.deployment.name": "d"}).Broken, "set_workload removed") + require.Equal(t, []string{"k8s.workload.name without k8s.workload.type"}, + s.Check(map[string]string{"k8s.workload.name": "d"}).Broken) +} + +func TestSchemaTally(t *testing.T) { + schema := GroupSchema{ + Resource: KeySchema{ + Required: []string{"k8s.pod.name"}, + Expected: []string{"k8s.node.name"}, + Implies: Together("host.id", "host.name"), + Optional: []string{"host.id", "host.name"}, + }, + Instrumentation: KeySchema{Required: []string{"cloudwatch.source"}}, + } + record := func(tally *SchemaTally, resource map[string]string) { + l := MetricLabels{Resource: resource, Instrumentation: map[string]string{"cloudwatch.source": "s"}} + r, i := tally.Check(l) + tally.Record(l, r, i) + } + + ok := NewSchemaTally(schema) + record(ok, map[string]string{"k8s.pod.name": "p1"}) // unscheduled + record(ok, map[string]string{"k8s.pod.name": "p2", "k8s.node.name": "n"}) + require.Empty(t, ok.Problems()) + + bad := NewSchemaTally(schema) + record(bad, map[string]string{"k8s.pod.name": "p1", "host.id": "i"}) + record(bad, map[string]string{"extra": "x"}) + require.Equal(t, []string{ + "@resource.host.id without @resource.host.name on 1 of 2 series", + "missing required @resource.k8s.pod.name on 1 of 2 series", + "undeclared @resource.extra on 1 of 2 series", + "expected @resource.k8s.node.name on at least one of 2 series, found on none", + }, bad.Problems(), "node keys removed from every series fail even though each series may lack them") +} + +func TestTallySchemaChecksEveryHostType(t *testing.T) { + series := func(hostType string, extra ...string) fakeSeries { + l := map[string]string{ + "__name__": "m", "@resource.k8s.cluster.name": "c1", + "@resource.host.type": hostType, "@resource.k8s.pod.name": "p-" + hostType, + } + for _, k := range extra { + l["@resource."+k] = "x" + } + return fakeSeries{labels: l} + } + _, c := newFake(t, 100, []fakeSeries{series("a"), series("b", "leaked")}) + qc := NewQueryCache(c, "c1", WithHostTypes([]string{"a"})) + routed, err := qc.Get(context.Background(), "m") + require.NoError(t, err) + require.Len(t, routed, 1, "Get sees only the configured host type") + + tally, err := TallySchema(context.Background(), qc, []string{"m"}, GroupSchema{ + Resource: KeySchema{Required: []string{"k8s.cluster.name", "k8s.pod.name"}, Optional: []string{"host.type"}}, + }, nil) + require.NoError(t, err) + require.Equal(t, []string{"undeclared @resource.leaked on 1 of 2 series"}, tally.Problems()) +} diff --git a/util/otelmetrics/label_reconcile.go b/util/otelmetrics/label_reconcile.go new file mode 100644 index 000000000..25eac1d96 --- /dev/null +++ b/util/otelmetrics/label_reconcile.go @@ -0,0 +1,250 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: MIT + +package otelmetrics + +import ( + "fmt" + "sort" + "strings" + + "gopkg.in/yaml.v3" +) + +// AttributeDenylist is the unconditional removal configuration of an +// awsattributelimit processor, plus its attribute cap. +type AttributeDenylist struct { + Prefixes []string + Keys map[string]struct{} + MaxTotal int + Processor string +} + +// ParseAttributeDenylist reads the awsattributelimit processor from a rendered +// collector configuration. It returns an error if the configuration has no such +// processor or more than one, because the expectation would then be ambiguous. +func ParseAttributeDenylist(otelConfigYAML string) (AttributeDenylist, error) { + var cfg struct { + Processors map[string]struct { + Prefixes []string `yaml:"unconditional_removal_prefixes"` + Keys []string `yaml:"unconditional_removal_keys"` + MaxTotal int `yaml:"max_total_attributes"` + } `yaml:"processors"` + } + if err := yaml.Unmarshal([]byte(otelConfigYAML), &cfg); err != nil { + return AttributeDenylist{}, fmt.Errorf("parsing collector configuration: %w", err) + } + var names []string + for name := range cfg.Processors { + if name == "awsattributelimit" || strings.HasPrefix(name, "awsattributelimit/") { + names = append(names, name) + } + } + if len(names) != 1 { + return AttributeDenylist{}, fmt.Errorf("want exactly one awsattributelimit processor, found %d: %v", len(names), names) + } + p := cfg.Processors[names[0]] + d := AttributeDenylist{ + Prefixes: p.Prefixes, + Keys: make(map[string]struct{}, len(p.Keys)), + MaxTotal: p.MaxTotal, + Processor: names[0], + } + for _, k := range p.Keys { + d.Keys[k] = struct{}{} + } + return d, nil +} + +// Denies reports whether the processor removes the resource attribute key, +// using the same exact-key-then-prefix rule as the processor itself. +func (d AttributeDenylist) Denies(key string) bool { + if _, ok := d.Keys[key]; ok { + return true + } + for _, p := range d.Prefixes { + if strings.HasPrefix(key, p) { + return true + } + } + return false +} + +// AtCap reports whether a series is at the processor's attribute cap, where +// tier-based removal can explain a missing attribute. The count matches the +// processor's: resource, scope and datapoint attributes. Keys the backend +// synthesises (the @-prefixed scope name and version, and __x__ datapoint keys) +// are not attributes and are excluded. +func (d AttributeDenylist) AtCap(labels MetricLabels) bool { + if d.MaxTotal <= 0 { + return false + } + n := len(labels.Resource) + for k := range labels.Instrumentation { + if !strings.HasPrefix(k, "@") { + n++ + } + } + for k := range labels.Datapoint { + if !(strings.HasPrefix(k, "__") && strings.HasSuffix(k, "__")) { + n++ + } + } + return n >= d.MaxTotal +} + +// LabelDiff is the result of reconciling an object's Kubernetes labels with the +// resource attributes on one series. +type LabelDiff struct { + Checked int // labels expected on the series + Missing []string // expected but absent + Mismatched []string // present with a different value, as "key: got, want" + Leaked []string // denied but present +} + +// OK reports whether the series matched the object exactly. +func (l LabelDiff) OK() bool { + return len(l.Missing) == 0 && len(l.Mismatched) == 0 && len(l.Leaked) == 0 +} + +// ReconcileLabels compares an object's labels with the attributes under prefix +// (for example "k8s.node.label.") in resource. Every label the denylist does not +// match must arrive with the same value, and every label it matches must not. +// Empty-valued labels are skipped, because PromQL drops empty label values. +func ReconcileLabels(objectLabels, resource map[string]string, prefix string, d AttributeDenylist) LabelDiff { + var diff LabelDiff + for k, want := range objectLabels { + attr := prefix + k + got, present := resource[attr] + if d.Denies(attr) { + if present { + diff.Leaked = append(diff.Leaked, k) + } + continue + } + if want == "" { + continue + } + diff.Checked++ + switch { + case !present: + diff.Missing = append(diff.Missing, k) + case got != want: + diff.Mismatched = append(diff.Mismatched, fmt.Sprintf("%s: got %q, want %q", k, got, want)) + } + } + sort.Strings(diff.Missing) + sort.Strings(diff.Mismatched) + sort.Strings(diff.Leaked) + return diff +} + +// Attributes no denylist entry may remove. The helm-charts render test +// prometheus_attribute_shape.sh holds the same list; keep the two in step. +var ( + // protectedResourceKeys and protectedPrefixes of the awsattributelimit processor; + // unconditional removal runs before that protection, so a denylist entry would override it. + NeverDeniedKeys = []string{ + "k8s.cluster.name", "k8s.node.name", "k8s.pod.name", "k8s.pod.uid", "k8s.namespace.name", + "k8s.container.name", "k8s.deployment.name", "k8s.statefulset.name", "k8s.daemonset.name", + "k8s.replicaset.name", "k8s.job.name", "k8s.cronjob.name", "k8s.workload.name", + "k8s.workload.type", "neurondevice", "neuroncore", "efa.device", "aws.efa.eni.id", + "volume_id", "instance_id", "k8s.pod.label.app.kubernetes.io/name", + "k8s.pod.label.app.kubernetes.io/instance", "k8s.pod.label.app.kubernetes.io/component", + "k8s.component.name", + // Kept on purpose inside a label domain the denylist partly removes, so widening an entry to the domain fails. + "k8s.node.label.kubernetes.io/os", "k8s.node.label.kubernetes.io/arch", + "k8s.node.label.eks.amazonaws.com/nodegroup", "k8s.node.label.eks.amazonaws.com/capacityType", + "k8s.node.label.karpenter.sh/nodepool", "k8s.node.label.karpenter.sh/capacity-type", + "k8s.node.label.karpenter.k8s.aws/ec2nodeclass", + } + NeverDeniedPrefixes = []string{"cloud.", "host.", "hw."} +) + +// NeverDeniedViolations lists each denylist entry that would remove a never-denied attribute. +func (d AttributeDenylist) NeverDeniedViolations() []string { + var out []string + for _, key := range NeverDeniedKeys { + if _, ok := d.Keys[key]; ok { + out = append(out, fmt.Sprintf("key %q", key)) + } + for _, p := range d.Prefixes { + if strings.HasPrefix(key, p) { + out = append(out, fmt.Sprintf("prefix %q removes %q", p, key)) + } + } + } + for _, protected := range NeverDeniedPrefixes { + for key := range d.Keys { + if strings.HasPrefix(key, protected) { + out = append(out, fmt.Sprintf("key %q is under %q", key, protected)) + } + } + for _, p := range d.Prefixes { + if strings.HasPrefix(p, protected) || strings.HasPrefix(protected, p) { + out = append(out, fmt.Sprintf("prefix %q overlaps %q", p, protected)) + } + } + } + sort.Strings(out) + return out +} + +// ObjectReconciliation is the result of reconciling every series of a set of +// Kubernetes objects. An object passes if any one of its series reconciles, +// because a label change or a collector restart leaves an older series inside +// the query lookback window next to the current one. +type ObjectReconciliation struct { + Objects int // objects with at least one series below the cap + Labels int // labels checked, counted once per passing object + AtCap int // series skipped at the attribute cap + Failed map[string]LabelDiff // objects no series reconciled, with their closest series + Required int // objects that must have a series + NoSeries []string // required objects with no series at all, sorted +} + +// ReconcileObjects groups series by object and reconciles each. objectOf returns the +// object key and its Kubernetes labels, or false for a series of an unknown object. +// Every key in required must have at least one series, at the cap or below it. +func ReconcileObjects(results []MetricResult, objectOf func(MetricLabels) (string, map[string]string, bool), + required []string, prefix string, d AttributeDenylist) ObjectReconciliation { + out := ObjectReconciliation{Failed: map[string]LabelDiff{}, Required: len(required)} + passed := map[string]bool{} + hasSeries := map[string]bool{} + for _, r := range results { + key, labels, ok := objectOf(r.Labels) + if !ok { + continue + } + hasSeries[key] = true + if d.AtCap(r.Labels) { + out.AtCap++ + continue + } + if passed[key] { + continue + } + diff := ReconcileLabels(labels, r.Labels.Resource, prefix, d) + if diff.OK() { + passed[key] = true + out.Labels += diff.Checked + delete(out.Failed, key) + continue + } + if cur, seen := out.Failed[key]; !seen || diff.problems() < cur.problems() { + out.Failed[key] = diff + } + } + out.Objects = len(passed) + len(out.Failed) + for _, key := range required { + if !hasSeries[key] { + out.NoSeries = append(out.NoSeries, key) + } + } + sort.Strings(out.NoSeries) + return out +} + +func (l LabelDiff) problems() int { + return len(l.Missing) + len(l.Mismatched) + len(l.Leaked) +} diff --git a/util/otelmetrics/label_reconcile_test.go b/util/otelmetrics/label_reconcile_test.go new file mode 100644 index 000000000..218ea605f --- /dev/null +++ b/util/otelmetrics/label_reconcile_test.go @@ -0,0 +1,204 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: MIT + +package otelmetrics + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +const denylistConfig = ` +processors: + batch/x: {} + awsattributelimit/cw_k8s_ci_v0: + max_total_attributes: 150 + unconditional_removal_prefixes: + - "k8s.node.label.karpenter.k8s.aws/instance-" + unconditional_removal_keys: + - "k8s.node.label.topology.kubernetes.io/zone" + - "k8s.pod.label.pod-template-hash" +` + +func mustDenylist(t *testing.T) AttributeDenylist { + t.Helper() + d, err := ParseAttributeDenylist(denylistConfig) + require.NoError(t, err) + return d +} + +func TestParseAttributeDenylist(t *testing.T) { + d := mustDenylist(t) + require.Equal(t, "awsattributelimit/cw_k8s_ci_v0", d.Processor) + require.Equal(t, 150, d.MaxTotal) + require.Equal(t, []string{"k8s.node.label.karpenter.k8s.aws/instance-"}, d.Prefixes) + require.Len(t, d.Keys, 2) + + _, err := ParseAttributeDenylist("processors:\n batch/x: {}\n") + require.Error(t, err, "no awsattributelimit processor must be an error, not an empty denylist") + + _, err = ParseAttributeDenylist("processors:\n awsattributelimit/a: {}\n awsattributelimit/b: {}\n") + require.Error(t, err, "two processors make the expectation ambiguous") +} + +func TestDenies(t *testing.T) { + d := mustDenylist(t) + require.True(t, d.Denies("k8s.node.label.topology.kubernetes.io/zone")) + require.True(t, d.Denies("k8s.node.label.karpenter.k8s.aws/instance-family")) + require.False(t, d.Denies("k8s.node.label.karpenter.k8s.aws/ec2nodeclass"), "the prefix must not widen to the domain") + require.False(t, d.Denies("k8s.pod.label.topology.kubernetes.io/zone"), "exact keys are scope-specific") +} + +func TestReconcileLabels(t *testing.T) { + d := mustDenylist(t) + node := map[string]string{ + "kubernetes.io/os": "linux", + "karpenter.k8s.aws/ec2nodeclass": "default", + "karpenter.k8s.aws/instance-family": "m5", + "topology.kubernetes.io/zone": "us-east-1a", + "node-role.kubernetes.io/control-plane": "", + } + clean := map[string]string{ + "k8s.node.label.kubernetes.io/os": "linux", + "k8s.node.label.karpenter.k8s.aws/ec2nodeclass": "default", + } + + t.Run("exact match", func(t *testing.T) { + diff := ReconcileLabels(node, clean, "k8s.node.label.", d) + require.True(t, diff.OK(), "%+v", diff) + require.Equal(t, 2, diff.Checked, "denied and empty-valued labels are not checked") + }) + + t.Run("over-broad denylist drops a retained label", func(t *testing.T) { + got := map[string]string{"k8s.node.label.kubernetes.io/os": "linux"} + diff := ReconcileLabels(node, got, "k8s.node.label.", d) + require.Equal(t, []string{"karpenter.k8s.aws/ec2nodeclass"}, diff.Missing) + }) + + t.Run("denied label still present", func(t *testing.T) { + got := map[string]string{"k8s.node.label.karpenter.k8s.aws/instance-family": "m5"} + for k, v := range clean { + got[k] = v + } + diff := ReconcileLabels(node, got, "k8s.node.label.", d) + require.Equal(t, []string{"karpenter.k8s.aws/instance-family"}, diff.Leaked) + }) + + t.Run("value changed in transit", func(t *testing.T) { + got := map[string]string{"k8s.node.label.kubernetes.io/os": "windows", + "k8s.node.label.karpenter.k8s.aws/ec2nodeclass": "default"} + diff := ReconcileLabels(node, got, "k8s.node.label.", d) + require.Len(t, diff.Mismatched, 1) + require.Contains(t, diff.Mismatched[0], "kubernetes.io/os") + }) + + t.Run("labels under another prefix are ignored", func(t *testing.T) { + got := map[string]string{"k8s.pod.label.kubernetes.io/os": "linux"} + diff := ReconcileLabels(node, got, "k8s.node.label.", d) + require.Len(t, diff.Missing, 2, "a pod label must not satisfy a node label") + }) +} + +func TestAtCap(t *testing.T) { + d := AttributeDenylist{MaxTotal: 3} + l := NewMetricLabels() + l.Resource["a"], l.Resource["b"] = "1", "2" + l.Instrumentation["@name"] = "scope" + l.Datapoint["__type__"] = "gauge" + require.False(t, d.AtCap(l), "scope name and backend keys are not attributes") + l.Datapoint["c"] = "3" + require.True(t, d.AtCap(l)) + require.False(t, AttributeDenylist{}.AtCap(l), "no cap configured") +} + +func TestNeverDeniedViolations(t *testing.T) { + require.Empty(t, mustDenylist(t).NeverDeniedViolations()) + + for name, tc := range map[string]struct { + keys, prefixes []string + want []string + }{ + "protected key": {keys: []string{"k8s.cluster.name"}, want: []string{`key "k8s.cluster.name"`}}, + "widened label domain": {prefixes: []string{"k8s.node.label.kubernetes.io/"}, want: []string{`prefix "k8s.node.label.kubernetes.io/" removes "k8s.node.label.kubernetes.io/arch"`, `prefix "k8s.node.label.kubernetes.io/" removes "k8s.node.label.kubernetes.io/os"`}}, + "key under protected prefix": {keys: []string{"cloud.availability_zone"}, want: []string{`key "cloud.availability_zone" is under "cloud."`}}, + "prefix covering protected prefix": {prefixes: []string{"ho"}, want: []string{`prefix "ho" overlaps "host."`}}, + } { + t.Run(name, func(t *testing.T) { + d := AttributeDenylist{Prefixes: tc.prefixes, Keys: map[string]struct{}{}} + for _, k := range tc.keys { + d.Keys[k] = struct{}{} + } + require.Equal(t, tc.want, d.NeverDeniedViolations()) + }) + } + + d := AttributeDenylist{Prefixes: []string{"k8s.node.label.karpenter.k8s.aws/instance-"}, Keys: map[string]struct{}{}} + require.Empty(t, d.NeverDeniedViolations(), "the instance- prefix spares karpenter.k8s.aws/ec2nodeclass") +} + +func TestReconcileObjects(t *testing.T) { + d := mustDenylist(t) + nodes := map[string]map[string]string{ + "n1": {"kubernetes.io/os": "linux", "role": "new"}, + "n2": {"kubernetes.io/os": "linux", "topology.kubernetes.io/zone": "a"}, + } + series := func(node string, attrs map[string]string) MetricResult { + res := map[string]string{"k8s.node.name": node} + for k, v := range attrs { + res["k8s.node.label."+k] = v + } + return MetricResult{Labels: MetricLabels{Resource: res}} + } + objectOf := func(l MetricLabels) (string, map[string]string, bool) { + name := l.Resource["k8s.node.name"] + labels, ok := nodes[name] + return name, labels, ok + } + + got := ReconcileObjects([]MetricResult{ + series("n1", map[string]string{"kubernetes.io/os": "linux", "role": "old"}), // before a relabel + series("n1", map[string]string{"kubernetes.io/os": "linux", "role": "new"}), + series("n2", map[string]string{"topology.kubernetes.io/zone": "a"}), // missing and leaked + series("n2", map[string]string{"kubernetes.io/os": "windows"}), // mismatched only + series("gone", map[string]string{}), + }, objectOf, []string{"n1", "n2"}, "k8s.node.label.", d) + + require.Equal(t, 2, got.Objects) + require.Equal(t, 2, got.Required) + require.Empty(t, got.NoSeries) + require.Equal(t, 2, got.Labels, "only the passing n1 series counts") + require.Equal(t, map[string]LabelDiff{"n2": {Checked: 1, Mismatched: []string{`kubernetes.io/os: got "windows", want "linux"`}}}, + got.Failed, "n2 keeps its closest series; the older n1 series no longer counts against it") +} + +func TestReconcileObjectsRequiresASeries(t *testing.T) { + nodes := map[string]map[string]string{"n1": {}, "n2": {}, "n3": {}, "capped": {}} + series := func(node string, extra ...string) MetricResult { + res := map[string]string{"k8s.node.name": node} + for _, k := range extra { + res[k] = "x" + } + return MetricResult{Labels: MetricLabels{Resource: res}} + } + objectOf := func(l MetricLabels) (string, map[string]string, bool) { + name := l.Resource["k8s.node.name"] + labels, ok := nodes[name] + return name, labels, ok + } + + got := ReconcileObjects([]MetricResult{ + series("n1"), + series("capped", "a", "b"), // at the cap of 3, so not reconciled, but still a series + series("gone"), // an object the API no longer lists + }, objectOf, []string{"n3", "n1", "capped", "n2"}, "k8s.node.label.", AttributeDenylist{MaxTotal: 3}) + + require.Equal(t, 1, got.Objects) + require.Equal(t, 1, got.AtCap) + require.Equal(t, 4, got.Required) + require.Equal(t, []string{"n2", "n3"}, got.NoSeries, "sorted, and a series at the cap counts as a series") + require.Empty(t, got.Failed) + + got = ReconcileObjects([]MetricResult{series("n1")}, objectOf, nil, "k8s.node.label.", AttributeDenylist{}) + require.Empty(t, got.NoSeries, "with nothing required, an object without a series passes") +} diff --git a/util/otelmetrics/schema_keys.go b/util/otelmetrics/schema_keys.go new file mode 100644 index 000000000..70d010f8c --- /dev/null +++ b/util/otelmetrics/schema_keys.go @@ -0,0 +1,90 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: MIT + +package otelmetrics + +import ( + "context" + "fmt" +) + +// Attribute keys the OTel Container Insights configuration sets on many metric +// groups, shared by the attribute schema rows of every suite. +var ( + // Set by the Prometheus receiver on every scrape target. + PrometheusReceiverKeys = []string{"server.address", "server.port", "service.instance.id", "service.name", "url.scheme"} + // Set by resourcedetection/cw_k8s_ci_v0 (eks, ec2) and set_cloud_resource_id. + CloudKeys = []string{ + "cloud.account.id", "cloud.availability_zone", "cloud.platform", "cloud.provider", + "cloud.region", "cloud.resource_id", + } + // Set by resourcedetection/cw_k8s_ci_v0 (ec2) on the node agent, and by nodemetadataenricher + // on the cluster scraper while the node's Lease is fresh. + HostKeys = []string{"host.id", "host.image.id", "host.name", "host.type"} + // Set by the set_scope_* transform of every pipeline. + ScopeKeys = []string{"cloudwatch.pipeline", "cloudwatch.solution", "cloudwatch.source"} + // Set by k8sattributes/cw_k8s_ci_v0_pod from the pod's owner chain, so a bare pod has none; protected. + OwnerKindKeys = []string{ + "k8s.cronjob.name", "k8s.daemonset.name", "k8s.deployment.name", "k8s.job.name", + "k8s.replicaset.name", "k8s.statefulset.name", + } + // Set by set_workload from the owner kind keys; protected. + WorkloadKeys = []string{"k8s.workload.name", "k8s.workload.type"} +) + +// WorkloadImplies is set_workload's rule: any owner kind key sets both workload keys. +func WorkloadImplies() map[string][]string { + out := Together(WorkloadKeys...) + for _, k := range OwnerKindKeys { + out[k] = append(out[k], WorkloadKeys...) + } + return out +} + +// ConcatKeys joins key lists. +func ConcatKeys(groups ...[]string) []string { + return concatStrings(groups...) +} + +// Without returns keys minus the removed ones. +func Without(keys []string, removed ...string) []string { + drop := make(map[string]struct{}, len(removed)) + for _, k := range removed { + drop[k] = struct{}{} + } + var out []string + for _, k := range keys { + if _, ok := drop[k]; !ok { + out = append(out, k) + } + } + return out +} + +// Forgive may clear the problems of one series that a known transient state explains. +type Forgive func(l MetricLabels, resource SeriesCheck) SeriesCheck + +// TallySchema checks every series of metrics in the cluster against schema. It does not +// use Get, whose host-type routing would hide the series of other instance types and, for +// cluster-scoped metrics, any series that gained host keys. It fails if a query fails or a +// metric has no series, because an empty group would pass vacuously. +func TallySchema(ctx context.Context, qc *QueryCache, metrics []string, schema GroupSchema, forgive Forgive) (*SchemaTally, error) { + tally := NewSchemaTally(schema) + for _, metricName := range metrics { + results, err := qc.GetWithFilter(ctx, metricName, nil) + if err != nil { + return nil, fmt.Errorf("querying %s: %w", metricName, err) + } + if len(results) == 0 { + return nil, fmt.Errorf("%s not available", metricName) + } + for _, r := range results { + resource, instrumentation := tally.Check(r.Labels) + if forgive != nil && !resource.OK() { + resource = forgive(r.Labels, resource) + } + tally.Record(r.Labels, resource, instrumentation) + } + } + return tally, nil +}