Do not cache empty query results in QueryCache

util/otelmetrics/query_cache.go

@@ -90,8 +90,12 @@ func (qc *QueryCache) Get(ctx context.Context, metricName string) ([]MetricResul
 		qc.mu.RUnlock()
 		<-ch // wait for the fetch to complete
 		qc.mu.RLock()
-		entry := qc.filtered[metricName]
+		entry, ok := qc.filtered[metricName]
 		qc.mu.RUnlock()
+		if !ok {
+			// Empty result was not cached; caller should retry.
+			return nil, nil
+		}
 		return entry.results, entry.err
 	}
 	qc.mu.RUnlock()
@@ -107,8 +111,12 @@ func (qc *QueryCache) Get(ctx context.Context, metricName string) ([]MetricResul
 		qc.mu.Unlock()
 		<-ch
 		qc.mu.RLock()
-		entry := qc.filtered[metricName]
+		entry, ok := qc.filtered[metricName]
 		qc.mu.RUnlock()
+		if !ok {
+			// Empty result was not cached; caller should retry.
+			return nil, nil
+		}
 		return entry.results, entry.err
 	}
 	ch := make(chan struct{})
@@ -118,9 +126,13 @@ func (qc *QueryCache) Get(ctx context.Context, metricName string) ([]MetricResul
 	// Fetch without holding any lock
 	entry := qc.fetchFiltered(ctx, metricName)
 
-	// Store and signal waiters
+	// Store and signal waiters.
+	// Do not cache empty successful results; a transient miss would poison
+	// all later lookups for this metric within the same test binary run.
 	qc.mu.Lock()
-	qc.filtered[metricName] = entry
+	if len(entry.results) > 0 || entry.err != nil {
+		qc.filtered[metricName] = entry
+	}
 	delete(qc.inflight, metricName)
 	qc.mu.Unlock()
 	close(ch)
@@ -249,8 +261,12 @@ func (qc *QueryCache) GetUnfiltered(ctx context.Context, metricName string) ([]M
 	results, queryErr := qc.client.Query(ctx, metricName)
 	entry := cacheEntry{results: results, err: queryErr}
 
+	// Do not cache empty successful results; a transient miss would poison
+	// all later lookups for this metric within the same test binary run.
 	qc.mu.Lock()
-	qc.unfiltered[metricName] = entry
+	if len(entry.results) > 0 || entry.err != nil {
+		qc.unfiltered[metricName] = entry
+	}
 	qc.mu.Unlock()
 
 	return entry.results, entry.err

util/otelmetrics/query_cache_test.go

@@ -0,0 +1,260 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: MIT
+
+package otelmetrics
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"sync"
+	"testing"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	v4 "github.com/aws/aws-sdk-go-v2/aws/signer/v4"
+)
+
+// newTestClient creates an OtelMetricsClient pointing at a test HTTP server.
+// Uses static credentials and a real signer; the test server ignores signatures.
+func newTestClient(url string) *OtelMetricsClient {
+	return &OtelMetricsClient{
+		httpClient:     &http.Client{},
+		signer:         v4.NewSigner(),
+		creds:          aws.CredentialsProviderFunc(func(ctx context.Context) (aws.Credentials, error) {
+			return aws.Credentials{AccessKeyID: "AKID", SecretAccessKey: "SECRET", SessionToken: "TOKEN"}, nil
+		}),
+		queryURL:       url + "/api/v1/query",
+		region:         "us-west-2",
+		signingService: "aps",
+		maxRetries:     1,
+	}
+}
+
+// promqlResponseJSON builds a minimal PromQL JSON response body.
+func promqlResponseJSON(series []map[string]string) string {
+	type result struct {
+		Metric map[string]string `json:"metric"`
+		Value  []json.RawMessage `json:"value"`
+	}
+	type data struct {
+		ResultType string   `json:"resultType"`
+		Result     []result `json:"result"`
+	}
+	type response struct {
+		Status string `json:"status"`
+		Data   data   `json:"data"`
+	}
+
+	results := make([]result, 0, len(series))
+	for _, labels := range series {
+		results = append(results, result{
+			Metric: labels,
+			Value:  []json.RawMessage{[]byte(`1234567890`), []byte(`"1.0"`)},
+		})
+	}
+
+	resp := response{
+		Status: "success",
+		Data:   data{ResultType: "vector", Result: results},
+	}
+	b, _ := json.Marshal(resp)
+	return string(b)
+}
+
+func TestQueryCache_EmptyResultNotStored(t *testing.T) {
+	// Server returns an empty result set (no series).
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		fmt.Fprint(w, promqlResponseJSON(nil))
+	}))
+	defer srv.Close()
+
+	client := newTestClient(srv.URL)
+	qc := NewQueryCache(client, "test-cluster")
+
+	results, err := qc.Get(context.Background(), "empty_metric")
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if len(results) != 0 {
+		t.Fatalf("expected empty results, got %d", len(results))
+	}
+
+	// Verify the entry was NOT stored in the cache.
+	qc.mu.RLock()
+	_, cached := qc.filtered["empty_metric"]
+	qc.mu.RUnlock()
+	if cached {
+		t.Fatal("empty result should not be cached")
+	}
+}
+
+func TestQueryCache_NonEmptyResultStored(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		fmt.Fprint(w, promqlResponseJSON([]map[string]string{
+			{"__name__": "cpu_usage", "host": "node-1"},
+		}))
+	}))
+	defer srv.Close()
+
+	client := newTestClient(srv.URL)
+	qc := NewQueryCache(client, "test-cluster")
+
+	results, err := qc.Get(context.Background(), "cpu_usage")
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if len(results) != 1 {
+		t.Fatalf("expected 1 result, got %d", len(results))
+	}
+
+	// Verify the entry WAS stored in the cache.
+	qc.mu.RLock()
+	_, cached := qc.filtered["cpu_usage"]
+	qc.mu.RUnlock()
+	if !cached {
+		t.Fatal("non-empty result should be cached")
+	}
+}
+
+func TestQueryCache_ErrorResultStored(t *testing.T) {
+	// Server returns HTTP 500 to trigger an error.
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		http.Error(w, "internal error", http.StatusInternalServerError)
+	}))
+	defer srv.Close()
+
+	client := newTestClient(srv.URL)
+	qc := NewQueryCache(client, "test-cluster")
+
+	_, err := qc.Get(context.Background(), "error_metric")
+	if err == nil {
+		t.Fatal("expected error, got nil")
+	}
+
+	// Verify the error entry WAS stored in the cache.
+	qc.mu.RLock()
+	entry, cached := qc.filtered["error_metric"]
+	qc.mu.RUnlock()
+	if !cached {
+		t.Fatal("error result should be cached")
+	}
+	if entry.err == nil {
+		t.Fatal("cached entry should contain the error")
+	}
+}
+
+func TestQueryCache_WaiterGetsNilOnEmptyResult(t *testing.T) {
+	// Simulate the waiter path: register an inflight channel,
+	// close it without storing (empty result), verify waiter gets nil, nil.
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		fmt.Fprint(w, promqlResponseJSON(nil))
+	}))
+	defer srv.Close()
+
+	client := newTestClient(srv.URL)
+	qc := NewQueryCache(client, "test-cluster")
+
+	ch := make(chan struct{})
+	qc.mu.Lock()
+	qc.inflight["waiter_metric"] = ch
+	qc.mu.Unlock()
+
+	var waiterResults []MetricResult
+	var waiterErr error
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		<-ch
+		qc.mu.RLock()
+		entry, ok := qc.filtered["waiter_metric"]
+		qc.mu.RUnlock()
+		if !ok {
+			waiterResults = nil
+			waiterErr = nil
+			return
+		}
+		waiterResults = entry.results
+		waiterErr = entry.err
+	}()
+
+	// Simulate: fetcher got empty, did NOT store, cleans up inflight and closes ch.
+	qc.mu.Lock()
+	delete(qc.inflight, "waiter_metric")
+	qc.mu.Unlock()
+	close(ch)
+
+	wg.Wait()
+
+	if waiterResults != nil {
+		t.Fatalf("waiter expected nil results, got %v", waiterResults)
+	}
+	if waiterErr != nil {
+		t.Fatalf("waiter expected nil error, got %v", waiterErr)
+	}
+}
+
+func TestQueryCache_GetUnfiltered_EmptyNotStored(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		fmt.Fprint(w, promqlResponseJSON(nil))
+	}))
+	defer srv.Close()
+
+	client := newTestClient(srv.URL)
+	qc := NewQueryCache(client, "test-cluster")
+
+	results, err := qc.GetUnfiltered(context.Background(), "empty_unfiltered")
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if len(results) != 0 {
+		t.Fatalf("expected empty results, got %d", len(results))
+	}
+
+	qc.mu.RLock()
+	_, cached := qc.unfiltered["empty_unfiltered"]
+	qc.mu.RUnlock()
+	if cached {
+		t.Fatal("empty unfiltered result should not be cached")
+	}
+}
+
+func TestQueryCache_GetUnfiltered_ErrorStored(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		http.Error(w, "internal error", http.StatusInternalServerError)
+	}))
+	defer srv.Close()
+
+	client := newTestClient(srv.URL)
+	qc := NewQueryCache(client, "test-cluster")
+
+	_, err := qc.GetUnfiltered(context.Background(), "error_unfiltered")
+	if err == nil {
+		t.Fatal("expected error, got nil")
+	}
+
+	qc.mu.RLock()
+	entry, cached := qc.unfiltered["error_unfiltered"]
+	qc.mu.RUnlock()
+	if !cached {
+		t.Fatal("error unfiltered result should be cached")
+	}
+	if entry.err == nil {
+		t.Fatal("cached entry should contain the error")
+	}
+}
+
+func TestPromqlMetricSelector(t *testing.T) {
+	got := promqlMetricSelector("node.cpu.usage")
+	want := `{"__name__"="node.cpu.usage",`
+	if got != want {
+		t.Fatalf("promqlMetricSelector(dotted) = %q, want %q", got, want)
+	}
+	got = promqlMetricSelector("cpu_usage")
+	want = `cpu_usage{`
+	if got != want {
+		t.Fatalf("promqlMetricSelector(plain) = %q, want %q", got, want)
+	}
+}