Skip to content

Commit 8d716a4

Browse files
committed
fix(otelmetrics): address review nits on query cache
- Add comma-ok checks at waiter map read sites so absent entries (empty result not cached) return nil, nil for retry - Add query_cache_test.go with 7 unit tests covering the cache predicate: empty-not-stored, non-empty-stored, error-stored, waiter-gets-nil-on-empty, unfiltered variants, and promqlMetricSelector
1 parent aafb931 commit 8d716a4

2 files changed

Lines changed: 270 additions & 2 deletions

File tree

util/otelmetrics/query_cache.go

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -90,8 +90,12 @@ func (qc *QueryCache) Get(ctx context.Context, metricName string) ([]MetricResul
9090
qc.mu.RUnlock()
9191
<-ch // wait for the fetch to complete
9292
qc.mu.RLock()
93-
entry := qc.filtered[metricName]
93+
entry, ok := qc.filtered[metricName]
9494
qc.mu.RUnlock()
95+
if !ok {
96+
// Empty result was not cached; caller should retry.
97+
return nil, nil
98+
}
9599
return entry.results, entry.err
96100
}
97101
qc.mu.RUnlock()
@@ -107,8 +111,12 @@ func (qc *QueryCache) Get(ctx context.Context, metricName string) ([]MetricResul
107111
qc.mu.Unlock()
108112
<-ch
109113
qc.mu.RLock()
110-
entry := qc.filtered[metricName]
114+
entry, ok := qc.filtered[metricName]
111115
qc.mu.RUnlock()
116+
if !ok {
117+
// Empty result was not cached; caller should retry.
118+
return nil, nil
119+
}
112120
return entry.results, entry.err
113121
}
114122
ch := make(chan struct{})
Lines changed: 260 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,260 @@
1+
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
2+
// SPDX-License-Identifier: MIT
3+
4+
package otelmetrics
5+
6+
import (
7+
"context"
8+
"encoding/json"
9+
"fmt"
10+
"net/http"
11+
"net/http/httptest"
12+
"sync"
13+
"testing"
14+
15+
"github.com/aws/aws-sdk-go-v2/aws"
16+
v4 "github.com/aws/aws-sdk-go-v2/aws/signer/v4"
17+
)
18+
19+
// newTestClient creates an OtelMetricsClient pointing at a test HTTP server.
20+
// Uses static credentials and a real signer; the test server ignores signatures.
21+
func newTestClient(url string) *OtelMetricsClient {
22+
return &OtelMetricsClient{
23+
httpClient: &http.Client{},
24+
signer: v4.NewSigner(),
25+
creds: aws.CredentialsProviderFunc(func(ctx context.Context) (aws.Credentials, error) {
26+
return aws.Credentials{AccessKeyID: "AKID", SecretAccessKey: "SECRET", SessionToken: "TOKEN"}, nil
27+
}),
28+
queryURL: url + "/api/v1/query",
29+
region: "us-west-2",
30+
signingService: "aps",
31+
maxRetries: 1,
32+
}
33+
}
34+
35+
// promqlResponseJSON builds a minimal PromQL JSON response body.
36+
func promqlResponseJSON(series []map[string]string) string {
37+
type result struct {
38+
Metric map[string]string `json:"metric"`
39+
Value []json.RawMessage `json:"value"`
40+
}
41+
type data struct {
42+
ResultType string `json:"resultType"`
43+
Result []result `json:"result"`
44+
}
45+
type response struct {
46+
Status string `json:"status"`
47+
Data data `json:"data"`
48+
}
49+
50+
results := make([]result, 0, len(series))
51+
for _, labels := range series {
52+
results = append(results, result{
53+
Metric: labels,
54+
Value: []json.RawMessage{[]byte(`1234567890`), []byte(`"1.0"`)},
55+
})
56+
}
57+
58+
resp := response{
59+
Status: "success",
60+
Data: data{ResultType: "vector", Result: results},
61+
}
62+
b, _ := json.Marshal(resp)
63+
return string(b)
64+
}
65+
66+
func TestQueryCache_EmptyResultNotStored(t *testing.T) {
67+
// Server returns an empty result set (no series).
68+
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
69+
fmt.Fprint(w, promqlResponseJSON(nil))
70+
}))
71+
defer srv.Close()
72+
73+
client := newTestClient(srv.URL)
74+
qc := NewQueryCache(client, "test-cluster")
75+
76+
results, err := qc.Get(context.Background(), "empty_metric")
77+
if err != nil {
78+
t.Fatalf("unexpected error: %v", err)
79+
}
80+
if len(results) != 0 {
81+
t.Fatalf("expected empty results, got %d", len(results))
82+
}
83+
84+
// Verify the entry was NOT stored in the cache.
85+
qc.mu.RLock()
86+
_, cached := qc.filtered["empty_metric"]
87+
qc.mu.RUnlock()
88+
if cached {
89+
t.Fatal("empty result should not be cached")
90+
}
91+
}
92+
93+
func TestQueryCache_NonEmptyResultStored(t *testing.T) {
94+
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
95+
fmt.Fprint(w, promqlResponseJSON([]map[string]string{
96+
{"__name__": "cpu_usage", "host": "node-1"},
97+
}))
98+
}))
99+
defer srv.Close()
100+
101+
client := newTestClient(srv.URL)
102+
qc := NewQueryCache(client, "test-cluster")
103+
104+
results, err := qc.Get(context.Background(), "cpu_usage")
105+
if err != nil {
106+
t.Fatalf("unexpected error: %v", err)
107+
}
108+
if len(results) != 1 {
109+
t.Fatalf("expected 1 result, got %d", len(results))
110+
}
111+
112+
// Verify the entry WAS stored in the cache.
113+
qc.mu.RLock()
114+
_, cached := qc.filtered["cpu_usage"]
115+
qc.mu.RUnlock()
116+
if !cached {
117+
t.Fatal("non-empty result should be cached")
118+
}
119+
}
120+
121+
func TestQueryCache_ErrorResultStored(t *testing.T) {
122+
// Server returns HTTP 500 to trigger an error.
123+
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
124+
http.Error(w, "internal error", http.StatusInternalServerError)
125+
}))
126+
defer srv.Close()
127+
128+
client := newTestClient(srv.URL)
129+
qc := NewQueryCache(client, "test-cluster")
130+
131+
_, err := qc.Get(context.Background(), "error_metric")
132+
if err == nil {
133+
t.Fatal("expected error, got nil")
134+
}
135+
136+
// Verify the error entry WAS stored in the cache.
137+
qc.mu.RLock()
138+
entry, cached := qc.filtered["error_metric"]
139+
qc.mu.RUnlock()
140+
if !cached {
141+
t.Fatal("error result should be cached")
142+
}
143+
if entry.err == nil {
144+
t.Fatal("cached entry should contain the error")
145+
}
146+
}
147+
148+
func TestQueryCache_WaiterGetsNilOnEmptyResult(t *testing.T) {
149+
// Simulate the waiter path: register an inflight channel,
150+
// close it without storing (empty result), verify waiter gets nil, nil.
151+
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
152+
fmt.Fprint(w, promqlResponseJSON(nil))
153+
}))
154+
defer srv.Close()
155+
156+
client := newTestClient(srv.URL)
157+
qc := NewQueryCache(client, "test-cluster")
158+
159+
ch := make(chan struct{})
160+
qc.mu.Lock()
161+
qc.inflight["waiter_metric"] = ch
162+
qc.mu.Unlock()
163+
164+
var waiterResults []MetricResult
165+
var waiterErr error
166+
var wg sync.WaitGroup
167+
wg.Add(1)
168+
go func() {
169+
defer wg.Done()
170+
<-ch
171+
qc.mu.RLock()
172+
entry, ok := qc.filtered["waiter_metric"]
173+
qc.mu.RUnlock()
174+
if !ok {
175+
waiterResults = nil
176+
waiterErr = nil
177+
return
178+
}
179+
waiterResults = entry.results
180+
waiterErr = entry.err
181+
}()
182+
183+
// Simulate: fetcher got empty, did NOT store, cleans up inflight and closes ch.
184+
qc.mu.Lock()
185+
delete(qc.inflight, "waiter_metric")
186+
qc.mu.Unlock()
187+
close(ch)
188+
189+
wg.Wait()
190+
191+
if waiterResults != nil {
192+
t.Fatalf("waiter expected nil results, got %v", waiterResults)
193+
}
194+
if waiterErr != nil {
195+
t.Fatalf("waiter expected nil error, got %v", waiterErr)
196+
}
197+
}
198+
199+
func TestQueryCache_GetUnfiltered_EmptyNotStored(t *testing.T) {
200+
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
201+
fmt.Fprint(w, promqlResponseJSON(nil))
202+
}))
203+
defer srv.Close()
204+
205+
client := newTestClient(srv.URL)
206+
qc := NewQueryCache(client, "test-cluster")
207+
208+
results, err := qc.GetUnfiltered(context.Background(), "empty_unfiltered")
209+
if err != nil {
210+
t.Fatalf("unexpected error: %v", err)
211+
}
212+
if len(results) != 0 {
213+
t.Fatalf("expected empty results, got %d", len(results))
214+
}
215+
216+
qc.mu.RLock()
217+
_, cached := qc.unfiltered["empty_unfiltered"]
218+
qc.mu.RUnlock()
219+
if cached {
220+
t.Fatal("empty unfiltered result should not be cached")
221+
}
222+
}
223+
224+
func TestQueryCache_GetUnfiltered_ErrorStored(t *testing.T) {
225+
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
226+
http.Error(w, "internal error", http.StatusInternalServerError)
227+
}))
228+
defer srv.Close()
229+
230+
client := newTestClient(srv.URL)
231+
qc := NewQueryCache(client, "test-cluster")
232+
233+
_, err := qc.GetUnfiltered(context.Background(), "error_unfiltered")
234+
if err == nil {
235+
t.Fatal("expected error, got nil")
236+
}
237+
238+
qc.mu.RLock()
239+
entry, cached := qc.unfiltered["error_unfiltered"]
240+
qc.mu.RUnlock()
241+
if !cached {
242+
t.Fatal("error unfiltered result should be cached")
243+
}
244+
if entry.err == nil {
245+
t.Fatal("cached entry should contain the error")
246+
}
247+
}
248+
249+
func TestPromqlMetricSelector(t *testing.T) {
250+
got := promqlMetricSelector("node.cpu.usage")
251+
want := `{"__name__"="node.cpu.usage",`
252+
if got != want {
253+
t.Fatalf("promqlMetricSelector(dotted) = %q, want %q", got, want)
254+
}
255+
got = promqlMetricSelector("cpu_usage")
256+
want = `cpu_usage{`
257+
if got != want {
258+
t.Fatalf("promqlMetricSelector(plain) = %q, want %q", got, want)
259+
}
260+
}

0 commit comments

Comments
 (0)