Skip to content

Commit beffe1f

Browse files
nadaverellnadaverell
authored
Handle unavailable metrics API without drawer errors (#1064)
1 parent ff5411f commit beffe1f

21 files changed

Lines changed: 1593 additions & 143 deletions

‎internal/k8s/cache.go‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -901,6 +901,8 @@ func (c *ResourceCache) getDynamicWithGroup(ctx context.Context, kind string, na
901901
u, err = dynamicCache.GetDirect(ctx, gvr, namespace, name)
902902
} else if gvr.Group == "apiextensions.k8s.io" && gvr.Resource == "customresourcedefinitions" {
903903
u, err = dynamicCache.GetDirect(ctx, gvr, namespace, name)
904+
} else if gvr.Group == "apiregistration.k8s.io" && gvr.Resource == "apiservices" {
905+
u, err = dynamicCache.GetDirect(ctx, gvr, namespace, name)
904906
} else if preserveLastApplied {
905907
u, err = dynamicCache.GetDirectPreserveLastApplied(ctx, gvr, namespace, name)
906908
} else {

‎internal/k8s/cache_dynamic_test.go‎

Lines changed: 49 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,49 @@
1+
package k8s
2+
3+
import (
4+
"context"
5+
"testing"
6+
7+
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
8+
"k8s.io/apimachinery/pkg/runtime"
9+
"k8s.io/apimachinery/pkg/runtime/schema"
10+
dynamicfake "k8s.io/client-go/dynamic/fake"
11+
)
12+
13+
func TestGetDynamicWithGroupDirectFetchesAPIService(t *testing.T) {
14+
defer ResetTestDynamicState()
15+
16+
gvr := schema.GroupVersionResource{Group: "apiregistration.k8s.io", Version: "v1", Resource: "apiservices"}
17+
apiService := &unstructured.Unstructured{Object: map[string]any{
18+
"apiVersion": "apiregistration.k8s.io/v1",
19+
"kind": "APIService",
20+
"metadata": map[string]any{
21+
"name": "v1beta1.metrics.k8s.io",
22+
},
23+
}}
24+
dyn := dynamicfake.NewSimpleDynamicClientWithCustomListKinds(
25+
runtime.NewScheme(),
26+
map[schema.GroupVersionResource]string{gvr: "APIServiceList"},
27+
apiService,
28+
)
29+
if err := InitTestDynamicResourceCache(dyn, []APIResource{{
30+
Group: "apiregistration.k8s.io",
31+
Version: "v1",
32+
Kind: "APIService",
33+
Name: "apiservices",
34+
Namespaced: false,
35+
}}); err != nil {
36+
t.Fatalf("InitTestDynamicResourceCache: %v", err)
37+
}
38+
39+
got, err := (&ResourceCache{}).GetDynamicWithGroup(context.Background(), "APIService", "", "v1beta1.metrics.k8s.io", "apiregistration.k8s.io")
40+
if err != nil {
41+
t.Fatalf("GetDynamicWithGroup: %v", err)
42+
}
43+
if got.GetName() != "v1beta1.metrics.k8s.io" {
44+
t.Fatalf("GetDynamicWithGroup name = %q", got.GetName())
45+
}
46+
if count := GetDynamicResourceCache().GetInformerCount(); count != 0 {
47+
t.Fatalf("GetDynamicWithGroup(APIService) started %d dynamic informer(s), want direct GET", count)
48+
}
49+
}

‎internal/prometheus/client.go‎

Lines changed: 21 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -53,6 +53,17 @@ var (
5353
clientMu sync.RWMutex
5454
)
5555

56+
type suppressDiscoveryDiagnosticsKey struct{}
57+
58+
func withSuppressedDiscoveryDiagnostics(ctx context.Context) context.Context {
59+
return context.WithValue(ctx, suppressDiscoveryDiagnosticsKey{}, true)
60+
}
61+
62+
func discoveryDiagnosticsSuppressed(ctx context.Context) bool {
63+
suppressed, _ := ctx.Value(suppressDiscoveryDiagnosticsKey{}).(bool)
64+
return suppressed
65+
}
66+
5667
// Initialize creates the global Prometheus client.
5768
func Initialize(client kubernetes.Interface, config *rest.Config, contextName string) {
5869
clientMu.Lock()
@@ -296,7 +307,7 @@ func (c *Client) probe(ctx context.Context, addr string) bool {
296307
tr.Headers = headers
297308
ok, reason := prom.NewClient(tr).Probe(ctx)
298309
if !ok {
299-
logProbeRejection(addr, reason)
310+
logProbeRejection(addr, reason, !discoveryDiagnosticsSuppressed(ctx))
300311
}
301312
return ok
302313
}
@@ -306,14 +317,18 @@ func (c *Client) probe(ctx context.Context, addr string) bool {
306317
// misconfiguration); empty instances get warning level (cluster state);
307318
// other failures use stdlib log so they appear in the discovery audit
308319
// trail without flooding errorlog.
309-
func logProbeRejection(addr string, reason prom.ProbeReason) {
320+
func logProbeRejection(addr string, reason prom.ProbeReason, recordDiagnostics bool) {
310321
switch reason {
311322
case prom.ProbeReasonAuthError:
312-
errorlog.Record("prometheus", "error",
313-
"endpoint %s rejected credentials (HTTP 401/403, check --prometheus-header)", addr)
323+
if recordDiagnostics {
324+
errorlog.Record("prometheus", "error",
325+
"endpoint %s rejected credentials (HTTP 401/403, check --prometheus-header)", addr)
326+
}
314327
case prom.ProbeReasonEmptyInstance:
315-
errorlog.Record("prometheus", "warning",
316-
"endpoint %s has no active scrape targets (empty instance), skipping", addr)
328+
if recordDiagnostics {
329+
errorlog.Record("prometheus", "warning",
330+
"endpoint %s has no active scrape targets (empty instance), skipping", addr)
331+
}
317332
case prom.ProbeReasonNotPrometheus:
318333
log.Printf("[prometheus] endpoint %s responded but not in Prometheus format, skipping", addr)
319334
case prom.ProbeReasonPromError:

‎internal/prometheus/discovery.go‎

Lines changed: 12 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -38,7 +38,9 @@ func (c *Client) discover(ctx context.Context) (string, string, error) {
3838
c.markConnected(addr, "")
3939
return addr, "", nil
4040
}
41-
errorlog.Record("prometheus", "error", "manual Prometheus URL %s not reachable", addr)
41+
if !discoveryDiagnosticsSuppressed(ctx) {
42+
errorlog.Record("prometheus", "error", "manual Prometheus URL %s not reachable", addr)
43+
}
4244
return "", "", fmt.Errorf("manual Prometheus URL %s not reachable", addr)
4345
}
4446

@@ -67,7 +69,9 @@ func (c *Client) discover(ctx context.Context) (string, string, error) {
6769
log.Printf("[prometheus] Discover error: %v", err)
6870
}
6971
if len(candidates) == 0 {
70-
errorlog.Record("prometheus", "warning", "no Prometheus service found in cluster")
72+
if !discoveryDiagnosticsSuppressed(ctx) {
73+
errorlog.Record("prometheus", "warning", "no Prometheus service found in cluster")
74+
}
7175
return "", "", fmt.Errorf("no Prometheus service found in cluster")
7276
}
7377

@@ -99,7 +103,9 @@ func (c *Client) discover(ctx context.Context) (string, string, error) {
99103
connInfo, pfErr := portforward.Start(ctx, cand.Namespace, cand.Name, cand.TargetPort, contextName)
100104
if pfErr != nil {
101105
lastErr = fmt.Errorf("port-forward to %s/%s failed: %w", cand.Namespace, cand.Name, pfErr)
102-
errorlog.Record("prometheus", "error", "port-forward to %s/%s failed: %v", cand.Namespace, cand.Name, pfErr)
106+
if !discoveryDiagnosticsSuppressed(ctx) {
107+
errorlog.Record("prometheus", "error", "port-forward to %s/%s failed: %v", cand.Namespace, cand.Name, pfErr)
108+
}
103109
continue
104110
}
105111

@@ -111,7 +117,9 @@ func (c *Client) discover(ctx context.Context) (string, string, error) {
111117

112118
portforward.Stop()
113119
lastErr = fmt.Errorf("Prometheus at %s/%s not responding after port-forward", cand.Namespace, cand.Name)
114-
errorlog.Record("prometheus", "error", "Prometheus at %s/%s not responding after port-forward", cand.Namespace, cand.Name)
120+
if !discoveryDiagnosticsSuppressed(ctx) {
121+
errorlog.Record("prometheus", "error", "Prometheus at %s/%s not responding after port-forward", cand.Namespace, cand.Name)
122+
}
115123
}
116124

117125
c.mu.Lock()

‎internal/prometheus/handlers.go‎

Lines changed: 12 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -64,9 +64,20 @@ func handleConnect(w http.ResponseWriter, r *http.Request) {
6464
return
6565
}
6666

67-
_, _, err := client.EnsureConnected(r.Context())
67+
ctx := r.Context()
68+
optional := r.URL.Query().Get("optional") == "true"
69+
if optional {
70+
ctx = withSuppressedDiscoveryDiagnostics(ctx)
71+
}
72+
_, _, err := client.EnsureConnected(ctx)
6873
if err != nil {
6974
log.Printf("[prometheus] Connection failed: %v", err)
75+
if optional {
76+
status := client.GetStatus()
77+
status.Error = "Prometheus connection failed: " + err.Error()
78+
writeJSON(w, http.StatusOK, status)
79+
return
80+
}
7081
errorlog.Record("prometheus", "error", "connection failed: %v", err)
7182
writeError(w, http.StatusBadGateway, "Prometheus connection failed: "+err.Error())
7283
return

‎internal/prometheus/handlers_test.go‎

Lines changed: 53 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,12 +1,65 @@
11
package prometheus
22

33
import (
4+
"encoding/json"
5+
"net/http"
6+
"net/http/httptest"
47
"testing"
8+
"time"
59

10+
"github.com/skyhook-io/radar/internal/errorlog"
11+
"github.com/skyhook-io/radar/pkg/prom"
612
corev1 "k8s.io/api/core/v1"
713
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
814
)
915

16+
func TestHandleConnectOptionalReturnsUnavailableStatus(t *testing.T) {
17+
errorlog.Reset()
18+
t.Cleanup(errorlog.Reset)
19+
20+
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
21+
http.Error(w, "not prometheus", http.StatusInternalServerError)
22+
}))
23+
defer srv.Close()
24+
25+
clientMu.Lock()
26+
previous := globalClient
27+
globalClient = &Client{
28+
manualURL: srv.URL,
29+
contextName: "test-context",
30+
httpClient: &http.Client{Timeout: 5 * time.Second},
31+
}
32+
clientMu.Unlock()
33+
t.Cleanup(func() {
34+
clientMu.Lock()
35+
globalClient = previous
36+
clientMu.Unlock()
37+
})
38+
39+
req := httptest.NewRequest(http.MethodPost, "/prometheus/connect?optional=true", nil)
40+
rec := httptest.NewRecorder()
41+
handleConnect(rec, req)
42+
43+
if rec.Code != http.StatusOK {
44+
t.Fatalf("status code = %d, want %d; body=%s", rec.Code, http.StatusOK, rec.Body.String())
45+
}
46+
var status prom.Status
47+
if err := json.NewDecoder(rec.Body).Decode(&status); err != nil {
48+
t.Fatalf("decode status: %v", err)
49+
}
50+
if status.Connected {
51+
t.Fatal("Connected = true, want false")
52+
}
53+
if status.Error == "" {
54+
t.Fatal("Error is empty, want connection failure detail")
55+
}
56+
for _, entry := range errorlog.GetEntries() {
57+
if entry.Source == "prometheus" && entry.Level == "error" {
58+
t.Fatalf("optional connect recorded error log entry: %+v", entry)
59+
}
60+
}
61+
}
62+
1063
func TestAnyNodeUsesDocker(t *testing.T) {
1164
tests := []struct {
1265
name string

0 commit comments

Comments
 (0)