Skip to content

fix(opencost): make agent OpenCost connection detection reliable #489

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
RamanKharchee wants to merge 2 commits into
base: main
Choose a base branch
from
Open

fix(opencost): make agent OpenCost connection detection reliable #489

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
afe9d0a
fix(opencost): make agent OpenCost connection detection reliable
RamanKharchee Jun 19, 2026
e9eb941
chore: update image tags for main release
github-actions[bot] Jun 19, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions charts/nudgebee-agent/templates/runner.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -94,6 +94,11 @@ stringData:
# disabled the runner must NOT autodiscover a neighbouring namespace's OpenCost,
# so the server side detects cost is off and takes over centrally.
OPENCOST_ENABLED: {{ .Values.opencost.enabled | quote }}
{{- if .Values.opencost.endpoint }}
# Explicit OpenCost cost-model API endpoint (serves /healthz). Overrides
# in-cluster autodiscovery; set when OpenCost runs under non-standard labels.
OPENCOST_ENDPOINT: {{ .Values.opencost.endpoint | quote }}
{{- end }}
{{- if .Values.globalConfig.prometheus_url }}
PROMETHEUS_URL: {{ .Values.globalConfig.prometheus_url | quote }}
{{- end }}
Expand Down
7 changes: 6 additions & 1 deletion charts/nudgebee-agent/values.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,7 +61,7 @@ runnerServiceAccount:
runner:
image:
repository: ghcr.io/nudgebee/nudgebee-agent
tag: 2026-06-18T08-03-27_f048f62d0348572793afa98b8a9b1cc383115071
tag: 2026-06-19T07-29-09_0c10bd2144db2a151cd74f2b5997347cebd67745
# Image template the pod_profiler action launches debugger pods from.
# The agent substitutes `{}` for the variant (bpf, jvm, python, perf, ruby).
# Surfaces as PROFILER_IMAGE; leave empty to fall back to the binary default.
Expand Down Expand Up @@ -220,6 +220,11 @@ nodeAgent:
# `--set opencost.enabled=true`.
opencost:
enabled: false
# Optional explicit OpenCost cost-model API endpoint (the one that serves
# /healthz, typically port 9003). Leave empty to autodiscover the in-cluster
# OpenCost Service by label; set it when OpenCost runs under non-standard
# labels that autodiscovery won't match. Maps to OPENCOST_ENDPOINT.
endpoint: ""
nameOverride: ""
fullnameOverride: ""
opencost:
Expand Down
31 changes: 19 additions & 12 deletions runner/cmd/agent/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -235,18 +235,13 @@ func run(ctx context.Context, logger *slog.Logger, cfg *config.Config) error {
logger.Warn("invalid OPENCOST_ENABLED, defaulting to enabled", "value", v, "err", err)
}
}
opencostURL := ""
if opencostEnabled {
// Mirrors OpenCostDiscovery.find_open_cost_url.
// OPENCOST_ENDPOINT env wins; falls back to in-cluster Service lookup.
opencostURL = os.Getenv("OPENCOST_ENDPOINT")
if opencostURL == "" {
if u := disc.FindFirst(ctx, svcdiscover.OpencostSelectors); u != "" {
opencostURL = u
logger.Info("opencost auto-discovered", "url", u)
}
}
} else {
// OPENCOST_ENDPOINT (if set) wins over autodiscovery; read once since env
// doesn't change. The URL itself is resolved per telemetry tick inside the
// Datasources closure below (not here) so the agent self-heals when OpenCost
// is deployed after the agent boots — resolving only at startup would latch
// the boot-time result (empty → opencostConnection stuck false) forever.
opencostEndpoint := os.Getenv("OPENCOST_ENDPOINT")
if !opencostEnabled {
logger.Info("opencost disabled (OPENCOST_ENABLED=false); skipping discovery and cost polling")
}

Expand Down Expand Up @@ -923,6 +918,18 @@ func run(ctx context.Context, logger *slog.Logger, cfg *config.Config) error {
logsProvider, logsURL, logsOK, logCfg := probeLogsProvider(probeCtx, cfg)
as := telemetry.DetectAutoScaler(probeCtx, typedKube, providerInfo.Provider, logger)
clickhouseStatus := probeClickhouse(probeCtx, probeClient, clickhouseHost, clickhousePort)
// Resolve OpenCost per tick (cached ~CacheTTL by the Discoverer) so a
// late-deployed OpenCost is picked up without an agent restart. Mirrors
// OpenCostDiscovery.find_open_cost_url: OPENCOST_ENDPOINT wins, else
// autodiscover — preferring the cost-model API port (9003, serves
// /healthz) over a UI port. Empty when OpenCost is disabled or absent.
opencostURL := ""
if opencostEnabled {
opencostURL = opencostEndpoint
if opencostURL == "" {
opencostURL = disc.FindFirstPreferPort(probeCtx, svcdiscover.OpencostSelectors, 9003)
}
}
return telemetry.Datasources{
PrometheusURL: cfg.PrometheusURL,
AlertManagerURL: cfg.AlertManagerURL,
Expand Down
35 changes: 32 additions & 3 deletions runner/pkg/svcdiscover/discover.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ import (
"sync"
"time"

corev1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/client-go/kubernetes"
)
Expand Down Expand Up @@ -106,10 +107,24 @@ func New(cs kubernetes.Interface, clusterDomain string) *Discoverer {
// or "" if none. Negative results are cached too so we don't keep listing
// services on every call.
func (d *Discoverer) FindFirst(ctx context.Context, selectors []string) string {
return d.FindFirstPreferPort(ctx, selectors)
}

// FindFirstPreferPort is FindFirst with port preference: when a matched Service
// exposes multiple ports, the first port whose number appears in preferredPorts
// (in order) is used, falling back to the Service's first port. This matters for
// OpenCost, whose Service can expose both the cost-model API (9003, which serves
// /healthz) and a UI port (9090) — probing /healthz on the UI port reports an
// otherwise-healthy OpenCost as down. Results (including misses) are cached per
// (selectors, preferredPorts) for CacheTTL.
func (d *Discoverer) FindFirstPreferPort(ctx context.Context, selectors []string, preferredPorts ...int32) string {
if d == nil || d.cs == nil {
return ""
}
cacheKey := strings.Join(selectors, "|")
for _, p := range preferredPorts {
cacheKey += fmt.Sprintf("#%d", p)
}
now := time.Now()

d.mu.Lock()
Expand All @@ -121,7 +136,7 @@ func (d *Discoverer) FindFirst(ctx context.Context, selectors []string) string {

url := ""
for _, sel := range selectors {
if u := d.findOne(ctx, sel); u != "" {
if u := d.findOne(ctx, sel, preferredPorts); u != "" {
url = u
break
}
Expand All @@ -133,7 +148,7 @@ func (d *Discoverer) FindFirst(ctx context.Context, selectors []string) string {
return url
}

func (d *Discoverer) findOne(ctx context.Context, selector string) string {
func (d *Discoverer) findOne(ctx context.Context, selector string, preferredPorts []int32) string {
list, err := d.cs.CoreV1().Services("").List(ctx, metav1.ListOptions{LabelSelector: selector})
if err != nil || len(list.Items) == 0 {
return ""
Comment on lines +151 to 154

@gemini-code-assist gemini-code-assist Bot Jun 19, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

Transient API Error Caching Bug

In the current implementation, if d.cs.CoreV1().Services("").List fails due to a transient API error (such as a temporary network glitch, API server rate limiting, or context timeout/cancellation), findOne returns "" (an empty string).

Because findOne does not distinguish between a transient API error and a service actually not being found (both return ""), FindFirstPreferPort will cache this negative result ("") for the full CacheTTL (1 hour).

This means that if a single transient error occurs during a telemetry tick, the agent will report opencostConnection=false (and potentially other auto-discovered connections as down) for a whole hour, even if OpenCost is perfectly healthy and running. This can lead to false-negative alerts or server-side issues like double-writes.

Suggested Fix

Refactor findOne to return (string, error) so that FindFirstPreferPort can distinguish between a successful "not found" result and an API error. Only cache the result in d.cache if no API errors were encountered during the discovery loop:

func (d *Discoverer) FindFirstPreferPort(ctx context.Context, selectors []string, preferredPorts ...int32) string {
    // ...
    url := ""
    var apiErr error
    for _, sel := range selectors {
        u, err := d.findOne(ctx, sel, preferredPorts)
        if err != nil {
            apiErr = err
        }
        if u != "" {
            url = u
            break
        }
    }

    d.mu.Lock()
    if apiErr == nil || url != "" {
        d.cache[cacheKey] = cacheEntry{url: url, expires: now.Add(CacheTTL)}
    }
    d.mu.Unlock()
    return url
}

func (d *Discoverer) findOne(ctx context.Context, selector string, preferredPorts []int32) (string, error) {
    list, err := d.cs.CoreV1().Services("").List(ctx, metav1.ListOptions{LabelSelector: selector})
    if err != nil {
        return "", err
    }
    if len(list.Items) == 0 {
        return "", nil
    }
    // ...
}

Expand All @@ -142,10 +157,24 @@ func (d *Discoverer) findOne(ctx context.Context, selector string) string {
if len(svc.Spec.Ports) == 0 {
return ""
}
port := svc.Spec.Ports[0].Port
port := selectPort(svc.Spec.Ports, preferredPorts)
return fmt.Sprintf("http://%s.%s.svc.%s:%d", svc.Name, svc.Namespace, d.clusterDomain, port)
}

// selectPort returns the first port whose number is in preferredPorts (in
// preference order); if none match — or preferredPorts is empty — it returns the
// Service's first port, preserving the original single-port behaviour.
func selectPort(ports []corev1.ServicePort, preferredPorts []int32) int32 {
for _, want := range preferredPorts {
for i := range ports {
if ports[i].Port == want {
return want
}
}
}
return ports[0].Port
}

// Coalesce returns the first non-empty value. Used at startup when wiring
// configured envs against autodiscovered URLs:
//
Expand Down
75 changes: 75 additions & 0 deletions runner/pkg/svcdiscover/discover_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -101,6 +101,70 @@ func TestNilDiscovererReturnsEmpty(t *testing.T) {
}
}

// TestFindFirstPreferPort_PicksAPIPortOverUI seeds an OpenCost Service whose
// first port is the UI (9090) and second is the cost-model API (9003, /healthz).
// Without a preference the first port wins and /healthz is probed on the UI port
// — reporting a healthy OpenCost as down. With prefer 9003 the API port wins.
func TestFindFirstPreferPort_PicksAPIPortOverUI(t *testing.T) {
cs := fake.NewClientset(
mkServicePorts("opencost", "opencost", map[string]string{"app": "opencost"}, 9090, 9003),
)
d := New(cs, "cluster.local")

// Default behaviour: first port (UI) — the bug.
if got, want := d.FindFirst(context.Background(), OpencostSelectors),
"http://opencost.opencost.svc.cluster.local:9090"; got != want {
t.Errorf("FindFirst got %q; want %q", got, want)
}
// With preference the cost-model API port wins.
if got, want := d.FindFirstPreferPort(context.Background(), OpencostSelectors, 9003),
"http://opencost.opencost.svc.cluster.local:9003"; got != want {
t.Errorf("FindFirstPreferPort got %q; want %q", got, want)
}
}

// TestFindFirstPreferPort_FallsBackToFirstPort verifies that when none of the
// preferred ports are exposed, selection falls back to the Service's first port.
func TestFindFirstPreferPort_FallsBackToFirstPort(t *testing.T) {
cs := fake.NewClientset(
mkServicePorts("opencost", "opencost", map[string]string{"app": "opencost"}, 9003),
)
d := New(cs, "cluster.local")
if got, want := d.FindFirstPreferPort(context.Background(), OpencostSelectors, 12345),
"http://opencost.opencost.svc.cluster.local:9003"; got != want {
t.Errorf("got %q; want %q", got, want)
}
}

func TestSelectPort(t *testing.T) {
ports := func(ns ...int32) []corev1.ServicePort {
out := make([]corev1.ServicePort, len(ns))
for i, n := range ns {
out[i] = corev1.ServicePort{Port: n}
}
return out
}
cases := []struct {
name string
ports []corev1.ServicePort
preferred []int32
want int32
}{
{"empty preference uses first", ports(9090, 9003), nil, 9090},
{"preferred present", ports(9090, 9003), []int32{9003}, 9003},
{"preference order honoured", ports(9090, 9003), []int32{9003, 9090}, 9003},
{"preferred absent falls back", ports(9090, 9003), []int32{8080}, 9090},
{"second preference matches", ports(9090, 9003), []int32{8080, 9090}, 9090},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
if got := selectPort(tc.ports, tc.preferred); got != tc.want {
t.Errorf("selectPort = %d; want %d", got, tc.want)
}
})
}
}

func mkService(name, namespace string, port int32, labels map[string]string) *corev1.Service {
return &corev1.Service{
ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: namespace, Labels: labels},
Expand All @@ -109,3 +173,14 @@ func mkService(name, namespace string, port int32, labels map[string]string) *co
},
}
}

func mkServicePorts(name, namespace string, labels map[string]string, ports ...int32) *corev1.Service {
sp := make([]corev1.ServicePort, len(ports))
for i, p := range ports {
sp[i] = corev1.ServicePort{Port: p}
}
return &corev1.Service{
ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: namespace, Labels: labels},
Spec: corev1.ServiceSpec{Ports: sp},
}
}