From b52d09c2385f958dcb42f856d7d8ee32ed4ad8da Mon Sep 17 00:00:00 2001 From: Stefan Penner Date: Tue, 21 Apr 2026 13:32:28 -0600 Subject: [PATCH 01/14] Add optional OpenTelemetry trace export for job lifecycle spans MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds an OTel trace recorder that implements the existing MetricsRecorder interface. When configured with an OTLP endpoint, the listener emits three child spans per completed job: - runner.queue: QueueTime → ScaleSetAssignTime - runner.startup: ScaleSetAssignTime → RunnerAssignTime - runner.execution: RunnerAssignTime → FinishTime Spans use deterministic trace/span IDs (MD5 of runID-attempt, big-endian jobID) compatible with tools that reconstruct GitHub Actions workflows as OpenTelemetry traces. Configuration: set otel_endpoint (and optionally otel_insecure) in the listener config JSON, or pass via Helm values. When no endpoint is configured, behavior is unchanged. Co-Authored-By: Claude Opus 4.6 --- cmd/ghalistener/config/config.go | 2 + cmd/ghalistener/main.go | 42 ++++- cmd/ghalistener/metrics/composite.go | 44 +++++ cmd/ghalistener/metrics/otel.go | 201 ++++++++++++++++++++++ cmd/ghalistener/metrics/otel_test.go | 239 +++++++++++++++++++++++++++ go.mod | 14 ++ go.sum | 35 ++++ 7 files changed, 573 insertions(+), 4 deletions(-) create mode 100644 cmd/ghalistener/metrics/composite.go create mode 100644 cmd/ghalistener/metrics/otel.go create mode 100644 cmd/ghalistener/metrics/otel_test.go diff --git a/cmd/ghalistener/config/config.go b/cmd/ghalistener/config/config.go index 69cfafb8ee..f4b1ecfb2a 100644 --- a/cmd/ghalistener/config/config.go +++ b/cmd/ghalistener/config/config.go @@ -44,6 +44,8 @@ type Config struct { MetricsAddr string `json:"metrics_addr"` MetricsEndpoint string `json:"metrics_endpoint"` Metrics *v1alpha1.MetricsConfig `json:"metrics"` + OTelEndpoint string `json:"otel_endpoint"` + OTelInsecure bool `json:"otel_insecure"` } func Read(ctx context.Context, configPath string) (*Config, error) { diff --git a/cmd/ghalistener/main.go b/cmd/ghalistener/main.go index 7dc4a17f46..192c84921d 100644 --- a/cmd/ghalistener/main.go +++ b/cmd/ghalistener/main.go @@ -14,6 +14,7 @@ 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" "golang.org/x/sync/errgroup" ) @@ -65,6 +66,27 @@ func run(ctx context.Context, config *config.Config) error { }) } + // OTel trace recorder (optional) + var otelRecorder *metrics.OTelRecorder + if config.OTelEndpoint != "" { + opts := []otlptracehttp.Option{ + otlptracehttp.WithEndpoint(config.OTelEndpoint), + } + if config.OTelInsecure { + opts = append(opts, otlptracehttp.WithInsecure()) + } + otlpExporter, err := otlptracehttp.New(ctx, opts...) + if err != nil { + return fmt.Errorf("failed to create OTel exporter: %w", err) + } + otelRecorder = metrics.NewOTelRecorder( + otlpExporter, + logger.With("component", "otel recorder"), + ) + defer otelRecorder.Shutdown(ctx) + logger.Info("OTel trace recorder enabled", "endpoint", config.OTelEndpoint) + } + hostname, err := os.Hostname() if err != nil { hostname = uuid.NewString() @@ -91,13 +113,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/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..ede2b9566d --- /dev/null +++ b/cmd/ghalistener/metrics/otel.go @@ -0,0 +1,201 @@ +package metrics + +import ( + "context" + "crypto/md5" + "encoding/binary" + "fmt" + "log/slog" + "strconv" + "sync" + "time" + + "github.com/actions/scaleset" + "github.com/actions/scaleset/listener" + "go.opentelemetry.io/otel/attribute" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + "go.opentelemetry.io/otel/sdk/trace/tracetest" + "go.opentelemetry.io/otel/trace" +) + +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 (MD5 of runID-attempt for trace ID, +// big-endian int64 for span ID) is compatible with otel-explorer's +// GitHub Actions trace view, allowing ARC spans to merge into existing +// workflow traces with zero correlation configuration. +type OTelRecorder struct { + mu sync.Mutex + exporter sdktrace.SpanExporter + logger *slog.Logger + + // runAttempt defaults to 1. ARC messages don't include + // run_attempt; override via SetRunAttempt if determinable. + runAttempt int64 +} + +// NewOTelRecorder creates a recorder that exports job lifecycle spans +// via the given SpanExporter. +func NewOTelRecorder(exporter sdktrace.SpanExporter, logger *slog.Logger) *OTelRecorder { + if logger == nil { + logger = slog.New(slog.DiscardHandler) + } + return &OTelRecorder{ + exporter: exporter, + logger: logger, + runAttempt: 1, + } +} + +// SetRunAttempt overrides the default run attempt (1) used for trace +// ID generation. +func (r *OTelRecorder) SetRunAttempt(attempt int64) { + r.mu.Lock() + defer r.mu.Unlock() + r.runAttempt = attempt +} + +// Shutdown flushes pending spans and releases exporter resources. +func (r *OTelRecorder) Shutdown(ctx context.Context) error { + return r.exporter.Shutdown(ctx) +} + +func (r *OTelRecorder) RecordJobStarted(_ *scaleset.JobStarted) {} + +func (r *OTelRecorder) RecordJobCompleted(msg *scaleset.JobCompleted) { + r.mu.Lock() + attempt := r.runAttempt + r.mu.Unlock() + + spans := buildJobSpans(msg, attempt) + if len(spans) == 0 { + return + } + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := r.exporter.ExportSpans(ctx, spans); err != nil { + r.logger.Warn("failed to export OTel spans", "error", err) + } +} + +func (r *OTelRecorder) RecordStatistics(_ *scaleset.RunnerScaleSetStatistic) {} +func (r *OTelRecorder) RecordDesiredRunners(_ int) {} + +func buildJobSpans(msg *scaleset.JobCompleted, attempt int64) []sdktrace.ReadOnlySpan { + traceID := newTraceID(msg.WorkflowRunID, attempt) + parentSpanID := toSpanID(msg.JobID) + parentSC := trace.NewSpanContext(trace.SpanContextConfig{ + TraceID: traceID, + SpanID: parentSpanID, + TraceFlags: trace.FlagsSampled, + }) + + commonAttrs := []attribute.KeyValue{ + attribute.Int64("github.run_id", msg.WorkflowRunID), + attribute.String("github.job_id", msg.JobID), + attribute.String("github.job_name", msg.JobDisplayName), + attribute.String("github.repository", msg.OwnerName+"/"+msg.RepositoryName), + attribute.String("github.runner_name", msg.RunnerName), + attribute.Int("github.runner_id", msg.RunnerID), + attribute.String("github.workflow_ref", msg.JobWorkflowRef), + attribute.String("github.event_name", msg.EventName), + } + + var stubs tracetest.SpanStubs + + if !msg.QueueTime.IsZero() && !msg.ScaleSetAssignTime.IsZero() { + stubs = append(stubs, newStub( + traceID, parentSC, "runner.queue", + msg.QueueTime, msg.ScaleSetAssignTime, + append(sliceClone(commonAttrs), attribute.String("type", "runner.queue")), + )) + } + + if !msg.ScaleSetAssignTime.IsZero() && !msg.RunnerAssignTime.IsZero() { + stubs = append(stubs, newStub( + traceID, parentSC, "runner.startup", + msg.ScaleSetAssignTime, msg.RunnerAssignTime, + append(sliceClone(commonAttrs), attribute.String("type", "runner.startup")), + )) + } + + if !msg.RunnerAssignTime.IsZero() && !msg.FinishTime.IsZero() { + stubs = append(stubs, newStub( + traceID, parentSC, "runner.execution", + msg.RunnerAssignTime, msg.FinishTime, + append(sliceClone(commonAttrs), + attribute.String("type", "runner.execution"), + attribute.String("github.conclusion", msg.Result), + ), + )) + } + + return stubs.Snapshots() +} + +func 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, + } +} + +// Deterministic ID generation — compatible with otel-explorer's +// pkg/githubapi/ids.go scheme. + +func newTraceID(runID, runAttempt int64) trace.TraceID { + if runAttempt == 0 { + runAttempt = 1 + } + return trace.TraceID(md5.Sum([]byte(fmt.Sprintf("%d-%d", runID, runAttempt)))) +} + +func newSpanID(id int64) trace.SpanID { + var sid trace.SpanID + binary.BigEndian.PutUint64(sid[:], uint64(id)) + return sid +} + +func newSpanIDFromString(s string) trace.SpanID { + sum := md5.Sum([]byte(s)) + var sid trace.SpanID + copy(sid[:], sum[:8]) + return sid +} + +func toSpanID(jobID string) trace.SpanID { + if id, err := strconv.ParseInt(jobID, 10, 64); err == nil { + return newSpanID(id) + } + return newSpanIDFromString(jobID) +} + +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..07557292d8 --- /dev/null +++ b/cmd/ghalistener/metrics/otel_test.go @@ -0,0 +1,239 @@ +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 TestOTelRecorder_EmitsThreeSpans(t *testing.T) { + exp := &captureExporter{} + rec := NewOTelRecorder(exp, nil) + + 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) + + 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()) + } + + expectedParent := newSpanID(42) + for _, s := range spans { + assert.Equal(t, expectedParent, s.Parent().SpanID()) + } + + q := byName["runner.queue"] + require.NotNil(t, q) + assert.Equal(t, now, q.StartTime()) + assert.Equal(t, now.Add(10*time.Second), q.EndTime()) + assertAttr(t, q, "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, "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, "type", "runner.execution") + assertAttr(t, e, "github.conclusion", "Succeeded") +} + +func TestOTelRecorder_SkipsMissingTimestamps(t *testing.T) { + exp := &captureExporter{} + rec := NewOTelRecorder(exp, nil) + + 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) + + 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 := NewOTelRecorder(exp, nil) + + 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) + + spans := exp.Spans() + require.Len(t, spans, 1) + s := spans[0] + + assertAttr(t, s, "github.job_name", "test-suite") + assertAttr(t, s, "github.repository", "org/repo") + assertAttr(t, s, "github.runner_name", "runner-3") + assertAttr(t, s, "github.conclusion", "Failed") +} + +func TestOTelRecorder_SetRunAttempt(t *testing.T) { + exp := &captureExporter{} + rec := NewOTelRecorder(exp, nil) + rec.SetRunAttempt(3) + + now := time.Now() + msg := &scaleset.JobCompleted{ + Result: "Succeeded", + JobMessageBase: scaleset.JobMessageBase{ + WorkflowRunID: 77777, + JobID: "1", + RunnerAssignTime: now, + FinishTime: now.Add(time.Second), + }, + } + + rec.RecordJobCompleted(msg) + + spans := exp.Spans() + require.Len(t, spans, 1) + assert.Equal(t, newTraceID(77777, 3), spans[0].SpanContext().TraceID()) +} + +func TestOTelRecorder_JobStartedIsNoOp(t *testing.T) { + exp := &captureExporter{} + rec := NewOTelRecorder(exp, nil) + rec.RecordJobStarted(&scaleset.JobStarted{}) + assert.Empty(t, exp.Spans()) +} + +func TestOTelRecorder_StatisticsIsNoOp(t *testing.T) { + exp := &captureExporter{} + rec := NewOTelRecorder(exp, nil) + rec.RecordStatistics(&scaleset.RunnerScaleSetStatistic{TotalRunningJobs: 5}) + assert.Empty(t, exp.Spans()) +} + +func TestCompositeRecorder_DelegatesAll(t *testing.T) { + exp1 := &captureExporter{} + exp2 := &captureExporter{} + r1 := NewOTelRecorder(exp1, nil) + r2 := NewOTelRecorder(exp2, nil) + 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) + + assert.Len(t, exp1.Spans(), 1) + assert.Len(t, exp2.Spans(), 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 := newSpanID(42) + sid2 := newSpanID(42) + sid3 := newSpanID(43) + assert.Equal(t, sid1, sid2) + assert.NotEqual(t, sid1, sid3) +} + +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()) +} diff --git a/go.mod b/go.mod index 28a0a93919..429c0db968 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.35.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/v4 v4.3.0 // 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.27.7 // 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.40.0 // indirect + go.opentelemetry.io/otel/metric v1.43.0 // indirect + go.opentelemetry.io/proto/otlp v1.9.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..71d1fbc59d 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/v4 v4.3.0 h1:MyRJ/UdXutAwSAT+s3wNd7MfTIcy71VQueUuFK343L8= +github.com/cenkalti/backoff/v4 v4.3.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE= 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.27.7 h1:X+2YciYSxvMQK0UZ7sg45ZVabVZBeBuvMkmuI2V3Fak= +github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.7/go.mod h1:lW34nIZuQ8UDPdkon5fmfp2l3+ZkQ2me/+oecHYLOII= 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.40.0 h1:QKdN8ly8zEMrByybbQgv8cWBcdAarwmIPZ6FThrWXJs= +go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.40.0/go.mod h1:bTdK1nhqF76qiPoCCdyFIV+N/sRHYXYCTQc+3VCi3MI= +go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.35.0 h1:xJ2qHD0C1BeYVTLLR9sX12+Qb95kfeD/byKj6Ky1pXg= +go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.35.0/go.mod h1:u5BF1xyjstDowA1R5QAO9JHzqK+ublenEW/dyqTjBVk= +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.9.0 h1:l706jCMITVouPOqEnii2fIAuO3IVGBRPV5ICjceRb/A= +go.opentelemetry.io/proto/otlp v1.9.0/go.mod h1:xE+Cx5E/eEHw+ISFkwPLwCZefwVjY+pqKg1qcK03+/4= 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= From d176d1da7c828aa721f62e2ae2d07c76938d6941 Mon Sep 17 00:00:00 2001 From: Stefan Penner Date: Tue, 21 Apr 2026 14:56:24 -0600 Subject: [PATCH 02/14] Add CI/CD semconv attrs, normalize conclusion values, env var config - Add OTel CI/CD semantic convention attributes (cicd.pipeline.*, cicd.worker.*, vcs.repository.url.full) alongside github.* attrs - Normalize conclusion values from ARC variants (Succeeded, Failed) to standard GitHub API values (success, failure, cancelled) - Map to cicd.pipeline.task.run.result using OTel semconv result values (cancellation, skip, timeout, error) - Support OTEL_EXPORTER_OTLP_ENDPOINT and OTEL_EXPORTER_OTLP_INSECURE env vars as config fallback for listener template injection Co-Authored-By: Claude Opus 4.6 --- cmd/ghalistener/config/config.go | 7 ++++ cmd/ghalistener/metrics/otel.go | 55 +++++++++++++++++++++++++++- cmd/ghalistener/metrics/otel_test.go | 16 +++++++- 3 files changed, 74 insertions(+), 4 deletions(-) diff --git a/cmd/ghalistener/config/config.go b/cmd/ghalistener/config/config.go index f4b1ecfb2a..ba4bdb84e5 100644 --- a/cmd/ghalistener/config/config.go +++ b/cmd/ghalistener/config/config.go @@ -60,6 +60,13 @@ func Read(ctx context.Context, configPath string) (*Config, error) { return nil, fmt.Errorf("failed to decode config: %w", err) } + if v := os.Getenv("OTEL_EXPORTER_OTLP_ENDPOINT"); v != "" && config.OTelEndpoint == "" { + config.OTelEndpoint = v + } + if os.Getenv("OTEL_EXPORTER_OTLP_INSECURE") == "true" && !config.OTelInsecure { + config.OTelInsecure = true + } + var vault vault.Vault switch config.VaultType { case "": diff --git a/cmd/ghalistener/metrics/otel.go b/cmd/ghalistener/metrics/otel.go index ede2b9566d..7244c2a50e 100644 --- a/cmd/ghalistener/metrics/otel.go +++ b/cmd/ghalistener/metrics/otel.go @@ -7,6 +7,7 @@ import ( "fmt" "log/slog" "strconv" + "strings" "sync" "time" @@ -99,15 +100,26 @@ func buildJobSpans(msg *scaleset.JobCompleted, attempt int64) []sdktrace.ReadOnl TraceFlags: trace.FlagsSampled, }) + conclusion := normalizeConclusion(msg.Result) + repo := msg.OwnerName + "/" + msg.RepositoryName + commonAttrs := []attribute.KeyValue{ + // GitHub Actions attributes attribute.Int64("github.run_id", msg.WorkflowRunID), attribute.String("github.job_id", msg.JobID), attribute.String("github.job_name", msg.JobDisplayName), - attribute.String("github.repository", msg.OwnerName+"/"+msg.RepositoryName), + attribute.String("github.repository", repo), attribute.String("github.runner_name", msg.RunnerName), attribute.Int("github.runner_id", msg.RunnerID), attribute.String("github.workflow_ref", msg.JobWorkflowRef), attribute.String("github.event_name", msg.EventName), + // OTel CI/CD semantic conventions (cicd.*) + attribute.String("cicd.pipeline.run.id", strconv.FormatInt(msg.WorkflowRunID, 10)), + attribute.String("cicd.pipeline.task.name", msg.JobDisplayName), + attribute.String("cicd.pipeline.task.run.id", msg.JobID), + attribute.String("cicd.worker.name", msg.RunnerName), + attribute.String("cicd.worker.id", strconv.Itoa(msg.RunnerID)), + attribute.String("vcs.repository.url.full", "https://github.com/"+repo), } var stubs tracetest.SpanStubs @@ -134,7 +146,8 @@ func buildJobSpans(msg *scaleset.JobCompleted, attempt int64) []sdktrace.ReadOnl msg.RunnerAssignTime, msg.FinishTime, append(sliceClone(commonAttrs), attribute.String("type", "runner.execution"), - attribute.String("github.conclusion", msg.Result), + attribute.String("github.conclusion", conclusion), + attribute.String("cicd.pipeline.task.run.result", conclusionToSemconv(conclusion)), ), )) } @@ -142,6 +155,44 @@ func buildJobSpans(msg *scaleset.JobCompleted, attempt int64) []sdktrace.ReadOnl return stubs.Snapshots() } +func normalizeConclusion(raw string) string { + switch strings.ToLower(raw) { + case "success", "succeeded": + return "success" + case "failure", "failed": + 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 strings.ToLower(raw) + } +} + +func conclusionToSemconv(conclusion string) string { + switch conclusion { + case "success": + return "success" + case "failure": + return "failure" + case "cancelled": + return "cancellation" + case "skipped": + return "skip" + case "timed_out": + return "timeout" + case "startup_failure": + return "error" + default: + return conclusion + } +} + func newStub( traceID trace.TraceID, parentSC trace.SpanContext, diff --git a/cmd/ghalistener/metrics/otel_test.go b/cmd/ghalistener/metrics/otel_test.go index 07557292d8..cebd21189d 100644 --- a/cmd/ghalistener/metrics/otel_test.go +++ b/cmd/ghalistener/metrics/otel_test.go @@ -95,7 +95,18 @@ func TestOTelRecorder_EmitsThreeSpans(t *testing.T) { assert.Equal(t, now.Add(40*time.Second), e.StartTime()) assert.Equal(t, now.Add(5*time.Minute), e.EndTime()) assertAttr(t, e, "type", "runner.execution") - assertAttr(t, e, "github.conclusion", "Succeeded") + assertAttr(t, e, "github.conclusion", "success") + assertAttr(t, e, "cicd.pipeline.task.run.result", "success") + + // Verify CI/CD semantic convention attributes on all spans + for _, span := range byName { + assertAttr(t, span, "cicd.pipeline.run.id", "99999") + assertAttr(t, span, "cicd.pipeline.task.name", "build") + assertAttr(t, span, "cicd.pipeline.task.run.id", "42") + 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") + } } func TestOTelRecorder_SkipsMissingTimestamps(t *testing.T) { @@ -150,7 +161,8 @@ func TestOTelRecorder_CommonAttributes(t *testing.T) { assertAttr(t, s, "github.job_name", "test-suite") assertAttr(t, s, "github.repository", "org/repo") assertAttr(t, s, "github.runner_name", "runner-3") - assertAttr(t, s, "github.conclusion", "Failed") + assertAttr(t, s, "github.conclusion", "failure") + assertAttr(t, s, "cicd.pipeline.task.run.result", "failure") } func TestOTelRecorder_SetRunAttempt(t *testing.T) { From 914cf584bc6f84201025d5cb197f684ed47ab3a8 Mon Sep 17 00:00:00 2001 From: Stefan Penner Date: Mon, 22 Jun 2026 09:17:11 -0600 Subject: [PATCH 03/14] Ignore locally built ghalistener binary Co-Authored-By: Claude Opus 4.8 (1M context) --- .gitignore | 3 +++ 1 file changed, 3 insertions(+) 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 From 7e55526b564e46c14da280ab7393f7cb1aae379e Mon Sep 17 00:00:00 2001 From: Stefan Penner Date: Wed, 1 Jul 2026 16:13:17 -0600 Subject: [PATCH 04/14] Align OTel Go module versions at v1.43.0 The otlptracehttp exporter was pinned at v1.35.0 while otel/sdk/trace were at v1.43.0. Mixed minors compile but miss exporter fixes (incl. endpoint/env handling) and would churn under dependabot immediately. Co-Authored-By: Claude Fable 5 --- go.mod | 10 +++++----- go.sum | 20 ++++++++++---------- 2 files changed, 15 insertions(+), 15 deletions(-) diff --git a/go.mod b/go.mod index 429c0db968..33617fd98d 100644 --- a/go.mod +++ b/go.mod @@ -27,7 +27,7 @@ require ( 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.35.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 @@ -92,7 +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/v4 v4.3.0 // 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 @@ -133,7 +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.27.7 // 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 @@ -192,9 +192,9 @@ require ( 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.40.0 // 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.9.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 diff --git a/go.sum b/go.sum index 71d1fbc59d..9198115a62 100644 --- a/go.sum +++ b/go.sum @@ -109,8 +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/v4 v4.3.0 h1:MyRJ/UdXutAwSAT+s3wNd7MfTIcy71VQueUuFK343L8= -github.com/cenkalti/backoff/v4 v4.3.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE= +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= @@ -248,8 +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.27.7 h1:X+2YciYSxvMQK0UZ7sg45ZVabVZBeBuvMkmuI2V3Fak= -github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.7/go.mod h1:lW34nIZuQ8UDPdkon5fmfp2l3+ZkQ2me/+oecHYLOII= +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= @@ -439,10 +439,10 @@ go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ 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.40.0 h1:QKdN8ly8zEMrByybbQgv8cWBcdAarwmIPZ6FThrWXJs= -go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.40.0/go.mod h1:bTdK1nhqF76qiPoCCdyFIV+N/sRHYXYCTQc+3VCi3MI= -go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.35.0 h1:xJ2qHD0C1BeYVTLLR9sX12+Qb95kfeD/byKj6Ky1pXg= -go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.35.0/go.mod h1:u5BF1xyjstDowA1R5QAO9JHzqK+ublenEW/dyqTjBVk= +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= @@ -451,8 +451,8 @@ go.opentelemetry.io/otel/sdk/metric v1.43.0 h1:S88dyqXjJkuBNLeMcVPRFXpRw2fuwdvfC 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.9.0 h1:l706jCMITVouPOqEnii2fIAuO3IVGBRPV5ICjceRb/A= -go.opentelemetry.io/proto/otlp v1.9.0/go.mod h1:xE+Cx5E/eEHw+ISFkwPLwCZefwVjY+pqKg1qcK03+/4= +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= From a6d217b9541cedbd4f3cd7e22c76a1b4720a1f9b Mon Sep 17 00:00:00 2001 From: Stefan Penner Date: Wed, 1 Jul 2026 16:14:57 -0600 Subject: [PATCH 05/14] Conform trace/span IDs to the runner's OTel ID contract MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Trace ID: sha256("{run_id}-{run_attempt}")[:16] (was MD5, which also hard-fails under FIPS-140-only Go). - Parent the listener spans to the contract job span, sha256("job-{run_id}-{run_attempt}-{job_name}")[:8], using the job display name — the span the runner actually emits. Previously they parented to toSpanID(jobId GUID), an ID nothing in the stack emits, so they rendered as orphans and never nested under the job span. - Add the contract's golden conformance vectors as tests (99999/1 -> acad1e2a107636235fd56bb742499bd0, job 'build' -> 81606d47848a59c0), shared with the runner's L0 suite. Normative contract: actions/runner docs/otel-id-contract.md. Co-Authored-By: Claude Fable 5 --- cmd/ghalistener/metrics/otel.go | 48 ++++++++++++++++------------ cmd/ghalistener/metrics/otel_test.go | 26 ++++++++++++--- 2 files changed, 49 insertions(+), 25 deletions(-) diff --git a/cmd/ghalistener/metrics/otel.go b/cmd/ghalistener/metrics/otel.go index 7244c2a50e..e01c0a9503 100644 --- a/cmd/ghalistener/metrics/otel.go +++ b/cmd/ghalistener/metrics/otel.go @@ -2,8 +2,7 @@ package metrics import ( "context" - "crypto/md5" - "encoding/binary" + "crypto/sha256" "fmt" "log/slog" "strconv" @@ -29,10 +28,15 @@ var _ listener.MetricsRecorder = &OTelRecorder{} // - runner.startup: ScaleSetAssignTime → RunnerAssignTime // - runner.execution: RunnerAssignTime → FinishTime // -// The deterministic ID scheme (MD5 of runID-attempt for trace ID, -// big-endian int64 for span ID) is compatible with otel-explorer's -// GitHub Actions trace view, allowing ARC spans to merge into existing -// workflow traces with zero correlation configuration. +// 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. type OTelRecorder struct { mu sync.Mutex exporter sdktrace.SpanExporter @@ -93,7 +97,10 @@ func (r *OTelRecorder) RecordDesiredRunners(_ int) {} func buildJobSpans(msg *scaleset.JobCompleted, attempt int64) []sdktrace.ReadOnlySpan { traceID := newTraceID(msg.WorkflowRunID, attempt) - parentSpanID := toSpanID(msg.JobID) + // 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, @@ -215,34 +222,33 @@ func newStub( } } -// Deterministic ID generation — compatible with otel-explorer's -// pkg/githubapi/ids.go scheme. +// 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 } - return trace.TraceID(md5.Sum([]byte(fmt.Sprintf("%d-%d", runID, runAttempt)))) -} - -func newSpanID(id int64) trace.SpanID { - var sid trace.SpanID - binary.BigEndian.PutUint64(sid[:], uint64(id)) - return sid + 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 := md5.Sum([]byte(s)) + sum := sha256.Sum256([]byte(s)) var sid trace.SpanID copy(sid[:], sum[:8]) return sid } -func toSpanID(jobID string) trace.SpanID { - if id, err := strconv.ParseInt(jobID, 10, 64); err == nil { - return newSpanID(id) +func newJobSpanID(runID, runAttempt int64, jobName string) trace.SpanID { + if runAttempt == 0 { + runAttempt = 1 } - return newSpanIDFromString(jobID) + return newSpanIDFromString(fmt.Sprintf("job-%d-%d-%s", runID, runAttempt, jobName)) } func sliceClone(s []attribute.KeyValue) []attribute.KeyValue { diff --git a/cmd/ghalistener/metrics/otel_test.go b/cmd/ghalistener/metrics/otel_test.go index cebd21189d..e6789e296d 100644 --- a/cmd/ghalistener/metrics/otel_test.go +++ b/cmd/ghalistener/metrics/otel_test.go @@ -73,10 +73,13 @@ func TestOTelRecorder_EmitsThreeSpans(t *testing.T) { assert.Equal(t, expectedTraceID, s.SpanContext().TraceID()) } - expectedParent := newSpanID(42) + // 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) @@ -225,6 +228,21 @@ func TestCompositeRecorder_DelegatesAll(t *testing.T) { 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) @@ -232,9 +250,9 @@ func TestDeterministicIDs(t *testing.T) { 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 := newSpanID(42) - sid2 := newSpanID(42) - sid3 := newSpanID(43) + 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) } From 058a64888335de390df48caf635546ace0400fe9 Mon Sep 17 00:00:00 2001 From: Stefan Penner Date: Wed, 1 Jul 2026 16:19:03 -0600 Subject: [PATCH 06/14] Align span attributes with the runner exporter; add Resource and scope Attribute alignment (names, types, values match Runner.Worker/OTelTraceExporter.cs so one query spans both sides): - github.run_id is now a string (runner emits the raw string form; TraceQL attribute matching is type-sensitive) - add github.run_attempt (string, matches runner) - cicd.pipeline.task.run.id now carries the contract job span ID hex, as the runner does (was the scale-set job GUID) - add cicd.pipeline.run.url.full and vcs.provider.name - vcs.repository.url.full / run URL derive from the configured GitHub server URL (GHES-safe) instead of hardcoded https://github.com - drop github.job_id/job_name (runner never emits them; display name lives in cicd.pipeline.task.name) and github.runner_name/runner_id (bespoke dupes of semconv cicd.worker.*) Spans now carry a proper OTel Resource (service.name=gha-listener, service.version, service.namespace, github.scale_set.name, cicd.system.component=controller; OTEL_SERVICE_NAME / OTEL_RESOURCE_ATTRIBUTES env override) and an InstrumentationScope. Previously they exported with an empty resource, invisible to every service-scoped query. NewOTelRecorder now takes an OTelRecorderConfig, mirroring metrics.ExporterConfig. Co-Authored-By: Claude Fable 5 --- cmd/ghalistener/main.go | 11 +- cmd/ghalistener/metrics/otel.go | 121 +++++++++++++++++----- cmd/ghalistener/metrics/otel_test.go | 149 ++++++++++++++++++++++++--- 3 files changed, 239 insertions(+), 42 deletions(-) diff --git a/cmd/ghalistener/main.go b/cmd/ghalistener/main.go index 192c84921d..da910139d3 100644 --- a/cmd/ghalistener/main.go +++ b/cmd/ghalistener/main.go @@ -79,10 +79,13 @@ func run(ctx context.Context, config *config.Config) error { if err != nil { return fmt.Errorf("failed to create OTel exporter: %w", err) } - otelRecorder = metrics.NewOTelRecorder( - otlpExporter, - logger.With("component", "otel recorder"), - ) + 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, + }) defer otelRecorder.Shutdown(ctx) logger.Info("OTel trace recorder enabled", "endpoint", config.OTelEndpoint) } diff --git a/cmd/ghalistener/metrics/otel.go b/cmd/ghalistener/metrics/otel.go index e01c0a9503..95ca4e1d86 100644 --- a/cmd/ghalistener/metrics/otel.go +++ b/cmd/ghalistener/metrics/otel.go @@ -10,14 +10,19 @@ import ( "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 @@ -38,24 +43,81 @@ var _ listener.MetricsRecorder = &OTelRecorder{} // exporter emits the job span with these exact IDs, so ARC spans merge // under it with zero correlation configuration. type OTelRecorder struct { - mu sync.Mutex - exporter sdktrace.SpanExporter - logger *slog.Logger + mu sync.Mutex + exporter sdktrace.SpanExporter + logger *slog.Logger + serverURL string + res *resource.Resource + scope instrumentation.Scope // runAttempt defaults to 1. ARC messages don't include // run_attempt; override via SetRunAttempt if determinable. runAttempt int64 } +// 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(exporter sdktrace.SpanExporter, logger *slog.Logger) *OTelRecorder { +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() + } + return &OTelRecorder{ - exporter: exporter, - logger: logger, + exporter: cfg.Exporter, + logger: logger, + serverURL: serverURL, + res: res, + scope: instrumentation.Scope{ + Name: otelScopeName, + Version: build.Version, + }, runAttempt: 1, } } @@ -80,7 +142,7 @@ func (r *OTelRecorder) RecordJobCompleted(msg *scaleset.JobCompleted) { attempt := r.runAttempt r.mu.Unlock() - spans := buildJobSpans(msg, attempt) + spans := r.buildJobSpans(msg, attempt) if len(spans) == 0 { return } @@ -95,7 +157,7 @@ func (r *OTelRecorder) RecordJobCompleted(msg *scaleset.JobCompleted) { func (r *OTelRecorder) RecordStatistics(_ *scaleset.RunnerScaleSetStatistic) {} func (r *OTelRecorder) RecordDesiredRunners(_ int) {} -func buildJobSpans(msg *scaleset.JobCompleted, attempt int64) []sdktrace.ReadOnlySpan { +func (r *OTelRecorder) buildJobSpans(msg *scaleset.JobCompleted, attempt int64) []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 @@ -109,30 +171,35 @@ func buildJobSpans(msg *scaleset.JobCompleted, attempt int64) []sdktrace.ReadOnl 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 - attribute.Int64("github.run_id", msg.WorkflowRunID), - attribute.String("github.job_id", msg.JobID), - attribute.String("github.job_name", msg.JobDisplayName), + // 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.runner_name", msg.RunnerName), - attribute.Int("github.runner_id", msg.RunnerID), attribute.String("github.workflow_ref", msg.JobWorkflowRef), attribute.String("github.event_name", msg.EventName), - // OTel CI/CD semantic conventions (cicd.*) - attribute.String("cicd.pipeline.run.id", strconv.FormatInt(msg.WorkflowRunID, 10)), + // 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", msg.JobID), + attribute.String("cicd.pipeline.task.run.id", parentSpanID.String()), attribute.String("cicd.worker.name", msg.RunnerName), attribute.String("cicd.worker.id", strconv.Itoa(msg.RunnerID)), - attribute.String("vcs.repository.url.full", "https://github.com/"+repo), + attribute.String("vcs.repository.url.full", r.serverURL+"/"+repo), + attribute.String("vcs.provider.name", "github"), } var stubs tracetest.SpanStubs if !msg.QueueTime.IsZero() && !msg.ScaleSetAssignTime.IsZero() { - stubs = append(stubs, newStub( + stubs = append(stubs, r.newStub( traceID, parentSC, "runner.queue", msg.QueueTime, msg.ScaleSetAssignTime, append(sliceClone(commonAttrs), attribute.String("type", "runner.queue")), @@ -140,7 +207,7 @@ func buildJobSpans(msg *scaleset.JobCompleted, attempt int64) []sdktrace.ReadOnl } if !msg.ScaleSetAssignTime.IsZero() && !msg.RunnerAssignTime.IsZero() { - stubs = append(stubs, newStub( + stubs = append(stubs, r.newStub( traceID, parentSC, "runner.startup", msg.ScaleSetAssignTime, msg.RunnerAssignTime, append(sliceClone(commonAttrs), attribute.String("type", "runner.startup")), @@ -148,7 +215,7 @@ func buildJobSpans(msg *scaleset.JobCompleted, attempt int64) []sdktrace.ReadOnl } if !msg.RunnerAssignTime.IsZero() && !msg.FinishTime.IsZero() { - stubs = append(stubs, newStub( + stubs = append(stubs, r.newStub( traceID, parentSC, "runner.execution", msg.RunnerAssignTime, msg.FinishTime, append(sliceClone(commonAttrs), @@ -200,7 +267,7 @@ func conclusionToSemconv(conclusion string) string { } } -func newStub( +func (r *OTelRecorder) newStub( traceID trace.TraceID, parentSC trace.SpanContext, name string, @@ -215,10 +282,12 @@ func newStub( SpanID: spanID, TraceFlags: trace.FlagsSampled, }), - Parent: parentSC, - StartTime: start, - EndTime: end, - Attributes: attrs, + Parent: parentSC, + StartTime: start, + EndTime: end, + Attributes: attrs, + Resource: r.res, + InstrumentationScope: r.scope, } } diff --git a/cmd/ghalistener/metrics/otel_test.go b/cmd/ghalistener/metrics/otel_test.go index e6789e296d..db00bbb448 100644 --- a/cmd/ghalistener/metrics/otel_test.go +++ b/cmd/ghalistener/metrics/otel_test.go @@ -34,9 +34,14 @@ func (e *captureExporter) Spans() []sdktrace.ReadOnlySpan { return out } +func newTestRecorder(t *testing.T, exp sdktrace.SpanExporter) *OTelRecorder { + t.Helper() + return NewOTelRecorder(OTelRecorderConfig{Exporter: exp}) +} + func TestOTelRecorder_EmitsThreeSpans(t *testing.T) { exp := &captureExporter{} - rec := NewOTelRecorder(exp, nil) + rec := newTestRecorder(t, exp) now := time.Date(2026, 4, 21, 12, 0, 0, 0, time.UTC) msg := &scaleset.JobCompleted{ @@ -101,20 +106,131 @@ func TestOTelRecorder_EmitsThreeSpans(t *testing.T) { assertAttr(t, e, "github.conclusion", "success") assertAttr(t, e, "cicd.pipeline.task.run.result", "success") - // Verify CI/CD semantic convention attributes on all spans + // 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", "42") + 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") + + // 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) + + 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) + + 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) + + 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 := NewOTelRecorder(exp, nil) + rec := newTestRecorder(t, exp) now := time.Now() msg := &scaleset.JobCompleted{ @@ -136,7 +252,7 @@ func TestOTelRecorder_SkipsMissingTimestamps(t *testing.T) { func TestOTelRecorder_CommonAttributes(t *testing.T) { exp := &captureExporter{} - rec := NewOTelRecorder(exp, nil) + rec := newTestRecorder(t, exp) now := time.Now() msg := &scaleset.JobCompleted{ @@ -161,16 +277,16 @@ func TestOTelRecorder_CommonAttributes(t *testing.T) { require.Len(t, spans, 1) s := spans[0] - assertAttr(t, s, "github.job_name", "test-suite") + assertAttr(t, s, "cicd.pipeline.task.name", "test-suite") assertAttr(t, s, "github.repository", "org/repo") - assertAttr(t, s, "github.runner_name", "runner-3") + assertAttr(t, s, "cicd.worker.name", "runner-3") assertAttr(t, s, "github.conclusion", "failure") assertAttr(t, s, "cicd.pipeline.task.run.result", "failure") } func TestOTelRecorder_SetRunAttempt(t *testing.T) { exp := &captureExporter{} - rec := NewOTelRecorder(exp, nil) + rec := newTestRecorder(t, exp) rec.SetRunAttempt(3) now := time.Now() @@ -193,14 +309,14 @@ func TestOTelRecorder_SetRunAttempt(t *testing.T) { func TestOTelRecorder_JobStartedIsNoOp(t *testing.T) { exp := &captureExporter{} - rec := NewOTelRecorder(exp, nil) + rec := newTestRecorder(t, exp) rec.RecordJobStarted(&scaleset.JobStarted{}) assert.Empty(t, exp.Spans()) } func TestOTelRecorder_StatisticsIsNoOp(t *testing.T) { exp := &captureExporter{} - rec := NewOTelRecorder(exp, nil) + rec := newTestRecorder(t, exp) rec.RecordStatistics(&scaleset.RunnerScaleSetStatistic{TotalRunningJobs: 5}) assert.Empty(t, exp.Spans()) } @@ -208,8 +324,8 @@ func TestOTelRecorder_StatisticsIsNoOp(t *testing.T) { func TestCompositeRecorder_DelegatesAll(t *testing.T) { exp1 := &captureExporter{} exp2 := &captureExporter{} - r1 := NewOTelRecorder(exp1, nil) - r2 := NewOTelRecorder(exp2, nil) + r1 := newTestRecorder(t, exp1) + r2 := newTestRecorder(t, exp2) comp := NewComposite(r1, r2) now := time.Now() @@ -267,3 +383,12 @@ func assertAttr(t *testing.T, span sdktrace.ReadOnlySpan, key, expected string) } 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()) + } + } +} From 8509a3125f0aa3eab2e4d4e4635a694b57871456 Mon Sep 17 00:00:00 2001 From: Stefan Penner Date: Wed, 1 Jul 2026 16:19:51 -0600 Subject: [PATCH 07/14] Remove dead SetRunAttempt; document the run-attempt limitation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Nothing in production ever called SetRunAttempt — the scale-set message protocol carries no run attempt, so the override was dead code that only the unit test exercised. Replace it with a documented defaultRunAttempt const and a KNOWN LIMITATION note: re-run (attempt >= 2) listener spans join the first attempt's trace until the message protocol carries the attempt. Co-Authored-By: Claude Fable 5 --- cmd/ghalistener/metrics/otel.go | 32 ++++++++++++---------------- cmd/ghalistener/metrics/otel_test.go | 23 -------------------- 2 files changed, 14 insertions(+), 41 deletions(-) diff --git a/cmd/ghalistener/metrics/otel.go b/cmd/ghalistener/metrics/otel.go index 95ca4e1d86..43d005652c 100644 --- a/cmd/ghalistener/metrics/otel.go +++ b/cmd/ghalistener/metrics/otel.go @@ -42,6 +42,19 @@ var _ listener.MetricsRecorder = &OTelRecorder{} // 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 + type OTelRecorder struct { mu sync.Mutex exporter sdktrace.SpanExporter @@ -49,10 +62,6 @@ type OTelRecorder struct { serverURL string res *resource.Resource scope instrumentation.Scope - - // runAttempt defaults to 1. ARC messages don't include - // run_attempt; override via SetRunAttempt if determinable. - runAttempt int64 } // OTelRecorderConfig configures an OTelRecorder. @@ -118,18 +127,9 @@ func NewOTelRecorder(cfg OTelRecorderConfig) *OTelRecorder { Name: otelScopeName, Version: build.Version, }, - runAttempt: 1, } } -// SetRunAttempt overrides the default run attempt (1) used for trace -// ID generation. -func (r *OTelRecorder) SetRunAttempt(attempt int64) { - r.mu.Lock() - defer r.mu.Unlock() - r.runAttempt = attempt -} - // Shutdown flushes pending spans and releases exporter resources. func (r *OTelRecorder) Shutdown(ctx context.Context) error { return r.exporter.Shutdown(ctx) @@ -138,11 +138,7 @@ func (r *OTelRecorder) Shutdown(ctx context.Context) error { func (r *OTelRecorder) RecordJobStarted(_ *scaleset.JobStarted) {} func (r *OTelRecorder) RecordJobCompleted(msg *scaleset.JobCompleted) { - r.mu.Lock() - attempt := r.runAttempt - r.mu.Unlock() - - spans := r.buildJobSpans(msg, attempt) + spans := r.buildJobSpans(msg, defaultRunAttempt) if len(spans) == 0 { return } diff --git a/cmd/ghalistener/metrics/otel_test.go b/cmd/ghalistener/metrics/otel_test.go index db00bbb448..fcdd3ee22c 100644 --- a/cmd/ghalistener/metrics/otel_test.go +++ b/cmd/ghalistener/metrics/otel_test.go @@ -284,29 +284,6 @@ func TestOTelRecorder_CommonAttributes(t *testing.T) { assertAttr(t, s, "cicd.pipeline.task.run.result", "failure") } -func TestOTelRecorder_SetRunAttempt(t *testing.T) { - exp := &captureExporter{} - rec := newTestRecorder(t, exp) - rec.SetRunAttempt(3) - - now := time.Now() - msg := &scaleset.JobCompleted{ - Result: "Succeeded", - JobMessageBase: scaleset.JobMessageBase{ - WorkflowRunID: 77777, - JobID: "1", - RunnerAssignTime: now, - FinishTime: now.Add(time.Second), - }, - } - - rec.RecordJobCompleted(msg) - - spans := exp.Spans() - require.Len(t, spans, 1) - assert.Equal(t, newTraceID(77777, 3), spans[0].SpanContext().TraceID()) -} - func TestOTelRecorder_JobStartedIsNoOp(t *testing.T) { exp := &captureExporter{} rec := newTestRecorder(t, exp) From 88e63ae9c350d339d9617add4fdd475bb115c58d Mon Sep 17 00:00:00 2001 From: Stefan Penner Date: Wed, 1 Jul 2026 16:28:08 -0600 Subject: [PATCH 08/14] Export spans asynchronously so the autoscaling loop never blocks MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit RecordJobCompleted previously did a synchronous OTLP export (up to 5s) on the listener's single message-loop goroutine — the one that acquires jobs and computes desired runner count — so a slow or down collector serialized 5s stalls into scaling decisions. Spans are now handed to a bounded queue (64 entries) drained by a background goroutine; on overflow the newest job's spans are dropped with a warning. Shutdown drains the queue, bounded by its context, and is idempotent. Co-Authored-By: Claude Fable 5 --- cmd/ghalistener/metrics/otel.go | 66 ++++++++++++++++--- cmd/ghalistener/metrics/otel_test.go | 97 ++++++++++++++++++++++++++++ 2 files changed, 155 insertions(+), 8 deletions(-) diff --git a/cmd/ghalistener/metrics/otel.go b/cmd/ghalistener/metrics/otel.go index 43d005652c..ecdf71d399 100644 --- a/cmd/ghalistener/metrics/otel.go +++ b/cmd/ghalistener/metrics/otel.go @@ -3,6 +3,7 @@ package metrics import ( "context" "crypto/sha256" + "errors" "fmt" "log/slog" "strconv" @@ -55,13 +56,26 @@ var _ listener.MetricsRecorder = &OTelRecorder{} // 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 + type OTelRecorder struct { - mu sync.Mutex 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{} } // OTelRecorderConfig configures an OTelRecorder. @@ -118,7 +132,7 @@ func NewOTelRecorder(cfg OTelRecorderConfig) *OTelRecorder { res = resource.Empty() } - return &OTelRecorder{ + r := &OTelRecorder{ exporter: cfg.Exporter, logger: logger, serverURL: serverURL, @@ -127,12 +141,42 @@ func NewOTelRecorder(cfg OTelRecorderConfig) *OTelRecorder { Name: otelScopeName, Version: build.Version, }, + queue: make(chan []sdktrace.ReadOnlySpan, spanQueueSize), + workerDone: make(chan struct{}), + } + 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 flushes pending spans and releases exporter resources. +// 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 { - return r.exporter.Shutdown(ctx) + 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)) + } } func (r *OTelRecorder) RecordJobStarted(_ *scaleset.JobStarted) {} @@ -143,10 +187,16 @@ func (r *OTelRecorder) RecordJobCompleted(msg *scaleset.JobCompleted) { return } - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() - if err := r.exporter.ExportSpans(ctx, spans); err != nil { - r.logger.Warn("failed to export OTel spans", "error", err) + 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) } } diff --git a/cmd/ghalistener/metrics/otel_test.go b/cmd/ghalistener/metrics/otel_test.go index fcdd3ee22c..d5c780c25c 100644 --- a/cmd/ghalistener/metrics/otel_test.go +++ b/cmd/ghalistener/metrics/otel_test.go @@ -39,6 +39,13 @@ func newTestRecorder(t *testing.T, exp sdktrace.SpanExporter) *OTelRecorder { 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) @@ -64,6 +71,7 @@ func TestOTelRecorder_EmitsThreeSpans(t *testing.T) { } rec.RecordJobCompleted(msg) + flush(t, rec) spans := exp.Spans() require.Len(t, spans, 3) @@ -151,6 +159,7 @@ func TestOTelRecorder_ServerURLIsConfigurable(t *testing.T) { } rec.RecordJobCompleted(msg) + flush(t, rec) spans := exp.Spans() require.Len(t, spans, 1) @@ -178,6 +187,7 @@ func TestOTelRecorder_ResourceAndScope(t *testing.T) { } rec.RecordJobCompleted(msg) + flush(t, rec) spans := exp.Spans() require.Len(t, spans, 1) @@ -216,6 +226,7 @@ func TestOTelRecorder_ResourceHonorsOTelEnv(t *testing.T) { } rec.RecordJobCompleted(msg) + flush(t, rec) spans := exp.Spans() require.Len(t, spans, 1) @@ -244,6 +255,7 @@ func TestOTelRecorder_SkipsMissingTimestamps(t *testing.T) { } rec.RecordJobCompleted(msg) + flush(t, rec) spans := exp.Spans() require.Len(t, spans, 1, "only runner.execution when queue/startup timestamps are zero") @@ -272,6 +284,7 @@ func TestOTelRecorder_CommonAttributes(t *testing.T) { } rec.RecordJobCompleted(msg) + flush(t, rec) spans := exp.Spans() require.Len(t, spans, 1) @@ -284,6 +297,88 @@ func TestOTelRecorder_CommonAttributes(t *testing.T) { 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) @@ -316,6 +411,8 @@ func TestCompositeRecorder_DelegatesAll(t *testing.T) { }, } comp.RecordJobCompleted(msg) + flush(t, r1) + flush(t, r2) assert.Len(t, exp1.Spans(), 1) assert.Len(t, exp2.Spans(), 1) From d6e5c0d3acb9a3b9de38dc5877c6eb165ba0759e Mon Sep 17 00:00:00 2001 From: Stefan Penner Date: Wed, 1 Jul 2026 16:28:35 -0600 Subject: [PATCH 09/14] Shut down OTel recorder with a fresh bounded context; log the error MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The deferred Shutdown captured the signal.NotifyContext ctx, which is already canceled on SIGTERM — exactly when the queued spans need flushing (pod rollout / scale-down). Mirror the sessionClient.Close pattern: fresh context.Background() with a 5s bound, error logged instead of dropped. Co-Authored-By: Claude Fable 5 --- cmd/ghalistener/main.go | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/cmd/ghalistener/main.go b/cmd/ghalistener/main.go index da910139d3..f4470b3460 100644 --- a/cmd/ghalistener/main.go +++ b/cmd/ghalistener/main.go @@ -7,6 +7,7 @@ import ( "os" "os/signal" "syscall" + "time" "github.com/actions/actions-runner-controller/cmd/ghalistener/config" "github.com/actions/actions-runner-controller/cmd/ghalistener/metrics" @@ -86,7 +87,16 @@ func run(ctx context.Context, config *config.Config) error { ScaleSetName: config.EphemeralRunnerSetName, ScaleSetNamespace: config.EphemeralRunnerSetNamespace, }) - defer otelRecorder.Shutdown(ctx) + // 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) + } + }() logger.Info("OTel trace recorder enabled", "endpoint", config.OTelEndpoint) } From b21955461717b28d84c677998b45efeece8cb5fb Mon Sep 17 00:00:00 2001 From: Stefan Penner Date: Wed, 1 Jul 2026 16:31:06 -0600 Subject: [PATCH 10/14] Fix OTLP endpoint handling: full URL semantics, spec-compliant env path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit OTEL_EXPORTER_OTLP_ENDPOINT is a full URL per the OTLP spec, but it was funneled into otlptracehttp.WithEndpoint (host:port only), which produced an unparseable URL — every export from a spec-configured deployment failed with only a per-job warning. - config otel_endpoint is now parsed as a full base URL (scheme validated, /v1/traces appended when no path, TLS from the scheme) and passed via WithEndpointURL; the otel_insecure field is gone. - When only the standard OTEL_EXPORTER_OTLP_(TRACES_)ENDPOINT env vars are set, the exporter is built with no endpoint option so the SDK's spec-compliant env handling applies (endpoint, headers, TLS). - config.Read no longer copies env vars into the config. Both activation paths are covered by httptest-backed exporter tests. Co-Authored-By: Claude Fable 5 --- cmd/ghalistener/config/config.go | 14 ++--- cmd/ghalistener/main.go | 48 +++++++++++++---- cmd/ghalistener/main_test.go | 91 ++++++++++++++++++++++++++++++++ 3 files changed, 134 insertions(+), 19 deletions(-) create mode 100644 cmd/ghalistener/main_test.go diff --git a/cmd/ghalistener/config/config.go b/cmd/ghalistener/config/config.go index ba4bdb84e5..8d7930e7d1 100644 --- a/cmd/ghalistener/config/config.go +++ b/cmd/ghalistener/config/config.go @@ -44,8 +44,11 @@ type Config struct { MetricsAddr string `json:"metrics_addr"` MetricsEndpoint string `json:"metrics_endpoint"` Metrics *v1alpha1.MetricsConfig `json:"metrics"` - OTelEndpoint string `json:"otel_endpoint"` - OTelInsecure bool `json:"otel_insecure"` + // 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) { @@ -60,13 +63,6 @@ func Read(ctx context.Context, configPath string) (*Config, error) { return nil, fmt.Errorf("failed to decode config: %w", err) } - if v := os.Getenv("OTEL_EXPORTER_OTLP_ENDPOINT"); v != "" && config.OTelEndpoint == "" { - config.OTelEndpoint = v - } - if os.Getenv("OTEL_EXPORTER_OTLP_INSECURE") == "true" && !config.OTelInsecure { - config.OTelInsecure = true - } - var vault vault.Vault switch config.VaultType { case "": diff --git a/cmd/ghalistener/main.go b/cmd/ghalistener/main.go index f4470b3460..8f4d76c986 100644 --- a/cmd/ghalistener/main.go +++ b/cmd/ghalistener/main.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "log" + "net/url" "os" "os/signal" "syscall" @@ -16,6 +17,7 @@ import ( "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" ) @@ -41,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 { @@ -67,16 +92,15 @@ func run(ctx context.Context, config *config.Config) error { }) } - // OTel trace recorder (optional) + // 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 - if config.OTelEndpoint != "" { - opts := []otlptracehttp.Option{ - otlptracehttp.WithEndpoint(config.OTelEndpoint), - } - if config.OTelInsecure { - opts = append(opts, otlptracehttp.WithInsecure()) - } - otlpExporter, err := otlptracehttp.New(ctx, opts...) + 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) } @@ -97,7 +121,11 @@ func run(ctx context.Context, config *config.Config) error { logger.Error("Failed to shut down OTel recorder", "error", err) } }() - logger.Info("OTel trace recorder enabled", "endpoint", config.OTelEndpoint) + endpointSource := config.OTelEndpoint + if endpointSource == "" { + endpointSource = "OTEL_EXPORTER_OTLP_* environment" + } + logger.Info("OTel trace recorder enabled", "endpoint", endpointSource) } hostname, err := os.Hostname() 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) +} From 0a642027d384ce18e93ea297b4b715bbfd625117 Mon Sep 17 00:00:00 2001 From: Stefan Penner Date: Wed, 1 Jul 2026 16:35:49 -0600 Subject: [PATCH 11/14] Add first-class deployment surface for listener OTel trace export Wire the OTLP endpoint the same way the existing listener Prometheus metrics flow: - controller flag --listener-otel-endpoint (default: disabled) - AutoscalingListenerReconciler.ListenerOTelEndpoint -> newScaleSetListenerConfig -> otel_endpoint in the listener's config.json secret - gha-runner-scale-set-controller chart value otel.listenerEndpoint (commented out by default, like metrics), with chart tests for the rendered arg and the disabled default - docs/gha-runner-scale-set-controller/otel-tracing.md documenting both activation paths (chart value and standard OTEL_* env via listenerTemplate), the ID contract, and known limitations Follow-up (out of scope for this PR): per-scale-set configuration via an AutoscalingRunnerSet field, and OTLP header/auth plumbing. Co-Authored-By: Claude Fable 5 --- .../templates/deployment.yaml | 5 ++ .../tests/template_test.go | 56 +++++++++++++++++++ .../values.yaml | 7 +++ .../autoscalinglistener_controller.go | 9 ++- .../actions.github.com/resourcebuilder.go | 3 +- .../resourcebuilder_test.go | 55 ++++++++++++++++++ .../otel-tracing.md | 53 ++++++++++++++++++ main.go | 3 + 8 files changed, 188 insertions(+), 3 deletions(-) create mode 100644 docs/gha-runner-scale-set-controller/otel-tracing.md 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/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/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") From 850542afeb2299c5c80e9eba8336f1c7b2210c95 Mon Sep 17 00:00:00 2001 From: Stefan Penner Date: Wed, 1 Jul 2026 17:20:49 -0600 Subject: [PATCH 12/14] metrics/otel: conformance fixes M1 + M3; M2 scope comment (runner cross-ref) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit M1 — rename span-type attribute key from `type` to `github.record_type` to match the runner's OTelTraceExporter.cs (lines 526, 618, 711). Values (runner.queue / runner.startup / runner.execution) are unchanged. M3 — port runner's result normalisation: • normalizeConclusion: add abandoned→failure; unknown raw → "error" (was: strings.ToLower pass-through) • conclusionToSemconv: unknown conclusion → "error" (was: raw conclusion pass-through) Both mirror the runner's NormalizeConclusion / ToSemconvResult behaviour (OTelTraceExporter.cs:1008-1033). M2 — cicd.worker.name/id remain as SPAN attrs on the listener (one listener serves many runners, so per-span is semantically correct). Added a code comment explaining the deliberate scope difference vs. the runner's resource-scoped attrs. Doc note added to runner repo separately. Co-Authored-By: Claude Fable 5 --- cmd/ghalistener/metrics/otel.go | 25 ++++-- cmd/ghalistener/metrics/otel_test.go | 117 ++++++++++++++++++++++++++- 2 files changed, 132 insertions(+), 10 deletions(-) diff --git a/cmd/ghalistener/metrics/otel.go b/cmd/ghalistener/metrics/otel.go index ecdf71d399..071196bd9a 100644 --- a/cmd/ghalistener/metrics/otel.go +++ b/cmd/ghalistener/metrics/otel.go @@ -236,6 +236,11 @@ func (r *OTelRecorder) buildJobSpans(msg *scaleset.JobCompleted, attempt int64) 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), @@ -248,7 +253,8 @@ func (r *OTelRecorder) buildJobSpans(msg *scaleset.JobCompleted, attempt int64) stubs = append(stubs, r.newStub( traceID, parentSC, "runner.queue", msg.QueueTime, msg.ScaleSetAssignTime, - append(sliceClone(commonAttrs), attribute.String("type", "runner.queue")), + // github.record_type matches the runner's attribute key (OTelTraceExporter.cs:526,618,711) + append(sliceClone(commonAttrs), attribute.String("github.record_type", "runner.queue")), )) } @@ -256,7 +262,7 @@ func (r *OTelRecorder) buildJobSpans(msg *scaleset.JobCompleted, attempt int64) stubs = append(stubs, r.newStub( traceID, parentSC, "runner.startup", msg.ScaleSetAssignTime, msg.RunnerAssignTime, - append(sliceClone(commonAttrs), attribute.String("type", "runner.startup")), + append(sliceClone(commonAttrs), attribute.String("github.record_type", "runner.startup")), )) } @@ -265,7 +271,7 @@ func (r *OTelRecorder) buildJobSpans(msg *scaleset.JobCompleted, attempt int64) traceID, parentSC, "runner.execution", msg.RunnerAssignTime, msg.FinishTime, append(sliceClone(commonAttrs), - attribute.String("type", "runner.execution"), + attribute.String("github.record_type", "runner.execution"), attribute.String("github.conclusion", conclusion), attribute.String("cicd.pipeline.task.run.result", conclusionToSemconv(conclusion)), ), @@ -276,11 +282,16 @@ func (r *OTelRecorder) buildJobSpans(msg *scaleset.JobCompleted, attempt int64) } 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": @@ -290,11 +301,13 @@ func normalizeConclusion(raw string) string { case "startup_failure": return "startup_failure" default: - return strings.ToLower(raw) + 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" @@ -306,10 +319,8 @@ func conclusionToSemconv(conclusion string) string { return "skip" case "timed_out": return "timeout" - case "startup_failure": - return "error" default: - return conclusion + return "error" } } diff --git a/cmd/ghalistener/metrics/otel_test.go b/cmd/ghalistener/metrics/otel_test.go index d5c780c25c..a48607db9c 100644 --- a/cmd/ghalistener/metrics/otel_test.go +++ b/cmd/ghalistener/metrics/otel_test.go @@ -98,19 +98,20 @@ func TestOTelRecorder_EmitsThreeSpans(t *testing.T) { require.NotNil(t, q) assert.Equal(t, now, q.StartTime()) assert.Equal(t, now.Add(10*time.Second), q.EndTime()) - assertAttr(t, q, "type", "runner.queue") + // 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, "type", "runner.startup") + 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, "type", "runner.execution") + assertAttr(t, e, "github.record_type", "runner.execution") assertAttr(t, e, "github.conclusion", "success") assertAttr(t, e, "cicd.pipeline.task.run.result", "success") @@ -129,6 +130,8 @@ func TestOTelRecorder_EmitsThreeSpans(t *testing.T) { 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") @@ -447,6 +450,114 @@ func TestDeterministicIDs(t *testing.T) { 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") +} + func assertAttr(t *testing.T, span sdktrace.ReadOnlySpan, key, expected string) { t.Helper() for _, a := range span.Attributes() { From 143281afef8002b1b73312084c6d1a2cdb5f0689 Mon Sep 17 00:00:00 2001 From: Stefan Penner Date: Wed, 1 Jul 2026 17:52:22 -0600 Subject: [PATCH 13/14] Fix runner.queue span never firing when broker sends QueueTime=0 on JobCompleted MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The GitHub broker omits QueueTime from JobCompleted in production (observed across multiple runs). RecordJobStarted now stores QueueTime and ScaleSetAssignTime from the earlier JobStarted message — which does carry them — in a bounded, TTL-capped map keyed by JobID. buildJobSpans falls back to these stored timestamps when msg.QueueTime is zero, so the runner.queue span is emitted correctly in the common production path. Bounds: map capped at 1 000 entries (warn + drop on overflow); entries older than 24 h are discarded at lookup to prevent leaks across months-long listener runs. Tests added: - TestOTelRecorder_QueueSpanFromJobStarted: JobStarted(QueueTime≠0) → JobCompleted(QueueTime=0) still emits runner.queue with correct interval - TestOTelRecorder_QueueEntryEvictedOnCompleted: duplicate JobCompleted for the same job emits runner.queue exactly once (entry evicted on first) Co-Authored-By: Claude Fable 5 --- cmd/ghalistener/metrics/otel.go | 102 +++++++++++++++++++++++++-- cmd/ghalistener/metrics/otel_test.go | 95 +++++++++++++++++++++++++ 2 files changed, 190 insertions(+), 7 deletions(-) diff --git a/cmd/ghalistener/metrics/otel.go b/cmd/ghalistener/metrics/otel.go index 071196bd9a..9a6694c472 100644 --- a/cmd/ghalistener/metrics/otel.go +++ b/cmd/ghalistener/metrics/otel.go @@ -65,6 +65,38 @@ 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 broker omits QueueTime from JobCompleted in +// production (confirmed across multiple runs). JobStarted, which +// arrives earlier, does carry the broker-set QueueTime and +// ScaleSetAssignTime needed to reconstruct the runner.queue span. +// +// 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 @@ -76,6 +108,10 @@ type OTelRecorder struct { stopped bool queue chan []sdktrace.ReadOnlySpan workerDone chan struct{} + + // pendingMu guards pendingQueue. + pendingMu sync.Mutex + pendingQueue map[string]pendingQueueEntry } // OTelRecorderConfig configures an OTelRecorder. @@ -141,8 +177,9 @@ func NewOTelRecorder(cfg OTelRecorderConfig) *OTelRecorder { Name: otelScopeName, Version: build.Version, }, - queue: make(chan []sdktrace.ReadOnlySpan, spanQueueSize), - workerDone: make(chan struct{}), + queue: make(chan []sdktrace.ReadOnlySpan, spanQueueSize), + workerDone: make(chan struct{}), + pendingQueue: make(map[string]pendingQueueEntry), } go r.exportLoop() return r @@ -179,10 +216,40 @@ func (r *OTelRecorder) Shutdown(ctx context.Context) error { } } -func (r *OTelRecorder) RecordJobStarted(_ *scaleset.JobStarted) {} +// 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) { - spans := r.buildJobSpans(msg, defaultRunAttempt) + // 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 } @@ -203,7 +270,16 @@ func (r *OTelRecorder) RecordJobCompleted(msg *scaleset.JobCompleted) { func (r *OTelRecorder) RecordStatistics(_ *scaleset.RunnerScaleSetStatistic) {} func (r *OTelRecorder) RecordDesiredRunners(_ int) {} -func (r *OTelRecorder) buildJobSpans(msg *scaleset.JobCompleted, attempt int64) []sdktrace.ReadOnlySpan { +// 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 @@ -249,10 +325,22 @@ func (r *OTelRecorder) buildJobSpans(msg *scaleset.JobCompleted, attempt int64) var stubs tracetest.SpanStubs - if !msg.QueueTime.IsZero() && !msg.ScaleSetAssignTime.IsZero() { + // 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", - msg.QueueTime, msg.ScaleSetAssignTime, + 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")), )) diff --git a/cmd/ghalistener/metrics/otel_test.go b/cmd/ghalistener/metrics/otel_test.go index a48607db9c..8db57e2150 100644 --- a/cmd/ghalistener/metrics/otel_test.go +++ b/cmd/ghalistener/metrics/otel_test.go @@ -558,6 +558,101 @@ func TestOTelRecorder_UnknownConclusion(t *testing.T) { 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() { From e39d3c51c34e12545891c527e26a4bf5585b600a Mon Sep 17 00:00:00 2001 From: Stefan Penner Date: Wed, 1 Jul 2026 18:07:10 -0600 Subject: [PATCH 14/14] otel: document that github.com sends QueueTime=0 on JobStarted too MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Verified live 2026-07-01 with diag logging: only ScaleSetAssignTime/ RunnerAssignTime are populated, so runner.queue currently never fires. Keep the JobStarted fallback — cheap, tested, lights up if the broker ever populates the field (GHES or future fix). Co-Authored-By: Claude Fable 5 --- cmd/ghalistener/metrics/otel.go | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/cmd/ghalistener/metrics/otel.go b/cmd/ghalistener/metrics/otel.go index 9a6694c472..03cb1391cc 100644 --- a/cmd/ghalistener/metrics/otel.go +++ b/cmd/ghalistener/metrics/otel.go @@ -82,10 +82,14 @@ const pendingQueueTTL = 24 * time.Hour // JobStarted message. They are keyed by JobID and evicted when the // corresponding JobCompleted arrives. // -// Background: the GitHub broker omits QueueTime from JobCompleted in -// production (confirmed across multiple runs). JobStarted, which -// arrives earlier, does carry the broker-set QueueTime and -// ScaleSetAssignTime needed to reconstruct the runner.queue span. +// 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