Skip to content
33 changes: 29 additions & 4 deletions internal/opencost/handlers.go
Original file line number Diff line number Diff line change
Expand Up @@ -67,13 +67,13 @@ func handleWorkloads(w http.ResponseWriter, r *http.Request) {
}

writeJSON(w, http.StatusOK, pkgopencost.ComputeWorkloadsFromProm(
r.Context(), client.Prom(), ns, buildPodOwnerLookup(ns)))
r.Context(), client.Prom(), ns, BuildPodOwnerLookup(ns)))
}

// buildPodOwnerLookup snapshots radar's pod informer for `ns` so
// BuildPodOwnerLookup snapshots radar's pod informer for `ns` so
// pkg/opencost.ComputeWorkloadsFromProm can resolve pod→workload without
// depending on client-go.
func buildPodOwnerLookup(ns string) pkgopencost.PodOwnerLookup {
func BuildPodOwnerLookup(ns string) pkgopencost.PodOwnerLookup {
rc := k8s.GetResourceCache()
if rc == nil || rc.Pods() == nil {
return nil
Expand Down Expand Up @@ -143,5 +143,30 @@ func handleNodes(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, pkgopencost.NodeCostResponse{Available: false, Reason: pkgopencost.ReasonNoPrometheus})
return
}
writeJSON(w, http.StatusOK, pkgopencost.ComputeNodeCosts(r.Context(), client.Prom()))
resp := pkgopencost.ComputeNodeCosts(r.Context(), client.Prom())
attachNodeProviderIDs(resp)
writeJSON(w, http.StatusOK, resp)
}

func attachNodeProviderIDs(resp *pkgopencost.NodeCostResponse) {
if resp == nil || !resp.Available || len(resp.Nodes) == 0 {
return
}
rc := k8s.GetResourceCache()
if rc == nil || rc.Nodes() == nil {
return
}
nodes, err := rc.Nodes().List(labels.Everything())
if err != nil || len(nodes) == 0 {
return
}
providerIDs := make(map[string]string, len(nodes))
for _, node := range nodes {
if node.Spec.ProviderID != "" {
providerIDs[node.Name] = node.Spec.ProviderID
}
}
for i := range resp.Nodes {
resp.Nodes[i].ProviderID = providerIDs[resp.Nodes[i].Name]
}
}
69 changes: 67 additions & 2 deletions internal/prometheus/client.go
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,8 @@ import (
// with a pkg/prom.Client that performs the actual HTTP calls once an
// endpoint has been discovered.
type Client struct {
mu sync.RWMutex
mu sync.RWMutex
discoverMu sync.Mutex

// Effective connection (populated after discover succeeds).
baseURL string
Expand All @@ -34,6 +35,8 @@ type Client struct {
discoveryService *prom.ServiceInfo // discovered service info for port-forward
manualURL string // --prometheus-url override
headers map[string]string
lastDiscoverErr error
lastDiscoverAt time.Time

// K8s clients for discovery
k8sClient kubernetes.Interface
Expand All @@ -47,6 +50,8 @@ type Client struct {
mcpHTTPClient *http.Client
}

const failedDiscoveryCacheTTL = 5 * time.Second

// Global client instance
var (
globalClient *Client
Expand Down Expand Up @@ -101,6 +106,8 @@ func SetManualURL(rawURL string) {
globalClient.mu.Lock()
defer globalClient.mu.Unlock()
globalClient.manualURL = strings.TrimRight(rawURL, "/")
globalClient.lastDiscoverErr = nil
globalClient.lastDiscoverAt = time.Time{}
}

// SetHeaders sets HTTP headers attached to every Prometheus request on the
Expand All @@ -118,6 +125,8 @@ func SetHeaders(h map[string]string) {
// Drop the cached prom.Client so the next request rebuilds its transport
// with the new headers.
globalClient.prom = nil
globalClient.lastDiscoverErr = nil
globalClient.lastDiscoverAt = time.Time{}
}

func copyHeaders(h map[string]string) map[string]string {
Expand Down Expand Up @@ -147,6 +156,8 @@ func Reset() {
globalClient.prom = nil
globalClient.discovered = false
globalClient.discoveryService = nil
globalClient.lastDiscoverErr = nil
globalClient.lastDiscoverAt = time.Time{}
globalClient.mu.Unlock()
}
}
Expand Down Expand Up @@ -221,6 +232,39 @@ func (c *Client) EnsureConnected(ctx context.Context) (string, string, error) {
if ok {
return base, bp, nil
}
if err := ctx.Err(); err != nil {
return "", "", err
}
log.Printf("[prometheus] cached connection to %s failed probe (reason=%s), rediscovering", base, reason)
c.mu.Lock()
c.baseURL = ""
c.basePath = ""
c.prom = nil
c.discovered = false
c.mu.Unlock()
}
}

c.discoverMu.Lock()
defer c.discoverMu.Unlock()

if err := c.recentDiscoveryError(time.Now()); err != nil {
return "", "", err
}

c.mu.RLock()
base = c.baseURL
bp = c.basePath
c.mu.RUnlock()
if base != "" {
if p := c.getPromClient(); p != nil {
ok, reason := p.Probe(ctx)
if ok {
return base, bp, nil
}
if err := ctx.Err(); err != nil {
return "", "", err
}
log.Printf("[prometheus] cached connection to %s failed probe (reason=%s), rediscovering", base, reason)
c.mu.Lock()
c.baseURL = ""
Expand All @@ -231,7 +275,28 @@ func (c *Client) EnsureConnected(ctx context.Context) (string, string, error) {
}
}

return c.discover(ctx)
base, bp, err := c.discover(ctx)
if err != nil {
if !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
c.mu.Lock()
c.lastDiscoverErr = err
c.lastDiscoverAt = time.Now()
c.mu.Unlock()
}
return "", "", err
}
return base, bp, nil
}

func (c *Client) recentDiscoveryError(now time.Time) error {
c.mu.RLock()
err := c.lastDiscoverErr
at := c.lastDiscoverAt
c.mu.RUnlock()
if err == nil || at.IsZero() || now.Sub(at) >= failedDiscoveryCacheTTL {
return nil
}
return err
}

// Prom returns the underlying pkg/prom.Client for callers that compose
Expand Down
57 changes: 57 additions & 0 deletions internal/prometheus/client_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@ package prometheus

import (
"context"
"errors"
"net/http"
"net/http/httptest"
"sync/atomic"
Expand Down Expand Up @@ -139,3 +140,59 @@ func TestHeadersNoneWhenUnset(t *testing.T) {
t.Error("Authorization header sent when none configured")
}
}

func TestEnsureConnectedReturnsRecentDiscoveryError(t *testing.T) {
wantErr := errors.New("cached discovery failure")
c := &Client{
httpClient: &http.Client{Timeout: 5 * time.Second},
lastDiscoverErr: wantErr,
lastDiscoverAt: time.Now(),
}

_, _, gotErr := c.EnsureConnected(context.Background())
if !errors.Is(gotErr, wantErr) {
t.Fatalf("EnsureConnected error = %v, want cached error %v", gotErr, wantErr)
}
}

func TestEnsureConnectedIgnoresExpiredDiscoveryError(t *testing.T) {
wantErr := errors.New("cached discovery failure")
c := &Client{
httpClient: &http.Client{Timeout: 5 * time.Second},
lastDiscoverErr: wantErr,
lastDiscoverAt: time.Now().Add(-failedDiscoveryCacheTTL - time.Second),
}

_, _, gotErr := c.EnsureConnected(context.Background())
if errors.Is(gotErr, wantErr) {
t.Fatalf("EnsureConnected returned expired cached error: %v", gotErr)
}
if gotErr == nil || gotErr.Error() != "no Kubernetes client available for discovery" {
t.Fatalf("EnsureConnected error = %v, want fresh discovery error", gotErr)
}
}

func TestEnsureConnectedDoesNotClearCachedConnectionOnCanceledProbe(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
<-r.Context().Done()
}))
defer srv.Close()

c := &Client{
baseURL: srv.URL,
httpClient: &http.Client{Timeout: 5 * time.Second},
}
ctx, cancel := context.WithCancel(context.Background())
cancel()

_, _, gotErr := c.EnsureConnected(ctx)
if !errors.Is(gotErr, context.Canceled) {
t.Fatalf("EnsureConnected error = %v, want context.Canceled", gotErr)
}
c.mu.RLock()
gotBase := c.baseURL
c.mu.RUnlock()
if gotBase != srv.URL {
t.Fatalf("baseURL = %q, want cached connection preserved %q", gotBase, srv.URL)
}
}
3 changes: 3 additions & 0 deletions internal/prometheus/discovery.go
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@ import (
"fmt"
"log"
"strings"
"time"

"github.com/skyhook-io/radar/internal/errorlog"
"github.com/skyhook-io/radar/internal/portforward"
Expand Down Expand Up @@ -155,5 +156,7 @@ func (c *Client) markConnected(addr, basePath string) {
c.basePath = basePath
c.prom = nil
c.discovered = true
c.lastDiscoverErr = nil
c.lastDiscoverAt = time.Time{}
c.mu.Unlock()
}
Loading
Loading