Wire the provider health monitor into the manager, with gauges

--provider-check-interval (1m; 0 = off, else at least 10s) and
--provider-check-timeout (15s) are validated before the providers config
loads. Checkers are collected from the unwrapped providers in config
order, so checks never count as provider requests. Three scrape-time
gauges per {provider,target} read the monitor's snapshot: up, check
duration and check timestamp. The e2e suite waits for the kubernetes
ProviderHealth to turn Ready and asserts its gauge in the scrape.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-09-11 20:23:55 +00:00
co-authored by Claude Opus 5
parent 88a15ecdb5
commit 9a3ee963a0
6 changed files with 215 additions and 3 deletions
+58 -2
View File
@@ -16,8 +16,8 @@ limitations under the License.
// Package main is the composition root: it loads the provider config (fail // Package main is the composition root: it loads the provider config (fail
// fast), assembles the provider registry, and wires the reconciler, health // fast), assembles the provider registry, and wires the reconciler, health
// engine, lease store, discovery API, orphan GC, and metrics onto one // engine, lease store, discovery API, orphan GC, provider health checks and
// controller-runtime manager. // metrics onto one controller-runtime manager.
package main package main
import ( import (
@@ -60,6 +60,7 @@ import (
"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider/kubernetes" "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider/kubernetes"
"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider/kvm" "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider/kvm"
"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider/registry" "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider/registry"
"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/providerhealth"
"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/tracing" "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/tracing"
"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/version" "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/version"
// +kubebuilder:scaffold:imports // +kubebuilder:scaffold:imports
@@ -97,6 +98,7 @@ func main() {
var showVersion bool var showVersion bool
var gcpWireFullPayloads bool var gcpWireFullPayloads bool
var traceHealthProbes bool var traceHealthProbes bool
var providerCheckInterval, providerCheckTimeout time.Duration
flag.StringVar(&metricsAddr, "metrics-bind-address", "0", "The address the metrics endpoint binds to. "+ flag.StringVar(&metricsAddr, "metrics-bind-address", "0", "The address the metrics endpoint binds to. "+
"Use :8443 for HTTPS or :8080 for HTTP, or leave as 0 to disable the metrics service.") "Use :8443 for HTTPS or :8080 for HTTP, or leave as 0 to disable the metrics service.")
@@ -140,6 +142,12 @@ func main() {
flag.BoolVar(&traceHealthProbes, "trace-health-probes", false, flag.BoolVar(&traceHealthProbes, "trace-health-probes", false,
"Emit one trace span per health probe. Off by default: probes run about "+ "Emit one trace span per health probe. Off by default: probes run about "+
"once per second per proxy and would dominate trace volume.") "once per second per proxy and would dominate trace volume.")
flag.DurationVar(&providerCheckInterval, "provider-check-interval", time.Minute,
"Interval between provider health checks (can the cluster reach each hypervisor, API "+
"and account), published as ProviderHealth objects and metrics. 0 turns them off; "+
"otherwise at least 10s.")
flag.DurationVar(&providerCheckTimeout, "provider-check-timeout", 15*time.Second,
"Time limit for one round of provider health checks; at most the interval.")
opts := zap.Options{ opts := zap.Options{
Development: true, Development: true,
@@ -174,6 +182,10 @@ func main() {
setupLog.Error(nil, "--providers-config is required") setupLog.Error(nil, "--providers-config is required")
os.Exit(1) os.Exit(1)
} }
if err := providerhealth.ValidateSchedule(providerCheckInterval, providerCheckTimeout); err != nil {
setupLog.Error(err, "Invalid provider check flags")
os.Exit(1)
}
cfg, err := provider.LoadConfigFile(providersConfig) cfg, err := provider.LoadConfigFile(providersConfig)
if err != nil { if err != nil {
setupLog.Error(err, "Failed to load providers config", "path", providersConfig) setupLog.Error(err, "Failed to load providers config", "path", providersConfig)
@@ -195,6 +207,14 @@ func main() {
os.Exit(1) os.Exit(1)
} }
m := metrics.New() m := metrics.New()
// Checkers come from the unwrapped providers, in config order: the
// metrics and tracing wrappers implement only Provider, and a check is
// not a provider request to be counted anyway.
checkTargets := make([]providerhealth.Target, 0, len(cfg.Providers))
for _, pc := range cfg.Providers {
checker, _ := providers[pc.Name].(provider.Checker)
checkTargets = append(checkTargets, providerhealth.Target{Name: pc.Name, Type: pc.Type, Checker: checker})
}
for name, p := range providers { for name, p := range providers {
// Tracing outermost: the span covers the metrics recording too. // Tracing outermost: the span covers the metrics recording too.
providers[name] = provider.WithTracing(name, provider.WithMetrics(name, p, m)) providers[name] = provider.WithTracing(name, provider.WithMetrics(name, p, m))
@@ -314,6 +334,18 @@ func main() {
os.Exit(1) os.Exit(1)
} }
monitor := &providerhealth.Monitor{
Client: mgr.GetClient(),
Reader: mgr.GetAPIReader(),
Targets: checkTargets,
Interval: providerCheckInterval,
Timeout: providerCheckTimeout,
}
if err := mgr.Add(monitor); err != nil {
setupLog.Error(err, "Failed to add provider health monitor")
os.Exit(1)
}
if err := m.Register(ctrlmetrics.Registry, if err := m.Register(ctrlmetrics.Registry,
proxyPhaseCounts(mgr.GetClient()), proxyPhaseCounts(mgr.GetClient()),
func() int { func() int {
@@ -327,6 +359,10 @@ func main() {
setupLog.Error(err, "Failed to register metrics") setupLog.Error(err, "Failed to register metrics")
os.Exit(1) os.Exit(1)
} }
if err := metrics.RegisterProviderTargets(ctrlmetrics.Registry, providerTargetSamples(monitor)); err != nil {
setupLog.Error(err, "Failed to register provider check metrics")
os.Exit(1)
}
if err := (&controller.ProxyReconciler{ if err := (&controller.ProxyReconciler{
Client: mgr.GetClient(), Client: mgr.GetClient(),
@@ -389,3 +425,23 @@ func proxyPhaseCounts(c client.Reader) func() map[string]int {
return counts return counts
} }
} }
// providerTargetSamples flattens the monitor's latest results for the
// scrape-time provider-check gauges.
func providerTargetSamples(mon *providerhealth.Monitor) func() []metrics.ProviderTarget {
return func() []metrics.ProviderTarget {
var out []metrics.ProviderTarget
for _, r := range mon.Snapshot() {
for _, ts := range r.Targets {
out = append(out, metrics.ProviderTarget{
Provider: r.Provider,
Target: ts.Name,
Up: ts.Ready,
Duration: ts.Latency,
CheckedAt: r.CheckedAt,
})
}
}
return out
}
}
+3
View File
@@ -66,6 +66,9 @@ spec:
- --providers-config=/etc/proxy-operator/providers.yaml - --providers-config=/etc/proxy-operator/providers.yaml
# Add --trace-health-probes to emit one span per health probe # Add --trace-health-probes to emit one span per health probe
# (high volume: ~1/s per proxy). # (high volume: ~1/s per proxy).
# Provider health checks (kubectl get providerhealth) run every
# minute; tune with --provider-check-interval (0 turns them off,
# otherwise at least 10s) and --provider-check-timeout (15s).
env: env:
# Bearer token for the discovery API. Optional: without the # Bearer token for the discovery API. Optional: without the
# Secret the API serves unauthenticated (with a loud warning). # Secret the API serves unauthenticated (with a loud warning).
@@ -6,7 +6,7 @@ Plan: [docs/plans/2026-09-11-1526-provider-health-checks.md](../plans/2026-09-11
- [x] Step 2 — Contract plus the kubernetes and GCP checks - [x] Step 2 — Contract plus the kubernetes and GCP checks
- [x] Step 3 — KVM check - [x] Step 3 — KVM check
- [x] Step 4 — Monitor - [x] Step 4 — Monitor
- [ ] Step 5 — Metrics and wiring - [x] Step 5 — Metrics and wiring
- [ ] Step 6 — Docs - [ ] Step 6 — Docs
- [ ] Step 7 — Validate, record, push, open the PR - [ ] Step 7 — Validate, record, push, open the PR
@@ -188,3 +188,50 @@ Worth noting:
- **Test renamed:** the planned `TestMonitor_timeoutBoundsBlockingChecker` is - **Test renamed:** the planned `TestMonitor_timeoutBoundsBlockingChecker` is
`TestMonitor_checkOnce_timeoutBoundsEachRound`. It also asserts that a fast provider `TestMonitor_checkOnce_timeoutBoundsEachRound`. It also asserts that a fast provider
still publishes next to a hung one. still publishes next to a hung one.
## Step 5 — Metrics and wiring
`metrics.RegisterProviderTargets` registers a scrape-time collector for three gauges,
each labelled `{provider,target}`: `up`, check duration and check timestamp.
`Register`'s signature is unchanged.
`cmd/main.go`:
- gains `--provider-check-interval` (1m) and `--provider-check-timeout` (15s);
- validates them before even loading the providers config;
- collects the `Checker`s from the unwrapped providers in config order, just before the
metrics and tracing wrap loop;
- adds the Monitor with the manager's API reader;
- registers the collector through a small `providerTargetSamples` closure next to
`proxyPhaseCounts`.
Build, vet and metrics tests:
```bash
go build ./... && go vet ./cmd/... ./internal/metrics/ && go test -race -count=1 ./internal/metrics/
```
To confirm bad flags fail fast, before anything touches the cluster, I ran the binary
with a 5s interval. zap's development mode prints a stack trace after the error, hence
the grep:
```bash
go run ./cmd --providers-config=/dev/null --provider-check-interval=5s 2>&1 | grep -m1 "Invalid provider check flags"
# ERROR setup Invalid provider check flags {"error": "--provider-check-interval must be 0 (off) or at least 10s, got 5s"}
go vet -tags=e2e ./test/e2e/
```
The e2e suite gets a new ordered spec, "should publish provider health". It waits for
`providerhealth kubernetes` to report `Ready=True`. It runs before the existing metrics
spec, which now also asserts
`proxy_operator_provider_target_up{provider="kubernetes",target="apiserver"} 1`. By then
the first round is known to be done, so the series is certain to exist.
Worth noting:
- **Main-only wiring:** `providerTargetSamples` is an 8-line loop in `package main`,
which has no test file. The e2e metrics assertion is its only coverage.
- **The KVM entry isn't asserted in e2e:** the default providers config also ships
`kvm-home` (beelink), which would report `SSHMaterialUnavailable` in kind. That entry
is the user's own host, and asserting it would break the suite as soon as the default
config changes.
- **Manifest:** `config/manager/manager.yaml` only gained a comment about the two flags.
The defaults need no args.
+48
View File
@@ -117,3 +117,51 @@ func (c *constCollector) Collect(ch chan<- prometheus.Metric) {
ch <- prometheus.MustNewConstMetric(c.desc, prometheus.GaugeValue, float64(value), label) ch <- prometheus.MustNewConstMetric(c.desc, prometheus.GaugeValue, float64(value), label)
} }
} }
// ProviderTarget is one provider health-check target as the gauges see it.
type ProviderTarget struct {
Provider, Target string
Up bool
Duration time.Duration
CheckedAt time.Time
}
// RegisterProviderTargets registers the provider-check gauges. Like the
// other scrape-time collectors they are read from the source at every
// scrape, so a target that goes away stops being reported rather than
// leaving a stale series behind.
func RegisterProviderTargets(reg prometheus.Registerer, read func() []ProviderTarget) error {
labels := []string{"provider", "target"}
return reg.Register(&providerTargetCollector{
up: prometheus.NewDesc("proxy_operator_provider_target_up",
"Whether the target passed its latest provider health check (1) or not (0).", labels, nil),
duration: prometheus.NewDesc("proxy_operator_provider_check_duration_seconds",
"Duration of the target's latest provider health check.", labels, nil),
checked: prometheus.NewDesc("proxy_operator_provider_check_timestamp_seconds",
"Unix time of the target's latest provider health check.", labels, nil),
read: read,
})
}
type providerTargetCollector struct {
up, duration, checked *prometheus.Desc
read func() []ProviderTarget
}
func (c *providerTargetCollector) Describe(ch chan<- *prometheus.Desc) {
ch <- c.up
ch <- c.duration
ch <- c.checked
}
func (c *providerTargetCollector) Collect(ch chan<- prometheus.Metric) {
for _, t := range c.read() {
up := 0.0
if t.Up {
up = 1
}
ch <- prometheus.MustNewConstMetric(c.up, prometheus.GaugeValue, up, t.Provider, t.Target)
ch <- prometheus.MustNewConstMetric(c.duration, prometheus.GaugeValue, t.Duration.Seconds(), t.Provider, t.Target)
ch <- prometheus.MustNewConstMetric(c.checked, prometheus.GaugeValue, float64(t.CheckedAt.UnixNano())/1e9, t.Provider, t.Target)
}
}
+42
View File
@@ -111,3 +111,45 @@ func TestRegister_freshRegistryPerTest(t *testing.T) {
register(t, m1, nil, 0) register(t, m1, nil, 0)
register(t, m2, nil, 0) register(t, m2, nil, 0)
} }
func TestRegisterProviderTargets_scrape(t *testing.T) {
t.Parallel()
checked := time.Unix(1757602800, 0)
targets := []ProviderTarget{
{Provider: "kvm-home", Target: "nas1", Up: true, Duration: 84 * time.Millisecond, CheckedAt: checked},
{Provider: "kvm-home", Target: "nas2", Up: false, Duration: 15 * time.Second, CheckedAt: checked},
}
reg := prometheus.NewRegistry()
if err := RegisterProviderTargets(reg, func() []ProviderTarget { return targets }); err != nil {
t.Fatalf("RegisterProviderTargets: %v", err)
}
expected := `
# HELP proxy_operator_provider_check_duration_seconds Duration of the target's latest provider health check.
# TYPE proxy_operator_provider_check_duration_seconds gauge
proxy_operator_provider_check_duration_seconds{provider="kvm-home",target="nas1"} 0.084
proxy_operator_provider_check_duration_seconds{provider="kvm-home",target="nas2"} 15
# HELP proxy_operator_provider_check_timestamp_seconds Unix time of the target's latest provider health check.
# TYPE proxy_operator_provider_check_timestamp_seconds gauge
proxy_operator_provider_check_timestamp_seconds{provider="kvm-home",target="nas1"} 1.7576028e+09
proxy_operator_provider_check_timestamp_seconds{provider="kvm-home",target="nas2"} 1.7576028e+09
# HELP proxy_operator_provider_target_up Whether the target passed its latest provider health check (1) or not (0).
# TYPE proxy_operator_provider_target_up gauge
proxy_operator_provider_target_up{provider="kvm-home",target="nas1"} 1
proxy_operator_provider_target_up{provider="kvm-home",target="nas2"} 0
`
if err := testutil.GatherAndCompare(reg, strings.NewReader(expected)); err != nil {
t.Error(err)
}
}
func TestRegisterProviderTargets_emptySnapshotHasNoSeries(t *testing.T) {
t.Parallel()
reg := prometheus.NewRegistry()
if err := RegisterProviderTargets(reg, func() []ProviderTarget { return nil }); err != nil {
t.Fatalf("RegisterProviderTargets: %v", err)
}
if n, err := testutil.GatherAndCount(reg); err != nil || n != 0 {
t.Errorf("series before the first check = %d (%v), want 0", n, err)
}
}
+16
View File
@@ -173,6 +173,20 @@ var _ = Describe("Manager", Ordered, func() {
Eventually(verifyControllerUp).Should(Succeed()) Eventually(verifyControllerUp).Should(Succeed())
}) })
// Runs before the metrics spec on purpose: once a ProviderHealth is
// Ready, the first round of checks is done and its gauges exist.
It("should publish provider health", func() {
By("waiting for the kubernetes provider's ProviderHealth to report Ready")
verifyProviderReady := func(g Gomega) {
cmd := exec.Command("kubectl", "get", "providerhealth", "kubernetes",
"-o", `jsonpath={.status.conditions[?(@.type=="Ready")].status}`)
output, err := utils.Run(cmd)
g.Expect(err).NotTo(HaveOccurred())
g.Expect(output).To(Equal("True"), "kubernetes provider not Ready")
}
Eventually(verifyProviderReady).Should(Succeed())
})
It("should ensure the metrics endpoint is serving metrics", func() { It("should ensure the metrics endpoint is serving metrics", func() {
By("creating a ClusterRoleBinding for the service account to allow access to metrics") By("creating a ClusterRoleBinding for the service account to allow access to metrics")
cmd := exec.Command("kubectl", "create", "clusterrolebinding", metricsRoleBindingName, cmd := exec.Command("kubectl", "create", "clusterrolebinding", metricsRoleBindingName,
@@ -264,6 +278,8 @@ var _ = Describe("Manager", Ordered, func() {
g.Expect(err).NotTo(HaveOccurred(), "Failed to retrieve logs from curl pod") g.Expect(err).NotTo(HaveOccurred(), "Failed to retrieve logs from curl pod")
g.Expect(metricsOutput).NotTo(BeEmpty()) g.Expect(metricsOutput).NotTo(BeEmpty())
g.Expect(metricsOutput).To(ContainSubstring("< HTTP/1.1 200 OK")) g.Expect(metricsOutput).To(ContainSubstring("< HTTP/1.1 200 OK"))
g.Expect(metricsOutput).To(ContainSubstring(
`proxy_operator_provider_target_up{provider="kubernetes",target="apiserver"} 1`))
} }
Eventually(verifyMetricsAvailable, 2*time.Minute).Should(Succeed()) Eventually(verifyMetricsAvailable, 2*time.Minute).Should(Succeed())
}) })