Skip to content

fix(otelmetrics): return complete query results and honor Retry-After #792

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
petruanica wants to merge 1 commit into from
+1,654 −64
Open

fix(otelmetrics): return complete query results and honor Retry-After #792

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion test/otel/efa/dedup_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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),
Expand Down
2 changes: 1 addition & 1 deletion test/otel/efa/efa_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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?")
Expand Down
2 changes: 1 addition & 1 deletion test/otel/efa/node_color_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
2 changes: 1 addition & 1 deletion test/otel/filters/filters_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -368,7 +368,7 @@ func query(ctx context.Context, metric string, resource map[string]string) ([]ot
for key, value := range resource {
selector += fmt.Sprintf(`,%q=%q`, key, value)
}
return client.Query(ctx, selector+"}")
return client.QueryComplete(ctx, selector+"}")
}

// logRecord is the part of an OTLP log record, as stored in CloudWatch Logs,
Expand Down
2 changes: 1 addition & 1 deletion test/otel/gpu/dcgm_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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?",
Expand Down
2 changes: 1 addition & 1 deletion test/otel/gpu/dedup_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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),
Expand Down
2 changes: 1 addition & 1 deletion test/otel/gpu/node_color_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
6 changes: 3 additions & 3 deletions test/otel/gpu_filters/gpu_filters_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -118,7 +118,7 @@ func settled(t *testing.T, metric string) []otelmetrics.MetricResult {
if i > 0 {
time.Sleep(retryInterval)
}
results, err := client.Query(context.Background(), selector)
results, err := client.QueryComplete(context.Background(), selector)
if err != nil {
problems = []string{err.Error()}
continue
Expand Down Expand Up @@ -155,7 +155,7 @@ func TestMetricNameFilterCase(t *testing.T) {
time.Sleep(retryInterval)
}
var results []otelmetrics.MetricResult
results, err = client.Query(context.Background(), nodeSelector(kept))
results, err = client.QueryComplete(context.Background(), nodeSelector(kept))
gpus = map[string]bool{}
for _, r := range results {
if g := uuid(r); g != "" {
Expand All @@ -166,7 +166,7 @@ func TestMetricNameFilterCase(t *testing.T) {
require.NoError(t, err, "querying %s", kept)
require.Len(t, gpus, expectedGPUCount, "%s GPUs from node %s after %d attempts", kept, node, attempts)

results, err := client.Query(context.Background(),
results, err := client.QueryComplete(context.Background(),
fmt.Sprintf(`{"__name__"=%q,"@resource.k8s.cluster.name"=%q}`, dropped, cfg.ClusterName))
require.NoError(t, err, "querying %s", dropped)
assert.Empty(t, results, "%s present, but otelFilters.metricNames.exclude has it", dropped)
Expand Down
2 changes: 1 addition & 1 deletion test/otel/multi_efa_filters/multi_efa_filters_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -180,7 +180,7 @@ func settled(t *testing.T, metric, nodeName string, checkFn func([]otelmetrics.M
continue
}
}
results, err := client.Query(context.Background(), selector)
results, err := client.QueryComplete(context.Background(), selector)
if err != nil {
problems = []string{err.Error()}
continue
Expand Down
4 changes: 2 additions & 2 deletions test/otel/neuron/multi_device_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)))
Expand All @@ -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)))

Expand Down
2 changes: 1 addition & 1 deletion test/otel/neuron/neuron_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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?",
Expand Down
4 changes: 2 additions & 2 deletions test/otel/neuron/node_color_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand All @@ -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)
Expand Down
2 changes: 1 addition & 1 deletion test/otel/neuron_filters/neuron_filters_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -173,7 +173,7 @@ func settled(t *testing.T, metric string) []otelmetrics.MetricResult {
if i > 0 {
time.Sleep(retryInterval)
}
results, err := client.Query(context.Background(), selector)
results, err := client.QueryComplete(context.Background(), selector)
if err != nil {
problems = []string{err.Error()}
continue
Expand Down
2 changes: 1 addition & 1 deletion test/otel/performance/performance_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
4 changes: 2 additions & 2 deletions test/otel/standard/cadvisor_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -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)
})
Expand Down
10 changes: 5 additions & 5 deletions test/otel/standard/dedup_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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")

Expand All @@ -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),
Expand All @@ -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")

Expand All @@ -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),
Expand Down Expand Up @@ -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),
Expand Down
12 changes: 6 additions & 6 deletions test/otel/standard/ksm_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -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 {
Expand All @@ -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 {
Expand All @@ -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)
Expand All @@ -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 {
Expand Down Expand Up @@ -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 {
Expand Down
2 changes: 1 addition & 1 deletion test/otel/standard/kubeletstats_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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?",
Expand Down
4 changes: 2 additions & 2 deletions test/otel/standard/labels_common_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -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)
Expand Down
2 changes: 1 addition & 1 deletion test/otel/standard/node_exporter_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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?",
Expand Down
Loading