OpenTelemetry tracing integrated with logr/zap logging #4
@@ -94,6 +94,7 @@ func main() {
|
|||||||
var leaseCooldown, maxLeaseTTL time.Duration
|
var leaseCooldown, maxLeaseTTL time.Duration
|
||||||
var showVersion bool
|
var showVersion bool
|
||||||
var gcpWireFullPayloads bool
|
var gcpWireFullPayloads bool
|
||||||
|
var traceHealthProbes bool
|
||||||
|
|
||||||
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.")
|
||||||
@@ -134,6 +135,9 @@ func main() {
|
|||||||
"Print the commit the binary was built from and exit.")
|
"Print the commit the binary was built from and exit.")
|
||||||
flag.BoolVar(&gcpWireFullPayloads, "gcp-wire-log-full-payloads", false,
|
flag.BoolVar(&gcpWireFullPayloads, "gcp-wire-log-full-payloads", false,
|
||||||
"Log GCP V(5) wire payloads verbatim instead of eliding fields larger than 1KiB.")
|
"Log GCP V(5) wire payloads verbatim instead of eliding fields larger than 1KiB.")
|
||||||
|
flag.BoolVar(&traceHealthProbes, "trace-health-probes", false,
|
||||||
|
"Emit one trace span per health probe. Off by default: probes run about "+
|
||||||
|
"once per second per proxy and would dominate trace volume.")
|
||||||
|
|
||||||
opts := zap.Options{
|
opts := zap.Options{
|
||||||
Development: true,
|
Development: true,
|
||||||
@@ -277,6 +281,7 @@ func main() {
|
|||||||
engine := health.NewEngine(mgr.GetClient())
|
engine := health.NewEngine(mgr.GetClient())
|
||||||
engine.Workers = healthWorkers
|
engine.Workers = healthWorkers
|
||||||
engine.Metrics = m
|
engine.Metrics = m
|
||||||
|
engine.TraceProbes = traceHealthProbes
|
||||||
if err := mgr.Add(engine); err != nil {
|
if err := mgr.Add(engine); err != nil {
|
||||||
setupLog.Error(err, "Failed to add health engine")
|
setupLog.Error(err, "Failed to add health engine")
|
||||||
os.Exit(1)
|
os.Exit(1)
|
||||||
|
|||||||
@@ -8,7 +8,7 @@ Plan: `docs/plans/2026-08-24-1025-otel-tracing.md`
|
|||||||
- [x] Step 4 — `cmd/main.go` wiring
|
- [x] Step 4 — `cmd/main.go` wiring
|
||||||
- [x] Step 5 — Reconciler spans
|
- [x] Step 5 — Reconciler spans
|
||||||
- [x] Step 6 — Discovery server
|
- [x] Step 6 — Discovery server
|
||||||
- [ ] Step 7 — GC + health
|
- [x] Step 7 — GC + health
|
||||||
- [ ] Step 8 — GCP wire-log enrichment
|
- [ ] Step 8 — GCP wire-log enrichment
|
||||||
- [ ] Step 9 — Manifests + docs
|
- [ ] Step 9 — Manifests + docs
|
||||||
|
|
||||||
@@ -133,3 +133,16 @@ Deviation from the plan's letter: `recoverMiddleware` keeps `s.log`. It sits
|
|||||||
*outside* the tracing layer, so its request ctx never has the enriched
|
*outside* the tracing layer, so its request ctx never has the enriched
|
||||||
logger — switching it to `FromContext` would silently drop the `"discovery"`
|
logger — switching it to `FromContext` would silently drop the `"discovery"`
|
||||||
name and gain nothing.
|
name and gain nothing.
|
||||||
|
|
||||||
|
## Step 7 — GC + health
|
||||||
|
|
||||||
|
`gc.sweep` runs in a root span with `gc.providers` / `gc.deleted` counters
|
||||||
|
(a failed orphan delete no longer counts as deleted — the loop grew an
|
||||||
|
explicit `continue`). The health engine got the deferred `--trace-health-probes`
|
||||||
|
flag and `TraceProbes` field; the probe call moved into `runProbe`, which
|
||||||
|
wraps it in a `health.probe` client span only when enabled, ending before
|
||||||
|
`record` (which stays ctx-free by design). Probe transports remain
|
||||||
|
uninstrumented so no traceparent can leak through a proxy to external
|
||||||
|
targets. Probe→reconcile span links stay out of scope (GenericEvent carries
|
||||||
|
no ctx; the workqueue coalesces events) — recorded in the architecture
|
||||||
|
Decisions in Step 9.
|
||||||
|
|||||||
@@ -12,8 +12,11 @@ import (
|
|||||||
"sigs.k8s.io/controller-runtime/pkg/client"
|
"sigs.k8s.io/controller-runtime/pkg/client"
|
||||||
logf "sigs.k8s.io/controller-runtime/pkg/log"
|
logf "sigs.k8s.io/controller-runtime/pkg/log"
|
||||||
|
|
||||||
|
"go.opentelemetry.io/otel/attribute"
|
||||||
|
|
||||||
crawlv1alpha1 "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/api/v1alpha1"
|
crawlv1alpha1 "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/api/v1alpha1"
|
||||||
"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider"
|
"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/provider"
|
||||||
|
"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/tracing"
|
||||||
)
|
)
|
||||||
|
|
||||||
// Sweeper is the manager Runnable running the sweep loop.
|
// Sweeper is the manager Runnable running the sweep loop.
|
||||||
@@ -81,6 +84,8 @@ func (s *Sweeper) Start(ctx context.Context) error {
|
|||||||
// that deletion, and GC racing it would double-delete. A UID becomes
|
// that deletion, and GC racing it would double-delete. A UID becomes
|
||||||
// orphan-eligible only once the object is fully gone.
|
// orphan-eligible only once the object is fully gone.
|
||||||
func (s *Sweeper) sweep(ctx context.Context) {
|
func (s *Sweeper) sweep(ctx context.Context) {
|
||||||
|
ctx, span := tracing.Start(ctx, "gc.sweep")
|
||||||
|
defer span.End()
|
||||||
log := logf.FromContext(ctx).WithName("orphan-gc")
|
log := logf.FromContext(ctx).WithName("orphan-gc")
|
||||||
|
|
||||||
var list crawlv1alpha1.ProxyList
|
var list crawlv1alpha1.ProxyList
|
||||||
@@ -95,6 +100,7 @@ func (s *Sweeper) sweep(ctx context.Context) {
|
|||||||
live[string(list.Items[i].UID)] = true
|
live[string(list.Items[i].UID)] = true
|
||||||
}
|
}
|
||||||
|
|
||||||
|
deleted := 0
|
||||||
for name, prov := range s.Providers {
|
for name, prov := range s.Providers {
|
||||||
instances, err := prov.ListByTag(ctx)
|
instances, err := prov.ListByTag(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -119,7 +125,13 @@ func (s *Sweeper) sweep(ctx context.Context) {
|
|||||||
if err := prov.Delete(ctx, inst.ID); err != nil {
|
if err := prov.Delete(ctx, inst.ID); err != nil {
|
||||||
log.Error(err, "deleting orphaned instance",
|
log.Error(err, "deleting orphaned instance",
|
||||||
"provider", name, "providerID", inst.ID, "uid", inst.UID)
|
"provider", name, "providerID", inst.ID, "uid", inst.UID)
|
||||||
|
continue
|
||||||
}
|
}
|
||||||
|
deleted++
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
span.SetAttributes(
|
||||||
|
attribute.Int("gc.providers", len(s.Providers)),
|
||||||
|
attribute.Int("gc.deleted", deleted),
|
||||||
|
)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -10,6 +10,9 @@ import (
|
|||||||
"sync"
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
"go.opentelemetry.io/otel/attribute"
|
||||||
|
"go.opentelemetry.io/otel/codes"
|
||||||
|
"go.opentelemetry.io/otel/trace"
|
||||||
apimeta "k8s.io/apimachinery/pkg/api/meta"
|
apimeta "k8s.io/apimachinery/pkg/api/meta"
|
||||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||||
"k8s.io/apimachinery/pkg/types"
|
"k8s.io/apimachinery/pkg/types"
|
||||||
@@ -18,6 +21,7 @@ import (
|
|||||||
logf "sigs.k8s.io/controller-runtime/pkg/log"
|
logf "sigs.k8s.io/controller-runtime/pkg/log"
|
||||||
|
|
||||||
crawlv1alpha1 "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/api/v1alpha1"
|
crawlv1alpha1 "gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/api/v1alpha1"
|
||||||
|
"gitea.home.hrajfrisbee.cz/kacerr/egress-proxies-operator/internal/tracing"
|
||||||
)
|
)
|
||||||
|
|
||||||
// Snapshot is the engine's current verdict for one proxy, read by the
|
// Snapshot is the engine's current verdict for one proxy, read by the
|
||||||
@@ -99,6 +103,10 @@ type Engine struct {
|
|||||||
// transition-only by design, so metrics are where high-frequency
|
// transition-only by design, so metrics are where high-frequency
|
||||||
// signal (true probe recency, every latency sample) lives.
|
// signal (true probe recency, every latency sample) lives.
|
||||||
Metrics ProbeMetrics
|
Metrics ProbeMetrics
|
||||||
|
// TraceProbes emits one root span per probe (--trace-health-probes).
|
||||||
|
// Off by default: probes run about once per second per proxy and would
|
||||||
|
// dominate trace volume.
|
||||||
|
TraceProbes bool
|
||||||
|
|
||||||
probeFn func(context.Context, *url.URL, crawlv1alpha1.HealthCheckSpec, *tls.Config) probeResult
|
probeFn func(context.Context, *url.URL, crawlv1alpha1.HealthCheckSpec, *tls.Config) probeResult
|
||||||
|
|
||||||
@@ -155,8 +163,7 @@ func (e *Engine) Start(ctx context.Context) error {
|
|||||||
case <-ctx.Done():
|
case <-ctx.Done():
|
||||||
return
|
return
|
||||||
case job := <-jobs:
|
case job := <-jobs:
|
||||||
res := e.probeFn(ctx, job.proxyURL, job.hc, e.ProbeTLSConfig)
|
e.record(job, e.runProbe(ctx, job), time.Now())
|
||||||
e.record(job, res, time.Now())
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
})
|
})
|
||||||
@@ -264,6 +271,29 @@ func newState(p *crawlv1alpha1.Proxy, now time.Time, interval time.Duration) *st
|
|||||||
return st
|
return st
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// runProbe executes one probe, inside its own root span when TraceProbes is
|
||||||
|
// on. The span ends before record folds the result (record stays ctx-free),
|
||||||
|
// and the probe transport is deliberately uninstrumented — no traceparent
|
||||||
|
// must ever leak through a proxy toward external targets.
|
||||||
|
func (e *Engine) runProbe(ctx context.Context, job probeJob) probeResult {
|
||||||
|
if !e.TraceProbes {
|
||||||
|
return e.probeFn(ctx, job.proxyURL, job.hc, e.ProbeTLSConfig)
|
||||||
|
}
|
||||||
|
ctx, span := tracing.Start(ctx, "health.probe",
|
||||||
|
trace.WithSpanKind(trace.SpanKindClient),
|
||||||
|
trace.WithAttributes(attribute.String("proxy", job.key.String())))
|
||||||
|
defer span.End()
|
||||||
|
res := e.probeFn(ctx, job.proxyURL, job.hc, e.ProbeTLSConfig)
|
||||||
|
span.SetAttributes(
|
||||||
|
attribute.Bool("probe.ok", res.ok),
|
||||||
|
attribute.Int64("probe.latency_ms", res.latency.Milliseconds()),
|
||||||
|
)
|
||||||
|
if !res.ok {
|
||||||
|
span.SetStatus(codes.Error, res.err.Error())
|
||||||
|
}
|
||||||
|
return res
|
||||||
|
}
|
||||||
|
|
||||||
// record folds one probe result into the proxy's threshold state and emits
|
// record folds one probe result into the proxy's threshold state and emits
|
||||||
// an event when the result is status-affecting: a first-ever verdict, a
|
// an event when the result is status-affecting: a first-ever verdict, a
|
||||||
// threshold-crossing flip, or a material latency change (beyond
|
// threshold-crossing flip, or a material latency change (beyond
|
||||||
|
|||||||
Reference in New Issue
Block a user