Add cost views and workload rightsizing

internal/opencost/handlers.go

@@ -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
@@ -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]
+	}
 }

internal/prometheus/client.go

@@ -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
@@ -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
@@ -47,6 +50,8 @@ type Client struct {
 	mcpHTTPClient *http.Client
 }
 
+const failedDiscoveryCacheTTL = 5 * time.Second
+
 // Global client instance
 var (
 	globalClient *Client
@@ -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
@@ -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 {
@@ -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()
 	}
 }
@@ -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 = ""
@@ -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

internal/prometheus/client_test.go

@@ -2,6 +2,7 @@ package prometheus
 
 import (
 	"context"
+	"errors"
 	"net/http"
 	"net/http/httptest"
 	"sync/atomic"
@@ -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)
+	}
+}

internal/prometheus/discovery.go

@@ -2,15 +2,19 @@ package prometheus
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"log"
 	"strings"
+	"time"
 
 	"github.com/skyhook-io/radar/internal/errorlog"
 	"github.com/skyhook-io/radar/internal/portforward"
 	"github.com/skyhook-io/radar/pkg/prom"
 )
 
+var ErrPrometheusNotFound = errors.New("no Prometheus service found in cluster")
+
 // discover finds and connects to Prometheus using a multi-layer approach:
 //  1. Manual URL override (--prometheus-url)
 //  2. Existing traffic system port-forward
@@ -69,10 +73,13 @@ func (c *Client) discover(ctx context.Context) (string, string, error) {
 		log.Printf("[prometheus] Discover error: %v", err)
 	}
 	if len(candidates) == 0 {
+		if err != nil {
+			return "", "", fmt.Errorf("Prometheus discovery failed: %w", err)
+		}
 		if !discoveryDiagnosticsSuppressed(ctx) {
 			errorlog.Record("prometheus", "warning", "no Prometheus service found in cluster")
 		}
-		return "", "", fmt.Errorf("no Prometheus service found in cluster")
+		return "", "", ErrPrometheusNotFound
 	}
 
 	log.Printf("[prometheus] Found %d candidate(s), probing...", len(candidates))
@@ -128,7 +135,7 @@ func (c *Client) discover(ctx context.Context) (string, string, error) {
 	if lastErr != nil {
 		return "", "", lastErr
 	}
-	return "", "", fmt.Errorf("no Prometheus service found in cluster")
+	return "", "", ErrPrometheusNotFound
 }
 
 // setDiscoveryServiceFromCandidate records the discovered service metadata
@@ -155,5 +162,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()
 }