diff --git a/.gitignore b/.gitignore index 713cf538a8..2e5f4ec6a0 100644 --- a/.gitignore +++ b/.gitignore @@ -9,6 +9,9 @@ release *.dylib bin +# Locally built command binaries (e.g. `go build ./cmd/ghalistener`) +/ghalistener + # Test binary, build with `go test -c` *.test diff --git a/charts/gha-runner-scale-set-controller/templates/deployment.yaml b/charts/gha-runner-scale-set-controller/templates/deployment.yaml index 57ad21a3f3..01d326bfc8 100644 --- a/charts/gha-runner-scale-set-controller/templates/deployment.yaml +++ b/charts/gha-runner-scale-set-controller/templates/deployment.yaml @@ -81,6 +81,11 @@ spec: - "--listener-metrics-endpoint=" - "--metrics-addr=0" {{- end }} + {{- with .Values.otel }} + {{- with .listenerEndpoint }} + - "--listener-otel-endpoint={{ . }}" + {{- end }} + {{- end }} {{- if .Values.pprof.addr }} - "--pprof-addr={{ .Values.pprof.addr }}" {{- end }} diff --git a/charts/gha-runner-scale-set-controller/tests/template_test.go b/charts/gha-runner-scale-set-controller/tests/template_test.go index c84e27bee3..b8fbca9808 100644 --- a/charts/gha-runner-scale-set-controller/tests/template_test.go +++ b/charts/gha-runner-scale-set-controller/tests/template_test.go @@ -1218,3 +1218,59 @@ func TestNamespaceOverride(t *testing.T) { }) } } + +func TestTemplate_ControllerDeployment_ListenerOTelEndpoint(t *testing.T) { + t.Parallel() + + // Path to the helm chart we will test + helmChartPath, err := filepath.Abs("../../gha-runner-scale-set-controller") + require.NoError(t, err) + + releaseName := "test-arc" + namespaceName := "test-" + strings.ToLower(random.UniqueId()) + + options := &helm.Options{ + Logger: logger.Discard, + SetValues: map[string]string{ + "image.tag": "dev", + "otel.listenerEndpoint": "http://otel-collector:4318", + }, + KubectlOptions: k8s.NewKubectlOptions("", "", namespaceName), + } + + output := helm.RenderTemplate(t, options, helmChartPath, releaseName, []string{"templates/deployment.yaml"}) + + var deployment appsv1.Deployment + helm.UnmarshalK8SYaml(t, output, &deployment) + + assert.Contains(t, deployment.Spec.Template.Spec.Containers[0].Args, + "--listener-otel-endpoint=http://otel-collector:4318") +} + +func TestTemplate_ControllerDeployment_NoOTelByDefault(t *testing.T) { + t.Parallel() + + // Path to the helm chart we will test + helmChartPath, err := filepath.Abs("../../gha-runner-scale-set-controller") + require.NoError(t, err) + + releaseName := "test-arc" + namespaceName := "test-" + strings.ToLower(random.UniqueId()) + + options := &helm.Options{ + Logger: logger.Discard, + SetValues: map[string]string{ + "image.tag": "dev", + }, + KubectlOptions: k8s.NewKubectlOptions("", "", namespaceName), + } + + output := helm.RenderTemplate(t, options, helmChartPath, releaseName, []string{"templates/deployment.yaml"}) + + var deployment appsv1.Deployment + helm.UnmarshalK8SYaml(t, output, &deployment) + + for _, arg := range deployment.Spec.Template.Spec.Containers[0].Args { + assert.NotContains(t, arg, "--listener-otel-endpoint") + } +} diff --git a/charts/gha-runner-scale-set-controller/values.yaml b/charts/gha-runner-scale-set-controller/values.yaml index b83d505385..68a17ba776 100644 --- a/charts/gha-runner-scale-set-controller/values.yaml +++ b/charts/gha-runner-scale-set-controller/values.yaml @@ -98,6 +98,13 @@ priorityClassName: "" pprof: {} # addr: ":6060" +## To export OpenTelemetry traces of job lifecycle spans (queue, +## startup, execution) from the listener pods, uncomment the following +## lines and set the OTLP/HTTP collector base URL. TLS follows the URL +## scheme. +# otel: +# listenerEndpoint: "http://otel-collector:4318" + flags: ## Log level can be set here with one of the following values: "debug", "info", "warn", "error". ## Defaults to "debug". diff --git a/cmd/ghalistener/config/config.go b/cmd/ghalistener/config/config.go index 69cfafb8ee..8d7930e7d1 100644 --- a/cmd/ghalistener/config/config.go +++ b/cmd/ghalistener/config/config.go @@ -44,6 +44,11 @@ type Config struct { MetricsAddr string `json:"metrics_addr"` MetricsEndpoint string `json:"metrics_endpoint"` Metrics *v1alpha1.MetricsConfig `json:"metrics"` + // OTelEndpoint enables OTLP trace export of job lifecycle spans. + // Full base URL (like OTEL_EXPORTER_OTLP_ENDPOINT), e.g. + // http://collector:4318; TLS follows the URL scheme. When empty, + // the standard OTEL_EXPORTER_OTLP_* env vars still apply. + OTelEndpoint string `json:"otel_endpoint"` } func Read(ctx context.Context, configPath string) (*Config, error) { diff --git a/cmd/ghalistener/main.go b/cmd/ghalistener/main.go index 7dc4a17f46..8f4d76c986 100644 --- a/cmd/ghalistener/main.go +++ b/cmd/ghalistener/main.go @@ -4,9 +4,11 @@ import ( "context" "fmt" "log" + "net/url" "os" "os/signal" "syscall" + "time" "github.com/actions/actions-runner-controller/cmd/ghalistener/config" "github.com/actions/actions-runner-controller/cmd/ghalistener/metrics" @@ -14,6 +16,8 @@ import ( "github.com/actions/actions-runner-controller/github/actions" "github.com/actions/scaleset/listener" "github.com/google/uuid" + "go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp" + sdktrace "go.opentelemetry.io/otel/sdk/trace" "golang.org/x/sync/errgroup" ) @@ -39,6 +43,29 @@ func main() { } } +// newOTLPTraceExporter builds the OTLP/HTTP span exporter. A non-empty +// endpoint is a full base URL (the OTLP spec form of +// OTEL_EXPORTER_OTLP_ENDPOINT, e.g. http://collector:4318); the +// /v1/traces signal path is appended when the URL has no path. With an +// empty endpoint the SDK's spec-compliant OTEL_EXPORTER_OTLP_* env +// handling applies (endpoint, headers, TLS, ...). +func newOTLPTraceExporter(ctx context.Context, endpoint string) (sdktrace.SpanExporter, error) { + if endpoint == "" { + return otlptracehttp.New(ctx) + } + u, err := url.Parse(endpoint) + if err != nil { + return nil, fmt.Errorf("invalid OTel endpoint %q: %w", endpoint, err) + } + if u.Scheme != "http" && u.Scheme != "https" { + return nil, fmt.Errorf("invalid OTel endpoint %q: scheme must be http or https", endpoint) + } + if u.Path == "" || u.Path == "/" { + u.Path = "/v1/traces" + } + return otlptracehttp.New(ctx, otlptracehttp.WithEndpointURL(u.String())) +} + func run(ctx context.Context, config *config.Config) error { ghConfig, err := actions.ParseGitHubConfigFromURL(config.ConfigureURL) if err != nil { @@ -65,6 +92,42 @@ func run(ctx context.Context, config *config.Config) error { }) } + // OTel trace recorder (optional). Enabled by the listener config + // (otel_endpoint, wired from the controller) or by the standard + // OTEL_EXPORTER_OTLP_* environment variables. + var otelRecorder *metrics.OTelRecorder + otelEnabled := config.OTelEndpoint != "" || + os.Getenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT") != "" || + os.Getenv("OTEL_EXPORTER_OTLP_ENDPOINT") != "" + if otelEnabled { + otlpExporter, err := newOTLPTraceExporter(ctx, config.OTelEndpoint) + if err != nil { + return fmt.Errorf("failed to create OTel exporter: %w", err) + } + otelRecorder = metrics.NewOTelRecorder(metrics.OTelRecorderConfig{ + Exporter: otlpExporter, + Logger: logger.With("component", "otel recorder"), + ServerURL: ghConfig.ConfigURL.Scheme + "://" + ghConfig.ConfigURL.Host, + ScaleSetName: config.EphemeralRunnerSetName, + ScaleSetNamespace: config.EphemeralRunnerSetNamespace, + }) + // Shut down with a fresh bounded context: the signal ctx is + // already canceled on SIGINT/SIGTERM, which would abort the + // final flush of queued spans. + defer func() { + sctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := otelRecorder.Shutdown(sctx); err != nil { + logger.Error("Failed to shut down OTel recorder", "error", err) + } + }() + endpointSource := config.OTelEndpoint + if endpointSource == "" { + endpointSource = "OTEL_EXPORTER_OTLP_* environment" + } + logger.Info("OTel trace recorder enabled", "endpoint", endpointSource) + } + hostname, err := os.Hostname() if err != nil { hostname = uuid.NewString() @@ -91,13 +154,25 @@ func run(ctx context.Context, config *config.Config) error { }() var listenerOptions []listener.Option - if metricsExporter != nil { + + // Build the metrics recorder: Prometheus, OTel, or both + var recorder listener.MetricsRecorder + switch { + case metricsExporter != nil && otelRecorder != nil: + recorder = metrics.NewComposite(metricsExporter, otelRecorder) + case metricsExporter != nil: + recorder = metricsExporter + case otelRecorder != nil: + recorder = otelRecorder + } + + if recorder != nil { listenerOptions = append( listenerOptions, - listener.WithMetricsRecorder( - metricsExporter, - ), + listener.WithMetricsRecorder(recorder), ) + } + if metricsExporter != nil { metricsExporter.RecordStatic(config.MinRunners, config.MaxRunners) } diff --git a/cmd/ghalistener/main_test.go b/cmd/ghalistener/main_test.go new file mode 100644 index 0000000000..b6282a52d8 --- /dev/null +++ b/cmd/ghalistener/main_test.go @@ -0,0 +1,91 @@ +package main + +import ( + "context" + "net/http" + "net/http/httptest" + "sync" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel/sdk/trace/tracetest" +) + +type otlpCapture struct { + mu sync.Mutex + paths []string +} + +func (c *otlpCapture) handler() http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + c.mu.Lock() + c.paths = append(c.paths, r.URL.Path) + c.mu.Unlock() + w.Header().Set("Content-Type", "application/x-protobuf") + w.WriteHeader(http.StatusOK) + }) +} + +func (c *otlpCapture) Paths() []string { + c.mu.Lock() + defer c.mu.Unlock() + out := make([]string, len(c.paths)) + copy(out, c.paths) + return out +} + +func exportOneSpan(t *testing.T, ctx context.Context, endpoint string) { + t.Helper() + exp, err := newOTLPTraceExporter(ctx, endpoint) + require.NoError(t, err) + stubs := tracetest.SpanStubs{{Name: "test"}} + require.NoError(t, exp.ExportSpans(ctx, stubs.Snapshots())) + require.NoError(t, exp.Shutdown(ctx)) +} + +// The listener config carries a full base URL (the OTLP spec form of +// OTEL_EXPORTER_OTLP_ENDPOINT, e.g. http://collector:4318). The +// exporter must parse it and append the /v1/traces signal path — not +// treat it as a bare host:port. +func TestNewOTLPTraceExporter_ConfigEndpointFullURL(t *testing.T) { + capture := &otlpCapture{} + srv := httptest.NewServer(capture.handler()) + defer srv.Close() + + exportOneSpan(t, context.Background(), srv.URL) + + require.NotEmpty(t, capture.Paths(), "collector received no export") + assert.Equal(t, "/v1/traces", capture.Paths()[0]) +} + +func TestNewOTLPTraceExporter_ConfigEndpointExplicitPath(t *testing.T) { + capture := &otlpCapture{} + srv := httptest.NewServer(capture.handler()) + defer srv.Close() + + exportOneSpan(t, context.Background(), srv.URL+"/custom/v1/traces") + + require.NotEmpty(t, capture.Paths(), "collector received no export") + assert.Equal(t, "/custom/v1/traces", capture.Paths()[0]) +} + +// With no config endpoint, the SDK's spec-compliant OTEL_EXPORTER_OTLP_* +// env handling applies (endpoint, headers, TLS, ...). +func TestNewOTLPTraceExporter_StandardEnvVars(t *testing.T) { + capture := &otlpCapture{} + srv := httptest.NewServer(capture.handler()) + defer srv.Close() + + t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", srv.URL) + + exportOneSpan(t, context.Background(), "") + + require.NotEmpty(t, capture.Paths(), "collector received no export") + assert.Equal(t, "/v1/traces", capture.Paths()[0]) +} + +func TestNewOTLPTraceExporter_RejectsInvalidEndpoint(t *testing.T) { + _, err := newOTLPTraceExporter(context.Background(), "not a url") + assert.Error(t, err) +} diff --git a/cmd/ghalistener/metrics/composite.go b/cmd/ghalistener/metrics/composite.go new file mode 100644 index 0000000000..5935d3afeb --- /dev/null +++ b/cmd/ghalistener/metrics/composite.go @@ -0,0 +1,44 @@ +package metrics + +import ( + "github.com/actions/scaleset" + "github.com/actions/scaleset/listener" +) + +var _ listener.MetricsRecorder = &CompositeRecorder{} + +// CompositeRecorder delegates every MetricsRecorder call to all +// wrapped recorders. This allows the Prometheus exporter and OTel +// recorder to operate side by side. +type CompositeRecorder struct { + recorders []listener.MetricsRecorder +} + +// NewComposite creates a MetricsRecorder that fans out to all given recorders. +func NewComposite(recorders ...listener.MetricsRecorder) *CompositeRecorder { + return &CompositeRecorder{recorders: recorders} +} + +func (c *CompositeRecorder) RecordStatistics(stats *scaleset.RunnerScaleSetStatistic) { + for _, r := range c.recorders { + r.RecordStatistics(stats) + } +} + +func (c *CompositeRecorder) RecordJobStarted(msg *scaleset.JobStarted) { + for _, r := range c.recorders { + r.RecordJobStarted(msg) + } +} + +func (c *CompositeRecorder) RecordJobCompleted(msg *scaleset.JobCompleted) { + for _, r := range c.recorders { + r.RecordJobCompleted(msg) + } +} + +func (c *CompositeRecorder) RecordDesiredRunners(count int) { + for _, r := range c.recorders { + r.RecordDesiredRunners(count) + } +} diff --git a/cmd/ghalistener/metrics/otel.go b/cmd/ghalistener/metrics/otel.go new file mode 100644 index 0000000000..03cb1391cc --- /dev/null +++ b/cmd/ghalistener/metrics/otel.go @@ -0,0 +1,476 @@ +package metrics + +import ( + "context" + "crypto/sha256" + "errors" + "fmt" + "log/slog" + "strconv" + "strings" + "sync" + "time" + + "github.com/actions/actions-runner-controller/build" + "github.com/actions/scaleset" + "github.com/actions/scaleset/listener" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/sdk/instrumentation" + "go.opentelemetry.io/otel/sdk/resource" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + "go.opentelemetry.io/otel/sdk/trace/tracetest" + "go.opentelemetry.io/otel/trace" +) + +const otelScopeName = "github.com/actions/actions-runner-controller/cmd/ghalistener/metrics" + +var _ listener.MetricsRecorder = &OTelRecorder{} + +// OTelRecorder implements listener.MetricsRecorder by emitting +// OpenTelemetry trace spans for each completed job. Three child spans +// are created under the job's parent span using deterministic IDs: +// +// - runner.queue: QueueTime → ScaleSetAssignTime +// - runner.startup: ScaleSetAssignTime → RunnerAssignTime +// - runner.execution: RunnerAssignTime → FinishTime +// +// The deterministic ID scheme follows the normative contract in +// actions/runner docs/otel-id-contract.md (SHA-256, truncated): +// +// trace ID = sha256("{run_id}-{run_attempt}")[:16] +// job span = sha256("job-{run_id}-{run_attempt}-{job_name}")[:8] +// +// where job_name is the job's display name. The runner's native OTLP +// exporter emits the job span with these exact IDs, so ARC spans merge +// under it with zero correlation configuration. +// defaultRunAttempt is the run attempt hashed into trace/span IDs. The +// scale-set message protocol (github.com/actions/scaleset +// JobMessageBase) does not carry github.run_attempt, so the listener +// cannot know which attempt a completed job belongs to and assumes the +// first. +// +// KNOWN LIMITATION: for workflow re-runs (attempt >= 2) these spans +// join the FIRST attempt's trace — the runner hashes the real attempt +// into its trace ID, so the re-run's own trace is missing queue and +// startup data. Fixing this requires the scale-set message to carry +// the run attempt (or a REST lookup per job). +const defaultRunAttempt = 1 + +// spanQueueSize bounds the export queue. Export runs on a background +// goroutine so the listener's autoscaling message loop never waits on +// the telemetry backend; when the queue is full (collector down or +// slow), new job spans are dropped with a warning. +const spanQueueSize = 64 + +// exportTimeout bounds a single OTLP export attempt. +const exportTimeout = 5 * time.Second + +// maxPendingQueueEntries caps the in-memory map that holds +// queue-phase timestamps from JobStarted messages. Sized well above +// the realistic concurrent-job count on any single scale set; if the +// cap is reached a warning is logged and the new entry is dropped +// (the runner.queue span will be missing for that job). +const maxPendingQueueEntries = 1000 + +// pendingQueueTTL is the maximum age of an unmatched queue-time entry. +// Entries older than this are silently discarded at lookup so that +// jobs which never send a JobCompleted (e.g. abandoned mid-run) cannot +// leak memory across a months-long listener lifetime. +const pendingQueueTTL = 24 * time.Hour + +// pendingQueueEntry holds the queue-phase timestamps captured from a +// JobStarted message. They are keyed by JobID and evicted when the +// corresponding JobCompleted arrives. +// +// Background: the github.com broker omits QueueTime from JobCompleted +// in production (confirmed across multiple runs). This fallback +// captures it from the earlier JobStarted message instead — but note: +// as of 2026-07-01, github.com sends QueueTime=0 on JobStarted too +// (verified live; only ScaleSetAssignTime/RunnerAssignTime are +// populated), so today no runner.queue span is emitted at all. The +// path is kept because it is cheap, tested, and lights up the moment +// the broker starts populating the field (e.g. GHES or a future fix). +// +// If the broker also omits QueueTime from JobStarted (i.e. both +// fields are zero), no runner.queue span is emitted — using the +// listener-observed wall-clock receipt time as the queue start would +// produce a near-zero span anchored at the wrong end of the interval. +type pendingQueueEntry struct { + queueTime time.Time + scaleSetAssignTime time.Time + storedAt time.Time // wall-clock receipt; used for TTL only +} + +type OTelRecorder struct { + exporter sdktrace.SpanExporter + logger *slog.Logger + serverURL string + res *resource.Resource + scope instrumentation.Scope + + mu sync.Mutex + stopped bool + queue chan []sdktrace.ReadOnlySpan + workerDone chan struct{} + + // pendingMu guards pendingQueue. + pendingMu sync.Mutex + pendingQueue map[string]pendingQueueEntry +} + +// OTelRecorderConfig configures an OTelRecorder. +type OTelRecorderConfig struct { + // Exporter receives the job lifecycle spans. Required. + Exporter sdktrace.SpanExporter + Logger *slog.Logger + // ServerURL is the GitHub server base URL used for vcs.*/cicd.* + // URL attributes (GHES-safe). Defaults to https://github.com. + ServerURL string + // ScaleSetName and ScaleSetNamespace identify the scale set this + // listener serves; they are attached to the OTel resource. + ScaleSetName string + ScaleSetNamespace string +} + +// NewOTelRecorder creates a recorder that exports job lifecycle spans +// via the given SpanExporter. +func NewOTelRecorder(cfg OTelRecorderConfig) *OTelRecorder { + logger := cfg.Logger + if logger == nil { + logger = slog.New(slog.DiscardHandler) + } + + serverURL := strings.TrimRight(cfg.ServerURL, "/") + if serverURL == "" { + serverURL = "https://github.com" + } + + resAttrs := []attribute.KeyValue{ + attribute.String("service.name", "gha-listener"), + attribute.String("service.version", build.Version), + // The listener is the scheduling side of the CI/CD system + // (semconv cicd.system.component); the runner reports "agent". + attribute.String("cicd.system.component", "controller"), + } + if cfg.ScaleSetNamespace != "" { + resAttrs = append(resAttrs, attribute.String("service.namespace", cfg.ScaleSetNamespace)) + } + if cfg.ScaleSetName != "" { + resAttrs = append(resAttrs, attribute.String("github.scale_set.name", cfg.ScaleSetName)) + } + + // WithFromEnv comes last so the standard OTEL_SERVICE_NAME and + // OTEL_RESOURCE_ATTRIBUTES env vars override the defaults above. + res, err := resource.New(context.Background(), + resource.WithAttributes(resAttrs...), + resource.WithFromEnv(), + ) + if err != nil { + logger.Warn("failed to build OTel resource", "error", err) + } + if res == nil { + res = resource.Empty() + } + + r := &OTelRecorder{ + exporter: cfg.Exporter, + logger: logger, + serverURL: serverURL, + res: res, + scope: instrumentation.Scope{ + Name: otelScopeName, + Version: build.Version, + }, + queue: make(chan []sdktrace.ReadOnlySpan, spanQueueSize), + workerDone: make(chan struct{}), + pendingQueue: make(map[string]pendingQueueEntry), + } + go r.exportLoop() + return r +} + +// exportLoop drains the span queue on a background goroutine so +// RecordJobCompleted never blocks the caller. +func (r *OTelRecorder) exportLoop() { + defer close(r.workerDone) + for spans := range r.queue { + ctx, cancel := context.WithTimeout(context.Background(), exportTimeout) + if err := r.exporter.ExportSpans(ctx, spans); err != nil { + r.logger.Warn("failed to export OTel spans", "error", err) + } + cancel() + } +} + +// Shutdown drains queued spans (bounded by ctx) and releases exporter +// resources. It is idempotent; records arriving afterwards are dropped. +func (r *OTelRecorder) Shutdown(ctx context.Context) error { + r.mu.Lock() + if !r.stopped { + r.stopped = true + close(r.queue) + } + r.mu.Unlock() + + select { + case <-r.workerDone: + return r.exporter.Shutdown(ctx) + case <-ctx.Done(): + return errors.Join(ctx.Err(), r.exporter.Shutdown(ctx)) + } +} + +// RecordJobStarted captures the queue-phase timestamps from the +// JobStarted message so they are available when the corresponding +// JobCompleted arrives. The GitHub broker sends QueueTime=0 on +// JobCompleted in production; JobStarted carries the real value. +func (r *OTelRecorder) RecordJobStarted(msg *scaleset.JobStarted) { + r.pendingMu.Lock() + defer r.pendingMu.Unlock() + + if len(r.pendingQueue) >= maxPendingQueueEntries { + r.logger.Warn("OTel pending-queue map full, runner.queue span may be missing", + "job_id", msg.JobID, "run_id", msg.WorkflowRunID) + return + } + r.pendingQueue[msg.JobID] = pendingQueueEntry{ + queueTime: msg.QueueTime, + scaleSetAssignTime: msg.ScaleSetAssignTime, + storedAt: time.Now(), + } +} + +func (r *OTelRecorder) RecordJobCompleted(msg *scaleset.JobCompleted) { + // Retrieve and evict the pending entry stored at JobStarted time. + r.pendingMu.Lock() + pending, found := r.pendingQueue[msg.JobID] + delete(r.pendingQueue, msg.JobID) + r.pendingMu.Unlock() + + // Discard TTL-expired entries (abandoned jobs, listener restarts). + if found && time.Since(pending.storedAt) > pendingQueueTTL { + found = false + pending = pendingQueueEntry{} + } + + spans := r.buildJobSpans(msg, defaultRunAttempt, pending) + if len(spans) == 0 { + return + } + + r.mu.Lock() + defer r.mu.Unlock() + if r.stopped { + return + } + select { + case r.queue <- spans: + default: + r.logger.Warn("OTel span queue full, dropping job spans", + "job", msg.JobDisplayName, "run_id", msg.WorkflowRunID) + } +} + +func (r *OTelRecorder) RecordStatistics(_ *scaleset.RunnerScaleSetStatistic) {} +func (r *OTelRecorder) RecordDesiredRunners(_ int) {} + +// buildJobSpans constructs the runner.queue / runner.startup / +// runner.execution span stubs for the given completed job. +// +// pending holds the queue-phase timestamps captured from the earlier +// JobStarted message (see RecordJobStarted). It is used as a fallback +// when msg.QueueTime is zero, which is the observed production +// behaviour of the GitHub broker: JobCompleted never carries QueueTime +// while JobStarted does. The span timestamps therefore measure +// broker-set queue time, not listener-observed wall-clock time. +func (r *OTelRecorder) buildJobSpans(msg *scaleset.JobCompleted, attempt int64, pending pendingQueueEntry) []sdktrace.ReadOnlySpan { + traceID := newTraceID(msg.WorkflowRunID, attempt) + // Parent to the contract job span the runner emits. JobDisplayName + // is the job's display name (GitHub API job.name), the job_name the + // contract mandates — not the github.job YAML key. + parentSpanID := newJobSpanID(msg.WorkflowRunID, attempt, msg.JobDisplayName) + parentSC := trace.NewSpanContext(trace.SpanContextConfig{ + TraceID: traceID, + SpanID: parentSpanID, + TraceFlags: trace.FlagsSampled, + }) + + conclusion := normalizeConclusion(msg.Result) + repo := msg.OwnerName + "/" + msg.RepositoryName + runID := strconv.FormatInt(msg.WorkflowRunID, 10) + runAttempt := strconv.FormatInt(attempt, 10) + runURL := fmt.Sprintf("%s/%s/actions/runs/%s/attempts/%s", r.serverURL, repo, runID, runAttempt) + + // Attribute names, types, and values must match the runner's + // native OTLP exporter (Runner.Worker/OTelTraceExporter.cs) so a + // single query matches both span sets in the merged trace. + commonAttrs := []attribute.KeyValue{ + // GitHub Actions attributes (no semconv equivalent) + attribute.String("github.run_id", runID), + attribute.String("github.run_attempt", runAttempt), + attribute.String("github.repository", repo), + attribute.String("github.workflow_ref", msg.JobWorkflowRef), + attribute.String("github.event_name", msg.EventName), + // OTel CI/CD semantic conventions (cicd.*, vcs.*) + attribute.String("cicd.pipeline.run.id", runID), + attribute.String("cicd.pipeline.run.url.full", runURL), + attribute.String("cicd.pipeline.task.name", msg.JobDisplayName), + attribute.String("cicd.pipeline.task.run.id", parentSpanID.String()), + // cicd.worker.* are SPAN attrs here (not resource attrs) because one + // listener serves many runners; the runner itself emits them as + // resource attrs (one worker per process). Query both with TraceQL: + // span.cicd.worker.name || resource.cicd.worker.name + // See runner docs/otel-id-contract.md § cicd.worker scope. + attribute.String("cicd.worker.name", msg.RunnerName), + attribute.String("cicd.worker.id", strconv.Itoa(msg.RunnerID)), + attribute.String("vcs.repository.url.full", r.serverURL+"/"+repo), + attribute.String("vcs.provider.name", "github"), + } + + var stubs tracetest.SpanStubs + + // Resolve queue-phase start/end. Prefer timestamps from JobCompleted; + // fall back to those stored from the earlier JobStarted when the + // broker sends QueueTime=0 on JobCompleted (production behaviour). + queueStart := msg.QueueTime + queueEnd := msg.ScaleSetAssignTime + if queueStart.IsZero() && !pending.queueTime.IsZero() { + queueStart = pending.queueTime + if queueEnd.IsZero() { + queueEnd = pending.scaleSetAssignTime + } + } + + if !queueStart.IsZero() && !queueEnd.IsZero() { + stubs = append(stubs, r.newStub( + traceID, parentSC, "runner.queue", + queueStart, queueEnd, + // github.record_type matches the runner's attribute key (OTelTraceExporter.cs:526,618,711) + append(sliceClone(commonAttrs), attribute.String("github.record_type", "runner.queue")), + )) + } + + if !msg.ScaleSetAssignTime.IsZero() && !msg.RunnerAssignTime.IsZero() { + stubs = append(stubs, r.newStub( + traceID, parentSC, "runner.startup", + msg.ScaleSetAssignTime, msg.RunnerAssignTime, + append(sliceClone(commonAttrs), attribute.String("github.record_type", "runner.startup")), + )) + } + + if !msg.RunnerAssignTime.IsZero() && !msg.FinishTime.IsZero() { + stubs = append(stubs, r.newStub( + traceID, parentSC, "runner.execution", + msg.RunnerAssignTime, msg.FinishTime, + append(sliceClone(commonAttrs), + attribute.String("github.record_type", "runner.execution"), + attribute.String("github.conclusion", conclusion), + attribute.String("cicd.pipeline.task.run.result", conclusionToSemconv(conclusion)), + ), + )) + } + + return stubs.Snapshots() +} + +func normalizeConclusion(raw string) string { + // Mirrors the runner's NormalizeConclusion (OTelTraceExporter.cs:1008-1019). + // Abandoned maps to failure (TaskResult.Abandoned → "failure" in runner). + // Any unknown value maps to "error" rather than being lowercased through. + switch strings.ToLower(raw) { + case "success", "succeeded": + return "success" + case "failure", "failed": + return "failure" + case "abandoned": + return "failure" + case "cancelled", "canceled": + return "cancelled" + case "skipped": + return "skipped" + case "timed_out": + return "timed_out" + case "startup_failure": + return "startup_failure" + default: + return "error" + } +} + +func conclusionToSemconv(conclusion string) string { + // semconv cicd.pipeline.task.run.result enum: success|failure|cancellation|skip|timeout|error + // Mirrors ToSemconvResult (OTelTraceExporter.cs:1022-1033): unknown → "error". + switch conclusion { + case "success": + return "success" + case "failure": + return "failure" + case "cancelled": + return "cancellation" + case "skipped": + return "skip" + case "timed_out": + return "timeout" + default: + return "error" + } +} + +func (r *OTelRecorder) newStub( + traceID trace.TraceID, + parentSC trace.SpanContext, + name string, + start, end time.Time, + attrs []attribute.KeyValue, +) tracetest.SpanStub { + spanID := newSpanIDFromString(fmt.Sprintf("%s-%s", name, parentSC.SpanID())) + return tracetest.SpanStub{ + Name: name, + SpanContext: trace.NewSpanContext(trace.SpanContextConfig{ + TraceID: traceID, + SpanID: spanID, + TraceFlags: trace.FlagsSampled, + }), + Parent: parentSC, + StartTime: start, + EndTime: end, + Attributes: attrs, + Resource: r.res, + InstrumentationScope: r.scope, + } +} + +// Deterministic ID generation per the normative contract in +// actions/runner docs/otel-id-contract.md: leading bytes of SHA-256 +// (never MD5 — MD5 fails on FIPS-enabled hosts). See +// TestIDContractGoldenVectors for the shared conformance vectors. + +func newTraceID(runID, runAttempt int64) trace.TraceID { + if runAttempt == 0 { + runAttempt = 1 + } + sum := sha256.Sum256([]byte(fmt.Sprintf("%d-%d", runID, runAttempt))) + var tid trace.TraceID + copy(tid[:], sum[:16]) + return tid +} + +func newSpanIDFromString(s string) trace.SpanID { + sum := sha256.Sum256([]byte(s)) + var sid trace.SpanID + copy(sid[:], sum[:8]) + return sid +} + +func newJobSpanID(runID, runAttempt int64, jobName string) trace.SpanID { + if runAttempt == 0 { + runAttempt = 1 + } + return newSpanIDFromString(fmt.Sprintf("job-%d-%d-%s", runID, runAttempt, jobName)) +} + +func sliceClone(s []attribute.KeyValue) []attribute.KeyValue { + out := make([]attribute.KeyValue, len(s)) + copy(out, s) + return out +} diff --git a/cmd/ghalistener/metrics/otel_test.go b/cmd/ghalistener/metrics/otel_test.go new file mode 100644 index 0000000000..8db57e2150 --- /dev/null +++ b/cmd/ghalistener/metrics/otel_test.go @@ -0,0 +1,674 @@ +package metrics + +import ( + "context" + "sync" + "testing" + "time" + + "github.com/actions/scaleset" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + sdktrace "go.opentelemetry.io/otel/sdk/trace" +) + +type captureExporter struct { + mu sync.Mutex + spans []sdktrace.ReadOnlySpan +} + +func (e *captureExporter) ExportSpans(_ context.Context, spans []sdktrace.ReadOnlySpan) error { + e.mu.Lock() + defer e.mu.Unlock() + e.spans = append(e.spans, spans...) + return nil +} + +func (e *captureExporter) Shutdown(_ context.Context) error { return nil } + +func (e *captureExporter) Spans() []sdktrace.ReadOnlySpan { + e.mu.Lock() + defer e.mu.Unlock() + out := make([]sdktrace.ReadOnlySpan, len(e.spans)) + copy(out, e.spans) + return out +} + +func newTestRecorder(t *testing.T, exp sdktrace.SpanExporter) *OTelRecorder { + t.Helper() + return NewOTelRecorder(OTelRecorderConfig{Exporter: exp}) +} + +// flush drains the recorder's async export queue so tests can assert +// on the exported spans deterministically. +func flush(t *testing.T, rec *OTelRecorder) { + t.Helper() + require.NoError(t, rec.Shutdown(context.Background())) +} + +func TestOTelRecorder_EmitsThreeSpans(t *testing.T) { + exp := &captureExporter{} + rec := newTestRecorder(t, exp) + + now := time.Date(2026, 4, 21, 12, 0, 0, 0, time.UTC) + msg := &scaleset.JobCompleted{ + Result: "Succeeded", + RunnerID: 7, + RunnerName: "runner-abc-xyz", + JobMessageBase: scaleset.JobMessageBase{ + RunnerRequestID: 1, + WorkflowRunID: 99999, + JobID: "42", + JobDisplayName: "build", + OwnerName: "acme", + RepositoryName: "widgets", + JobWorkflowRef: "acme/widgets/.github/workflows/ci.yml@refs/heads/main", + QueueTime: now, + ScaleSetAssignTime: now.Add(10 * time.Second), + RunnerAssignTime: now.Add(40 * time.Second), + FinishTime: now.Add(5 * time.Minute), + }, + } + + rec.RecordJobCompleted(msg) + flush(t, rec) + + spans := exp.Spans() + require.Len(t, spans, 3) + + byName := map[string]sdktrace.ReadOnlySpan{} + for _, s := range spans { + byName[s.Name()] = s + } + + expectedTraceID := newTraceID(99999, 1) + for _, s := range spans { + assert.Equal(t, expectedTraceID, s.SpanContext().TraceID()) + } + + // Parent must be the contract job span ID the runner emits: + // sha256("job-{run_id}-{run_attempt}-{job_name}")[:8]. + expectedParent := newJobSpanID(99999, 1, "build") + for _, s := range spans { + assert.Equal(t, expectedParent, s.Parent().SpanID()) + } + assert.Equal(t, "81606d47848a59c0", expectedParent.String()) + + q := byName["runner.queue"] + require.NotNil(t, q) + assert.Equal(t, now, q.StartTime()) + assert.Equal(t, now.Add(10*time.Second), q.EndTime()) + // M1: key must be github.record_type (matches runner's OTelTraceExporter.cs:526,618,711) + assertAttr(t, q, "github.record_type", "runner.queue") + + s := byName["runner.startup"] + require.NotNil(t, s) + assert.Equal(t, now.Add(10*time.Second), s.StartTime()) + assert.Equal(t, now.Add(40*time.Second), s.EndTime()) + assertAttr(t, s, "github.record_type", "runner.startup") + + e := byName["runner.execution"] + require.NotNil(t, e) + assert.Equal(t, now.Add(40*time.Second), e.StartTime()) + assert.Equal(t, now.Add(5*time.Minute), e.EndTime()) + assertAttr(t, e, "github.record_type", "runner.execution") + assertAttr(t, e, "github.conclusion", "success") + assertAttr(t, e, "cicd.pipeline.task.run.result", "success") + + // Shared attributes must match the runner exporter's names, types, + // and values so a single TraceQL query matches both span sets. + for _, span := range byName { + assertAttr(t, span, "github.run_id", "99999") + assertAttr(t, span, "github.run_attempt", "1") + assertAttr(t, span, "github.repository", "acme/widgets") + assertAttr(t, span, "cicd.pipeline.run.id", "99999") + assertAttr(t, span, "cicd.pipeline.run.url.full", "https://github.com/acme/widgets/actions/runs/99999/attempts/1") + assertAttr(t, span, "cicd.pipeline.task.name", "build") + assertAttr(t, span, "cicd.pipeline.task.run.id", "81606d47848a59c0") + assertAttr(t, span, "cicd.worker.name", "runner-abc-xyz") + assertAttr(t, span, "cicd.worker.id", "7") + assertAttr(t, span, "vcs.repository.url.full", "https://github.com/acme/widgets") + assertAttr(t, span, "vcs.provider.name", "github") + + // Old key must not appear; runner uses github.record_type (M1). + assertNoAttr(t, span, "type") + // Keys the runner never emits must not be invented here. + assertNoAttr(t, span, "github.job_id") + assertNoAttr(t, span, "github.job_name") + assertNoAttr(t, span, "github.runner_name") + assertNoAttr(t, span, "github.runner_id") + } +} + +func TestOTelRecorder_ServerURLIsConfigurable(t *testing.T) { + exp := &captureExporter{} + rec := NewOTelRecorder(OTelRecorderConfig{ + Exporter: exp, + ServerURL: "https://ghes.example.com/", + }) + + now := time.Now() + msg := &scaleset.JobCompleted{ + Result: "Succeeded", + JobMessageBase: scaleset.JobMessageBase{ + WorkflowRunID: 12345, + JobID: "1", + JobDisplayName: "build", + OwnerName: "org", + RepositoryName: "repo", + RunnerAssignTime: now, + FinishTime: now.Add(time.Minute), + }, + } + + rec.RecordJobCompleted(msg) + flush(t, rec) + + spans := exp.Spans() + require.Len(t, spans, 1) + assertAttr(t, spans[0], "vcs.repository.url.full", "https://ghes.example.com/org/repo") + assertAttr(t, spans[0], "cicd.pipeline.run.url.full", "https://ghes.example.com/org/repo/actions/runs/12345/attempts/1") +} + +func TestOTelRecorder_ResourceAndScope(t *testing.T) { + exp := &captureExporter{} + rec := NewOTelRecorder(OTelRecorderConfig{ + Exporter: exp, + ScaleSetName: "my-scale-set", + ScaleSetNamespace: "arc-runners", + }) + + now := time.Now() + msg := &scaleset.JobCompleted{ + Result: "Succeeded", + JobMessageBase: scaleset.JobMessageBase{ + WorkflowRunID: 1, + JobID: "1", + RunnerAssignTime: now, + FinishTime: now.Add(time.Second), + }, + } + + rec.RecordJobCompleted(msg) + flush(t, rec) + + spans := exp.Spans() + require.Len(t, spans, 1) + s := spans[0] + + res := s.Resource() + require.NotNil(t, res) + got := map[string]string{} + for _, kv := range res.Attributes() { + got[string(kv.Key)] = kv.Value.AsString() + } + assert.Equal(t, "gha-listener", got["service.name"]) + assert.NotEmpty(t, got["service.version"]) + assert.Equal(t, "arc-runners", got["service.namespace"]) + assert.Equal(t, "my-scale-set", got["github.scale_set.name"]) + + scope := s.InstrumentationScope() + assert.Equal(t, "github.com/actions/actions-runner-controller/cmd/ghalistener/metrics", scope.Name) +} + +func TestOTelRecorder_ResourceHonorsOTelEnv(t *testing.T) { + t.Setenv("OTEL_SERVICE_NAME", "my-custom-listener") + + exp := &captureExporter{} + rec := newTestRecorder(t, exp) + + now := time.Now() + msg := &scaleset.JobCompleted{ + Result: "Succeeded", + JobMessageBase: scaleset.JobMessageBase{ + WorkflowRunID: 1, + JobID: "1", + RunnerAssignTime: now, + FinishTime: now.Add(time.Second), + }, + } + + rec.RecordJobCompleted(msg) + flush(t, rec) + + spans := exp.Spans() + require.Len(t, spans, 1) + for _, kv := range spans[0].Resource().Attributes() { + if string(kv.Key) == "service.name" { + assert.Equal(t, "my-custom-listener", kv.Value.AsString()) + return + } + } + t.Error("service.name not found on resource") +} + +func TestOTelRecorder_SkipsMissingTimestamps(t *testing.T) { + exp := &captureExporter{} + rec := newTestRecorder(t, exp) + + now := time.Now() + msg := &scaleset.JobCompleted{ + Result: "Succeeded", + JobMessageBase: scaleset.JobMessageBase{ + WorkflowRunID: 12345, + JobID: "1", + RunnerAssignTime: now, + FinishTime: now.Add(time.Minute), + }, + } + + rec.RecordJobCompleted(msg) + flush(t, rec) + + spans := exp.Spans() + require.Len(t, spans, 1, "only runner.execution when queue/startup timestamps are zero") + assert.Equal(t, "runner.execution", spans[0].Name()) +} + +func TestOTelRecorder_CommonAttributes(t *testing.T) { + exp := &captureExporter{} + rec := newTestRecorder(t, exp) + + now := time.Now() + msg := &scaleset.JobCompleted{ + Result: "Failed", + RunnerID: 3, + RunnerName: "runner-3", + JobMessageBase: scaleset.JobMessageBase{ + WorkflowRunID: 55555, + JobID: "100", + JobDisplayName: "test-suite", + OwnerName: "org", + RepositoryName: "repo", + JobWorkflowRef: "org/repo/.github/workflows/test.yml@main", + RunnerAssignTime: now, + FinishTime: now.Add(2 * time.Minute), + }, + } + + rec.RecordJobCompleted(msg) + flush(t, rec) + + spans := exp.Spans() + require.Len(t, spans, 1) + s := spans[0] + + assertAttr(t, s, "cicd.pipeline.task.name", "test-suite") + assertAttr(t, s, "github.repository", "org/repo") + assertAttr(t, s, "cicd.worker.name", "runner-3") + assertAttr(t, s, "github.conclusion", "failure") + assertAttr(t, s, "cicd.pipeline.task.run.result", "failure") +} + +// blockingExporter blocks every ExportSpans call until released. +type blockingExporter struct { + captureExporter + release chan struct{} +} + +func (e *blockingExporter) ExportSpans(ctx context.Context, spans []sdktrace.ReadOnlySpan) error { + <-e.release + return e.captureExporter.ExportSpans(ctx, spans) +} + +func completedMsg(runID int64) *scaleset.JobCompleted { + now := time.Now() + return &scaleset.JobCompleted{ + Result: "Succeeded", + JobMessageBase: scaleset.JobMessageBase{ + WorkflowRunID: runID, + JobID: "1", + RunnerAssignTime: now, + FinishTime: now.Add(time.Second), + }, + } +} + +// The autoscaling message loop calls RecordJobCompleted synchronously; +// it must never wait on the telemetry backend. +func TestOTelRecorder_RecordDoesNotBlockOnSlowExporter(t *testing.T) { + exp := &blockingExporter{release: make(chan struct{})} + defer close(exp.release) + rec := newTestRecorder(t, exp) + + done := make(chan struct{}) + go func() { + defer close(done) + for i := range 100 { + rec.RecordJobCompleted(completedMsg(int64(i + 1))) + } + }() + + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("RecordJobCompleted blocked on a stalled exporter") + } +} + +func TestOTelRecorder_ShutdownFlushesQueuedSpans(t *testing.T) { + exp := &captureExporter{} + rec := newTestRecorder(t, exp) + + for i := range 10 { + rec.RecordJobCompleted(completedMsg(int64(i + 1))) + } + + require.NoError(t, rec.Shutdown(context.Background())) + assert.Len(t, exp.Spans(), 10) +} + +func TestOTelRecorder_RecordAfterShutdownIsNoOp(t *testing.T) { + exp := &captureExporter{} + rec := newTestRecorder(t, exp) + require.NoError(t, rec.Shutdown(context.Background())) + + rec.RecordJobCompleted(completedMsg(1)) + + assert.Empty(t, exp.Spans()) + require.NoError(t, rec.Shutdown(context.Background()), "Shutdown must be idempotent") +} + +func TestOTelRecorder_ShutdownHonorsContext(t *testing.T) { + exp := &blockingExporter{release: make(chan struct{})} + defer close(exp.release) + rec := newTestRecorder(t, exp) + + rec.RecordJobCompleted(completedMsg(1)) + + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + err := rec.Shutdown(ctx) + assert.ErrorIs(t, err, context.DeadlineExceeded) +} + +func TestOTelRecorder_JobStartedIsNoOp(t *testing.T) { + exp := &captureExporter{} + rec := newTestRecorder(t, exp) + rec.RecordJobStarted(&scaleset.JobStarted{}) + assert.Empty(t, exp.Spans()) +} + +func TestOTelRecorder_StatisticsIsNoOp(t *testing.T) { + exp := &captureExporter{} + rec := newTestRecorder(t, exp) + rec.RecordStatistics(&scaleset.RunnerScaleSetStatistic{TotalRunningJobs: 5}) + assert.Empty(t, exp.Spans()) +} + +func TestCompositeRecorder_DelegatesAll(t *testing.T) { + exp1 := &captureExporter{} + exp2 := &captureExporter{} + r1 := newTestRecorder(t, exp1) + r2 := newTestRecorder(t, exp2) + comp := NewComposite(r1, r2) + + now := time.Now() + msg := &scaleset.JobCompleted{ + Result: "Succeeded", + JobMessageBase: scaleset.JobMessageBase{ + WorkflowRunID: 1, + JobID: "1", + RunnerAssignTime: now, + FinishTime: now.Add(time.Second), + }, + } + comp.RecordJobCompleted(msg) + flush(t, r1) + flush(t, r2) + + assert.Len(t, exp1.Spans(), 1) + assert.Len(t, exp2.Spans(), 1) +} + +// Conformance vectors from the normative ID contract +// (actions/runner docs/otel-id-contract.md). A conforming +// implementation MUST reproduce these exactly; the runner's L0 suite +// asserts the same values. +func TestIDContractGoldenVectors(t *testing.T) { + assert.Equal(t, "acad1e2a107636235fd56bb742499bd0", newTraceID(99999, 1).String(), + `trace ID must be sha256("{run_id}-{run_attempt}")[:16]`) + assert.Equal(t, "81606d47848a59c0", newJobSpanID(99999, 1, "build").String(), + `job span ID must be sha256("job-{run_id}-{run_attempt}-{job_name}")[:8]`) + assert.Equal(t, newTraceID(99999, 1), newTraceID(99999, 0), + "run_attempt 0 must normalize to 1") + assert.Equal(t, newJobSpanID(99999, 1, "build"), newJobSpanID(99999, 0, "build"), + "run_attempt 0 must normalize to 1") +} + +func TestDeterministicIDs(t *testing.T) { + tid1 := newTraceID(12345, 1) + tid2 := newTraceID(12345, 1) + tid3 := newTraceID(12345, 2) + assert.Equal(t, tid1, tid2, "same inputs must produce same trace ID") + assert.NotEqual(t, tid1, tid3, "different attempt must produce different trace ID") + + sid1 := newJobSpanID(42, 1, "build") + sid2 := newJobSpanID(42, 1, "build") + sid3 := newJobSpanID(42, 1, "deploy") + assert.Equal(t, sid1, sid2) + assert.NotEqual(t, sid1, sid3) +} + +// TestNormalizeConclusion_Conformance verifies M3: ARC must mirror the runner's +// NormalizeConclusion mapping (OTelTraceExporter.cs:1008-1019). +// Abandoned → "failure"; any unknown raw value → "error" (not lowercased pass-through). +func TestNormalizeConclusion_Conformance(t *testing.T) { + cases := []struct { + raw string + want string + }{ + // known set passes through + {"success", "success"}, + {"Succeeded", "success"}, + {"failure", "failure"}, + {"Failed", "failure"}, + {"cancelled", "cancelled"}, + {"Canceled", "cancelled"}, + {"skipped", "skipped"}, + {"timed_out", "timed_out"}, + {"startup_failure", "startup_failure"}, + // abandoned maps to failure (mirrors TaskResult.Abandoned in runner) + {"abandoned", "failure"}, + {"Abandoned", "failure"}, + // unknown raw values → "error", not lowercased pass-through + {"totally_unknown", "error"}, + {"", "error"}, + } + for _, tc := range cases { + t.Run(tc.raw, func(t *testing.T) { + assert.Equal(t, tc.want, normalizeConclusion(tc.raw)) + }) + } +} + +// TestConclusionToSemconv_Conformance verifies M3: unknown conclusion must map to +// "error" for cicd.pipeline.task.run.result (mirrors ToSemconvResult default in runner). +func TestConclusionToSemconv_Conformance(t *testing.T) { + cases := []struct { + in string + want string + }{ + {"success", "success"}, + {"failure", "failure"}, + {"cancelled", "cancellation"}, + {"skipped", "skip"}, + {"timed_out", "timeout"}, + {"startup_failure", "error"}, + // unknown → "error", not the raw value passed through + {"error", "error"}, + {"unknown_value", "error"}, + } + for _, tc := range cases { + t.Run(tc.in, func(t *testing.T) { + assert.Equal(t, tc.want, conclusionToSemconv(tc.in)) + }) + } +} + +// TestOTelRecorder_AbandonedConclusion verifies M3 end-to-end: +// a job with Result="Abandoned" must emit github.conclusion="failure" +// and cicd.pipeline.task.run.result="failure". +func TestOTelRecorder_AbandonedConclusion(t *testing.T) { + exp := &captureExporter{} + rec := newTestRecorder(t, exp) + + now := time.Now() + msg := &scaleset.JobCompleted{ + Result: "Abandoned", + JobMessageBase: scaleset.JobMessageBase{ + WorkflowRunID: 1, + JobID: "1", + RunnerAssignTime: now, + FinishTime: now.Add(time.Second), + }, + } + rec.RecordJobCompleted(msg) + flush(t, rec) + + spans := exp.Spans() + require.Len(t, spans, 1) + assertAttr(t, spans[0], "github.conclusion", "failure") + assertAttr(t, spans[0], "cicd.pipeline.task.run.result", "failure") +} + +// TestOTelRecorder_UnknownConclusion verifies M3 end-to-end: +// an unrecognised Result must emit github.conclusion="error" +// and cicd.pipeline.task.run.result="error" (not the raw lowercased value). +func TestOTelRecorder_UnknownConclusion(t *testing.T) { + exp := &captureExporter{} + rec := newTestRecorder(t, exp) + + now := time.Now() + msg := &scaleset.JobCompleted{ + Result: "SomeNewFutureState", + JobMessageBase: scaleset.JobMessageBase{ + WorkflowRunID: 1, + JobID: "1", + RunnerAssignTime: now, + FinishTime: now.Add(time.Second), + }, + } + rec.RecordJobCompleted(msg) + flush(t, rec) + + spans := exp.Spans() + require.Len(t, spans, 1) + assertAttr(t, spans[0], "github.conclusion", "error") + assertAttr(t, spans[0], "cicd.pipeline.task.run.result", "error") +} + +// TestOTelRecorder_QueueSpanFromJobStarted verifies that runner.queue is +// emitted even when JobCompleted.QueueTime is zero (the confirmed +// production behaviour: the GitHub broker omits QueueTime from +// JobCompleted). The recorder must fall back to the QueueTime and +// ScaleSetAssignTime captured from the earlier JobStarted message, +// keyed by JobID. +func TestOTelRecorder_QueueSpanFromJobStarted(t *testing.T) { + exp := &captureExporter{} + rec := newTestRecorder(t, exp) + + now := time.Date(2026, 4, 21, 12, 0, 0, 0, time.UTC) + + // JobStarted arrives first and carries the real queue-phase timestamps. + rec.RecordJobStarted(&scaleset.JobStarted{ + JobMessageBase: scaleset.JobMessageBase{ + JobID: "job-42", + WorkflowRunID: 99999, + JobDisplayName: "build", + QueueTime: now, + ScaleSetAssignTime: now.Add(10 * time.Second), + RunnerAssignTime: now.Add(40 * time.Second), + }, + }) + + // JobCompleted arrives with QueueTime=0 — the broker bug. + rec.RecordJobCompleted(&scaleset.JobCompleted{ + Result: "Succeeded", + JobMessageBase: scaleset.JobMessageBase{ + JobID: "job-42", + WorkflowRunID: 99999, + JobDisplayName: "build", + OwnerName: "acme", + RepositoryName: "widgets", + QueueTime: time.Time{}, // zero — broker does not send it + ScaleSetAssignTime: now.Add(10 * time.Second), + RunnerAssignTime: now.Add(40 * time.Second), + FinishTime: now.Add(5 * time.Minute), + }, + }) + flush(t, rec) + + spans := exp.Spans() + byName := map[string]sdktrace.ReadOnlySpan{} + for _, s := range spans { + byName[s.Name()] = s + } + + q := byName["runner.queue"] + require.NotNil(t, q, "runner.queue must fire even when JobCompleted.QueueTime=0") + assert.Equal(t, now, q.StartTime(), "start must be QueueTime from JobStarted") + assert.Equal(t, now.Add(10*time.Second), q.EndTime(), "end must be ScaleSetAssignTime") +} + +// TestOTelRecorder_QueueEntryEvictedOnCompleted verifies that the +// pending-queue map is cleared after JobCompleted so a second +// JobCompleted for the same JobID does not re-emit runner.queue. +func TestOTelRecorder_QueueEntryEvictedOnCompleted(t *testing.T) { + exp := &captureExporter{} + rec := newTestRecorder(t, exp) + + now := time.Now() + + rec.RecordJobStarted(&scaleset.JobStarted{ + JobMessageBase: scaleset.JobMessageBase{ + JobID: "job-99", + WorkflowRunID: 1, + QueueTime: now, + ScaleSetAssignTime: now.Add(5 * time.Second), + }, + }) + + first := &scaleset.JobCompleted{ + Result: "Succeeded", + JobMessageBase: scaleset.JobMessageBase{ + JobID: "job-99", + WorkflowRunID: 1, + RunnerAssignTime: now.Add(10 * time.Second), + FinishTime: now.Add(time.Minute), + }, + } + rec.RecordJobCompleted(first) + + // Simulate a duplicate/stale JobCompleted for the same job. + rec.RecordJobCompleted(first) + flush(t, rec) + + queueCount := 0 + for _, s := range exp.Spans() { + if s.Name() == "runner.queue" { + queueCount++ + } + } + assert.Equal(t, 1, queueCount, "runner.queue must be emitted exactly once per job") +} + +func assertAttr(t *testing.T, span sdktrace.ReadOnlySpan, key, expected string) { + t.Helper() + for _, a := range span.Attributes() { + if string(a.Key) == key { + assert.Equal(t, expected, a.Value.AsString(), "attribute %q", key) + return + } + } + t.Errorf("attribute %q not found on span %q", key, span.Name()) +} + +func assertNoAttr(t *testing.T, span sdktrace.ReadOnlySpan, key string) { + t.Helper() + for _, a := range span.Attributes() { + if string(a.Key) == key { + t.Errorf("attribute %q must not be set on span %q", key, span.Name()) + } + } +} diff --git a/controllers/actions.github.com/autoscalinglistener_controller.go b/controllers/actions.github.com/autoscalinglistener_controller.go index 7d12f895d6..ab8dd7e243 100644 --- a/controllers/actions.github.com/autoscalinglistener_controller.go +++ b/controllers/actions.github.com/autoscalinglistener_controller.go @@ -56,6 +56,11 @@ type AutoscalingListenerReconciler struct { // If it is set to "0", the metrics server is not started. ListenerMetricsAddr string ListenerMetricsEndpoint string + // ListenerOTelEndpoint is the OTLP/HTTP collector base URL for the + // listener's job lifecycle trace export. Empty disables the config + // wiring (the standard OTEL_EXPORTER_OTLP_* env vars still apply + // on the listener pod, e.g. via listenerTemplate). + ListenerOTelEndpoint string ResourceBuilder } @@ -389,7 +394,7 @@ func (r *AutoscalingListenerReconciler) Reconcile(ctx context.Context, req ctrl. return ctrl.Result{}, fmt.Errorf("failed to build GitHub server TLS certificate value for listener config: %w", err) } } - desiredSecret, err := r.newScaleSetListenerConfig(&autoscalingListener, cfg, metricsConfig, cert) + desiredSecret, err := r.newScaleSetListenerConfig(&autoscalingListener, cfg, metricsConfig, cert, r.ListenerOTelEndpoint) if err != nil { return ctrl.Result{}, fmt.Errorf("failed to build listener config secret: %w", err) } @@ -426,7 +431,7 @@ func (r *AutoscalingListenerReconciler) Reconcile(ctx context.Context, req ctrl. return ctrl.Result{}, fmt.Errorf("failed to build GitHub server TLS certificate value for listener config: %w", err) } } - desiredSecret, err := r.newScaleSetListenerConfig(&autoscalingListener, cfg, metricsConfig, cert) + desiredSecret, err := r.newScaleSetListenerConfig(&autoscalingListener, cfg, metricsConfig, cert, r.ListenerOTelEndpoint) if err != nil { return ctrl.Result{}, fmt.Errorf("failed to build listener config secret: %w", err) } diff --git a/controllers/actions.github.com/resourcebuilder.go b/controllers/actions.github.com/resourcebuilder.go index fd456b56f1..ad7f700e03 100644 --- a/controllers/actions.github.com/resourcebuilder.go +++ b/controllers/actions.github.com/resourcebuilder.go @@ -207,7 +207,7 @@ func (lm *listenerMetricsServerConfig) containerPort() (corev1.ContainerPort, er }, nil } -func (b *ResourceBuilder) newScaleSetListenerConfig(autoscalingListener *v1alpha1.AutoscalingListener, appConfig *appconfig.AppConfig, metricsConfig *listenerMetricsServerConfig, cert string) (*corev1.Secret, error) { +func (b *ResourceBuilder) newScaleSetListenerConfig(autoscalingListener *v1alpha1.AutoscalingListener, appConfig *appconfig.AppConfig, metricsConfig *listenerMetricsServerConfig, cert string, otelEndpoint string) (*corev1.Secret, error) { var ( metricsAddr = "" metricsEndpoint = "" @@ -231,6 +231,7 @@ func (b *ResourceBuilder) newScaleSetListenerConfig(autoscalingListener *v1alpha MetricsAddr: metricsAddr, MetricsEndpoint: metricsEndpoint, Metrics: autoscalingListener.Spec.Metrics, + OTelEndpoint: otelEndpoint, } vault := autoscalingListener.Spec.VaultConfig diff --git a/controllers/actions.github.com/resourcebuilder_test.go b/controllers/actions.github.com/resourcebuilder_test.go index d08851173a..373d8d0d1b 100644 --- a/controllers/actions.github.com/resourcebuilder_test.go +++ b/controllers/actions.github.com/resourcebuilder_test.go @@ -1,16 +1,19 @@ package actionsgithubcom import ( + "encoding/json" "fmt" "strings" "testing" "github.com/actions/actions-runner-controller/apis/actions.github.com/v1alpha1" + "github.com/actions/actions-runner-controller/apis/actions.github.com/v1alpha1/appconfig" "github.com/actions/scaleset" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/utils/ptr" ) func TestMetadataPropagation(t *testing.T) { @@ -542,3 +545,55 @@ func TestListenerPodNodeSelector(t *testing.T) { "explicitly empty nodeSelector should override the linux default") }) } + +func TestListenerConfigOTelEndpoint(t *testing.T) { + autoscalingRunnerSet := v1alpha1.AutoscalingRunnerSet{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-scale-set", + Namespace: "test-ns", + Labels: map[string]string{ + LabelKeyKubernetesPartOf: labelValueKubernetesPartOf, + LabelKeyKubernetesVersion: "0.2.0", + }, + Annotations: map[string]string{ + runnerScaleSetIDAnnotationKey: "1", + AnnotationKeyGitHubRunnerGroupName: "test-group", + AnnotationKeyGitHubRunnerScaleSetName: "test-scale-set", + }, + }, + Spec: v1alpha1.AutoscalingRunnerSetSpec{ + GitHubConfigUrl: "https://github.com/org/repo", + MaxRunners: ptr.To(5), + }, + } + + b := ResourceBuilder{} + ephemeralRunnerSet, err := b.newEphemeralRunnerSet(&autoscalingRunnerSet) + require.NoError(t, err) + + listener, err := b.newAutoscalingListener(&autoscalingRunnerSet, ephemeralRunnerSet, autoscalingRunnerSet.Namespace, "test:latest", nil) + require.NoError(t, err) + + appConfig := &appconfig.AppConfig{Token: "token"} + + decodeConfig := func(t *testing.T, secret *corev1.Secret) map[string]any { + t.Helper() + var cfg map[string]any + require.NoError(t, json.Unmarshal(secret.Data["config.json"], &cfg)) + return cfg + } + + t.Run("otel endpoint flows into config.json", func(t *testing.T) { + secret, err := b.newScaleSetListenerConfig(listener, appConfig, nil, "", "http://otel-collector:4318") + require.NoError(t, err) + cfg := decodeConfig(t, secret) + assert.Equal(t, "http://otel-collector:4318", cfg["otel_endpoint"]) + }) + + t.Run("empty endpoint leaves config empty", func(t *testing.T) { + secret, err := b.newScaleSetListenerConfig(listener, appConfig, nil, "", "") + require.NoError(t, err) + cfg := decodeConfig(t, secret) + assert.Equal(t, "", cfg["otel_endpoint"]) + }) +} diff --git a/docs/gha-runner-scale-set-controller/otel-tracing.md b/docs/gha-runner-scale-set-controller/otel-tracing.md new file mode 100644 index 0000000000..dd44ef5315 --- /dev/null +++ b/docs/gha-runner-scale-set-controller/otel-tracing.md @@ -0,0 +1,53 @@ +# OpenTelemetry tracing for listener job lifecycle + +The listener can export OpenTelemetry trace spans for every completed +job, alongside (or instead of) the Prometheus metrics: + +- `runner.queue` — job queued → assigned to the scale set +- `runner.startup` — assigned → runner picked the job up +- `runner.execution` — runner start → job finish + +Spans use the deterministic ID contract published by the runner +(`actions/runner` `docs/otel-id-contract.md`): + +- trace ID = `sha256("{run_id}-{run_attempt}")[:16]` +- parent span = the runner's job span, + `sha256("job-{run_id}-{run_attempt}-{job_name}")[:8]` + +so they merge into the runner's native OTLP trace with no propagation +or shared state. Spans carry the resource +`service.name=gha-listener` (override with `OTEL_SERVICE_NAME`), +`service.namespace`, and `github.scale_set.name`. + +## Enabling + +Via the `gha-runner-scale-set-controller` chart (flows to every +listener through its config secret, like the Prometheus metrics +flags): + +```yaml +otel: + listenerEndpoint: "http://otel-collector:4318" +``` + +The value is an OTLP/HTTP base URL (the same form as +`OTEL_EXPORTER_OTLP_ENDPOINT`); `/v1/traces` is appended, and TLS +follows the URL scheme. + +Alternatively, set the standard `OTEL_EXPORTER_OTLP_*` environment +variables on the listener container (e.g. through the scale set's +`listenerTemplate`, or an OTel Operator that injects them); the SDK's +spec-compliant env handling applies (endpoint, headers, TLS). + +Export is asynchronous and bounded: a slow or down collector never +delays autoscaling decisions; on overflow the newest job's spans are +dropped with a warning. + +## Known limitations + +- The scale-set message protocol does not carry `run_attempt`, so + spans always assume attempt 1. For workflow re-runs the listener + spans join the first attempt's trace. +- Configuration is controller-wide (one collector endpoint for all + scale sets). Per-scale-set configuration would need an + `AutoscalingRunnerSet` field. diff --git a/go.mod b/go.mod index 28a0a93919..33617fd98d 100644 --- a/go.mod +++ b/go.mod @@ -26,6 +26,10 @@ require ( github.com/prometheus/client_golang v1.23.2 github.com/stretchr/testify v1.11.1 github.com/teambition/rrule-go v1.8.2 + go.opentelemetry.io/otel v1.43.0 + go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.43.0 + go.opentelemetry.io/otel/sdk v1.43.0 + go.opentelemetry.io/otel/trace v1.43.0 go.uber.org/multierr v1.11.0 go.uber.org/zap v1.28.0 golang.org/x/net v0.55.0 @@ -88,6 +92,7 @@ require ( github.com/beorn7/perks v1.0.1 // indirect github.com/boombuler/barcode v1.1.0 // indirect github.com/brunoga/deep v1.2.4 // indirect + github.com/cenkalti/backoff/v5 v5.0.3 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/cloudflare/circl v1.6.3 // indirect github.com/cpuguy83/go-md2man/v2 v2.0.7 // indirect @@ -97,6 +102,7 @@ require ( github.com/fsnotify/fsnotify v1.9.0 // indirect github.com/fxamacker/cbor/v2 v2.9.0 // indirect github.com/go-errors/errors v1.5.1 // indirect + github.com/go-logr/stdr v1.2.2 // indirect github.com/go-logr/zapr v1.3.0 // indirect github.com/go-openapi/jsonpointer v0.22.5 // indirect github.com/go-openapi/jsonreference v0.21.5 // indirect @@ -127,6 +133,7 @@ require ( github.com/google/go-querystring v1.2.0 // indirect github.com/google/pprof v0.0.0-20260402051712-545e8a4df936 // indirect github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674 // indirect + github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0 // indirect github.com/gruntwork-io/go-commons v0.17.2 // indirect github.com/hashicorp/errwrap v1.1.0 // indirect github.com/hashicorp/go-cleanhttp v0.5.2 // indirect @@ -184,6 +191,10 @@ require ( github.com/xeipuuv/gojsonreference v0.0.0-20180127040603-bd5ef7bd5415 // indirect github.com/xeipuuv/gojsonschema v1.2.0 // indirect github.com/xrash/smetrics v0.0.0-20250705151800-55b8f293f342 // indirect + go.opentelemetry.io/auto/sdk v1.2.1 // indirect + go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.43.0 // indirect + go.opentelemetry.io/otel/metric v1.43.0 // indirect + go.opentelemetry.io/proto/otlp v1.10.0 // indirect go.yaml.in/yaml/v2 v2.4.4 // indirect go.yaml.in/yaml/v3 v3.0.4 // indirect golang.org/x/crypto v0.51.0 // indirect @@ -194,6 +205,9 @@ require ( golang.org/x/text v0.37.0 // indirect golang.org/x/time v0.15.0 // indirect golang.org/x/tools v0.44.0 // indirect + google.golang.org/genproto/googleapis/api v0.0.0-20260401024825-9d38bb4040a9 // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20260401024825-9d38bb4040a9 // indirect + google.golang.org/grpc v1.80.0 // indirect google.golang.org/protobuf v1.36.12-0.20260120151049-f2248ac996af // indirect gopkg.in/evanphx/json-patch.v4 v4.13.0 // indirect gopkg.in/inf.v0 v0.9.1 // indirect diff --git a/go.sum b/go.sum index ef4e3ff6ba..9198115a62 100644 --- a/go.sum +++ b/go.sum @@ -109,6 +109,8 @@ github.com/bradleyfalzon/ghinstallation/v2 v2.19.0 h1:KQfD+43pRw9NUJhGycGrFr9vF1 github.com/bradleyfalzon/ghinstallation/v2 v2.19.0/go.mod h1:fe5ECIhCdEnxwLiBlNTxx9CP455wt42BELnlDVMvaAA= github.com/brunoga/deep v1.2.4 h1:Aj9E9oUbE+ccbyh35VC/NHlzzjfIVU69BXu2mt2LmL8= github.com/brunoga/deep v1.2.4/go.mod h1:GDV6dnXqn80ezsLSZ5Wlv1PdKAWAO4L5PnKYtv2dgaI= +github.com/cenkalti/backoff/v5 v5.0.3 h1:ZN+IMa753KfX5hd8vVaMixjnqRZ3y8CuJKRKj1xcsSM= +github.com/cenkalti/backoff/v5 v5.0.3/go.mod h1:rkhZdG3JZukswDf7f0cwqPNk4K0sa+F97BxZthm/crw= github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= github.com/cloudflare/circl v1.6.3 h1:9GPOhQGF9MCYUeXyMYlqTR6a5gTrgR/fBLXvUgtVcg8= @@ -145,8 +147,11 @@ github.com/gkampitakis/go-snaps v0.5.15 h1:amyJrvM1D33cPHwVrjo9jQxX8g/7E2wYdZ+01 github.com/gkampitakis/go-snaps v0.5.15/go.mod h1:HNpx/9GoKisdhw9AFOBT1N7DBs9DiHo/hGheFGBZ+mc= github.com/go-errors/errors v1.5.1 h1:ZwEMSLRCapFLflTpT7NKaAc7ukJ8ZPEjzlxt8rPN8bk= github.com/go-errors/errors v1.5.1/go.mod h1:sIVyrIiJhuEF+Pj9Ebtd6P/rEYROXFi3BopGUQ5a5Og= +github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A= github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI= github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= +github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= +github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= github.com/go-logr/zapr v1.3.0 h1:XGdV8XW8zdwFiwOA2Dryh1gj2KRQyOOoNmBy4EplIcQ= github.com/go-logr/zapr v1.3.0/go.mod h1:YKepepNBd1u/oyhd/yQmtjVXmm9uML4IXUgMOwR8/Gg= github.com/go-openapi/jsonpointer v0.22.5 h1:8on/0Yp4uTb9f4XvTrM2+1CPrV05QPZXu+rvu2o9jcA= @@ -204,6 +209,8 @@ github.com/golang/protobuf v1.4.0-rc.2/go.mod h1:LlEzMj4AhA7rCAGe4KMBDvJI+AwstrU github.com/golang/protobuf v1.4.0-rc.4.0.20200313231945-b860323f09d0/go.mod h1:WU3c8KckQ9AFe+yFwt9sWVRKCVIyN9cPHBJSNnbL67w= github.com/golang/protobuf v1.4.0/go.mod h1:jodUvKwWbYaEsadDk5Fwe5c77LiNKVO9IDvqG2KuDX0= github.com/golang/protobuf v1.4.2/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw735rRwI= +github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= +github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps= github.com/gonvenience/bunt v1.4.3 h1:MLd8YWu1Vl1tiL+XfXJvVA9kL71yQT0N+x7gXVH9H7w= github.com/gonvenience/bunt v1.4.3/go.mod h1:ggA6odP6FNOh50mGxxytSSJTs2Ghy5Veq9wIVSbuoAw= github.com/gonvenience/idem v0.0.3 h1:rZ2f17JU5GHa3b5M5R2fClz0dYN3EFGhHHGo3AZz/1U= @@ -241,6 +248,8 @@ github.com/gorilla/mux v1.8.1 h1:TuBL49tXwgrFYWhqrNgrUNEY92u81SPhu7sTdzQEiWY= github.com/gorilla/mux v1.8.1/go.mod h1:AKf9I4AEqPTmMytcMc0KkNouC66V3BtZ4qD5fmWSiMQ= github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674 h1:JeSE6pjso5THxAzdVpqr6/geYxZytqFMBCOtn/ujyeo= github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674/go.mod h1:r4w70xmWCQKmi1ONH4KIaBptdivuRPyosB9RmPlGEwA= +github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0 h1:HWRh5R2+9EifMyIHV7ZV+MIZqgz+PMpZ14Jynv3O2Zs= +github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0/go.mod h1:JfhWUomR1baixubs02l85lZYYOm7LV6om4ceouMv45c= github.com/gruntwork-io/go-commons v0.17.2 h1:14dsCJ7M5Vv2X3BIPKeG9Kdy6vTMGhM8L4WZazxfTuY= github.com/gruntwork-io/go-commons v0.17.2/go.mod h1:zs7Q2AbUKuTarBPy19CIxJVUX/rBamfW8IwuWKniWkE= github.com/gruntwork-io/terratest v1.0.0 h1:Zk7VJ5Z9vBSwv8OQ/zzkG5D/tfqyVyjMK+lq2v+Kn/c= @@ -426,6 +435,24 @@ github.com/xeipuuv/gojsonschema v1.2.0/go.mod h1:anYRn/JVcOK2ZgGU+IjEV4nwlhoK5sQ github.com/xrash/smetrics v0.0.0-20250705151800-55b8f293f342 h1:FnBeRrxr7OU4VvAzt5X7s6266i6cSVkkFPS0TuXWbIg= github.com/xrash/smetrics v0.0.0-20250705151800-55b8f293f342/go.mod h1:Ohn+xnUBiLI6FVj/9LpzZWtj1/D6lUovWYBkxHVV3aM= github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= +go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= +go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= +go.opentelemetry.io/otel v1.43.0 h1:mYIM03dnh5zfN7HautFE4ieIig9amkNANT+xcVxAj9I= +go.opentelemetry.io/otel v1.43.0/go.mod h1:JuG+u74mvjvcm8vj8pI5XiHy1zDeoCS2LB1spIq7Ay0= +go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.43.0 h1:88Y4s2C8oTui1LGM6bTWkw0ICGcOLCAI5l6zsD1j20k= +go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.43.0/go.mod h1:Vl1/iaggsuRlrHf/hfPJPvVag77kKyvrLeD10kpMl+A= +go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.43.0 h1:3iZJKlCZufyRzPzlQhUIWVmfltrXuGyfjREgGP3UUjc= +go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.43.0/go.mod h1:/G+nUPfhq2e+qiXMGxMwumDrP5jtzU+mWN7/sjT2rak= +go.opentelemetry.io/otel/metric v1.43.0 h1:d7638QeInOnuwOONPp4JAOGfbCEpYb+K6DVWvdxGzgM= +go.opentelemetry.io/otel/metric v1.43.0/go.mod h1:RDnPtIxvqlgO8GRW18W6Z/4P462ldprJtfxHxyKd2PY= +go.opentelemetry.io/otel/sdk v1.43.0 h1:pi5mE86i5rTeLXqoF/hhiBtUNcrAGHLKQdhg4h4V9Dg= +go.opentelemetry.io/otel/sdk v1.43.0/go.mod h1:P+IkVU3iWukmiit/Yf9AWvpyRDlUeBaRg6Y+C58QHzg= +go.opentelemetry.io/otel/sdk/metric v1.43.0 h1:S88dyqXjJkuBNLeMcVPRFXpRw2fuwdvfCGLEo89fDkw= +go.opentelemetry.io/otel/sdk/metric v1.43.0/go.mod h1:C/RJtwSEJ5hzTiUz5pXF1kILHStzb9zFlIEe85bhj6A= +go.opentelemetry.io/otel/trace v1.43.0 h1:BkNrHpup+4k4w+ZZ86CZoHHEkohws8AY+WTX09nk+3A= +go.opentelemetry.io/otel/trace v1.43.0/go.mod h1:/QJhyVBUUswCphDVxq+8mld+AvhXZLhe+8WVFxiFff0= +go.opentelemetry.io/proto/otlp v1.10.0 h1:IQRWgT5srOCYfiWnpqUYz9CVmbO8bFmKcwYxpuCSL2g= +go.opentelemetry.io/proto/otlp v1.10.0/go.mod h1:/CV4QoCR/S9yaPj8utp3lvQPoqMtxXdzn7ozvvozVqk= go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0= @@ -494,6 +521,14 @@ golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8T golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= gomodules.xyz/jsonpatch/v2 v2.5.0 h1:JELs8RLM12qJGXU4u/TO3V25KW8GreMKl9pdkk14RM0= gomodules.xyz/jsonpatch/v2 v2.5.0/go.mod h1:AH3dM2RI6uoBZxn3LVrfvJ3E0/9dG4cSrbuBJT4moAY= +gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4= +gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E= +google.golang.org/genproto/googleapis/api v0.0.0-20260401024825-9d38bb4040a9 h1:VPWxll4HlMw1Vs/qXtN7BvhZqsS9cdAittCNvVENElA= +google.golang.org/genproto/googleapis/api v0.0.0-20260401024825-9d38bb4040a9/go.mod h1:7QBABkRtR8z+TEnmXTqIqwJLlzrZKVfAUm7tY3yGv0M= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260401024825-9d38bb4040a9 h1:m8qni9SQFH0tJc1X0vmnpw/0t+AImlSvp30sEupozUg= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260401024825-9d38bb4040a9/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8= +google.golang.org/grpc v1.80.0 h1:Xr6m2WmWZLETvUNvIUmeD5OAagMw3FiKmMlTdViWsHM= +google.golang.org/grpc v1.80.0/go.mod h1:ho/dLnxwi3EDJA4Zghp7k2Ec1+c2jqup0bFkw07bwF4= google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8= google.golang.org/protobuf v0.0.0-20200221191635-4d8936d0db64/go.mod h1:kwYJMbMJ01Woi6D6+Kah6886xMZcty6N08ah7+eCXa0= google.golang.org/protobuf v0.0.0-20200228230310-ab0ca4ff8a60/go.mod h1:cfTl7dwQJ+fmap5saPgwCLgHXTUD7jkjRqWcaiX5VyM= diff --git a/main.go b/main.go index 80c047e63f..9903b9805d 100644 --- a/main.go +++ b/main.go @@ -84,6 +84,7 @@ func main() { // metrics server configuration for AutoscalingListener listenerMetricsAddr string listenerMetricsEndpoint string + listenerOTelEndpoint string metricsAddr string pprofAddr string @@ -126,6 +127,7 @@ func main() { flag.StringVar(&listenerMetricsAddr, "listener-metrics-addr", ":8080", "The address applied to AutoscalingListener metrics server") flag.StringVar(&listenerMetricsEndpoint, "listener-metrics-endpoint", "/metrics", "The AutoscalingListener metrics server endpoint from which the metrics are collected") + flag.StringVar(&listenerOTelEndpoint, "listener-otel-endpoint", "", "The OTLP/HTTP collector base URL (e.g. http://otel-collector:4318) for AutoscalingListener job lifecycle trace export. Empty disables it.") flag.StringVar(&metricsAddr, "metrics-addr", ":8080", "The address the metric endpoint binds to.") flag.StringVar(&pprofAddr, "pprof-addr", "", "The address the pprof endpoint binds to.") flag.StringVar(&probeAddr, "health-probe-bind-address", "", "The address the health probe endpoint binds to. Disabled if empty.") @@ -363,6 +365,7 @@ func main() { Scheme: mgr.GetScheme(), ListenerMetricsAddr: listenerMetricsAddr, ListenerMetricsEndpoint: listenerMetricsEndpoint, + ListenerOTelEndpoint: listenerOTelEndpoint, ResourceBuilder: rb, }).SetupWithManager(mgr, controllerOpts...); err != nil { log.Error(err, "unable to create controller", "controller", "AutoscalingListener")