feat: OpenTelemetry tracing (#1206)

Adds OpenTelemetry tracing, configured through the standard `OTEL_*` environment variables and off until an OTLP endpoint is set. ![image.png](https://gitea.com/attachments/c4b49cdf-a32d-403c-94e9-30cd06ee6477) Each job becomes a trace with a span per step, so operators can see where CI time goes in their existing observability backend. Jobs receive only `TRACEPARENT`, so instrumented tools inside them join the trace without seeing the collector's credentials. Metrics stay on the Prometheus endpoint, which a collector can scrape. See [docs/telemetry.md](https://gitea.com/wmTJc9IK0Q/runner/src/branch/otel/docs/telemetry.md) for setup and security notes. Also keeps `GITHUB_SHA` and `GITHUB_REF` fixed for a whole run when HEAD moves during `exec` runs and the act tests. [image.png](https://gitea.com/attachments/c4b49cdf-a32d-403c-94e9-30cd06ee6477) [image.png](https://gitea.com/attachments/c4b49cdf-a32d-403c-94e9-30cd06ee6477) --------- Co-authored-by: silverwind <[email protected]> Co-authored-by: silverwind <[email protected]> Co-authored-by: bircni <[email protected]> Reviewed-on: https://gitea.com/gitea/runner/pulls/1206 Reviewed-by: bircni <[email protected]> Reviewed-by: silverwind <[email protected]> Co-authored-by: wmTJc9IK0Q <[email protected]>
2026-11-07 15:50:28 +00:00 · 2026-09-24 15:57:18 +00:00
1 parent 7ad12038d1
commit 1952e5cdb6
@@ -0,0 +1,240 @@
+// Copyright 2026 The Gitea Authors. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+package telemetry
+
+import (
+	"bytes"
+	"cmp"
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"os"
+	"slices"
+	"strconv"
+	"strings"
+
+	"gitea.com/gitea/runner/internal/pkg/ver"
+
+	"gitea.dev/actionslib/pkg/model"
+	runnerv1 "gitea.dev/actionslib/runner/v1"
+	"github.com/avast/retry-go/v5"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/codes"
+	"go.opentelemetry.io/otel/exporters/otlp/otlptrace"
+	"go.opentelemetry.io/otel/propagation"
+	"go.opentelemetry.io/otel/sdk/resource"
+	sdktrace "go.opentelemetry.io/otel/sdk/trace"
+	semconv "go.opentelemetry.io/otel/semconv/v1.43.0"
+	"go.opentelemetry.io/otel/trace"
+	tracenoop "go.opentelemetry.io/otel/trace/noop"
+	tracepb "go.opentelemetry.io/proto/otlp/trace/v1"
+	"google.golang.org/protobuf/proto"
+)
+
+const scope = "gitea.com/gitea/runner"
+
+type taskRunKey struct{}
+
+type taskRun struct {
+	id, url, pipeline string
+	span              trace.Span
+}
+
+// tracer stays off the global provider and ctx carries only span contexts, so the Docker client's otelhttp records nothing.
+var tracer trace.Tracer = tracenoop.Tracer{}
+
+func SetTracerProvider(provider trace.TracerProvider) {
+	tracer = provider.Tracer(scope)
+}
+
+func Setup(ctx context.Context, uuid, name string) (func(context.Context) error, error) {
+	noop := func(context.Context) error { return nil }
+	endpoint := os.Getenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT")
+	if base := os.Getenv("OTEL_EXPORTER_OTLP_ENDPOINT"); endpoint == "" && base != "" {
+		endpoint = strings.TrimSuffix(base, "/") + "/v1/traces"
+	}
+	if endpoint == "" || strings.EqualFold(os.Getenv("OTEL_SDK_DISABLED"), "true") || strings.EqualFold(os.Getenv("OTEL_TRACES_EXPORTER"), "none") {
+		return noop, nil
+	}
+	if protocol := cmp.Or(os.Getenv("OTEL_EXPORTER_OTLP_TRACES_PROTOCOL"), os.Getenv("OTEL_EXPORTER_OTLP_PROTOCOL")); protocol != "" && !strings.EqualFold(protocol, "http/protobuf") {
+		return noop, fmt.Errorf("OTLP protocol %q is not supported, only http/protobuf is", protocol)
+	}
+	res, err := resource.New(ctx, resource.WithTelemetrySDK(), resource.WithAttributes(
+		semconv.ServiceName("gitea-runner"),
+		semconv.ServiceVersion(ver.Version()),
+		semconv.ServiceInstanceID(uuid),
+		semconv.CICDWorkerID(uuid),
+		semconv.CICDWorkerName(name),
+	), resource.WithFromEnv())
+	if err != nil {
+		return noop, err
+	}
+	headers := http.Header{}
+	for _, key := range []string{"OTEL_EXPORTER_OTLP_HEADERS", "OTEL_EXPORTER_OTLP_TRACES_HEADERS"} {
+		for pair := range strings.SplitSeq(os.Getenv(key), ",") {
+			key, value, found := strings.Cut(pair, "=")
+			if decoded, err := url.PathUnescape(strings.TrimSpace(value)); found && err == nil {
+				headers.Set(strings.TrimSpace(key), decoded)
+			}
+		}
+	}
+	provider := sdktrace.NewTracerProvider(sdktrace.WithBatcher(otlptrace.NewUnstarted(&client{endpoint, headers})), sdktrace.WithResource(res))
+	SetTracerProvider(provider)
+	return provider.Shutdown, nil
+}
+
+// client replaces otlptracehttp, which links gRPC (+7 MB): https://github.com/open-telemetry/opentelemetry-go/issues/2579
+type client struct {
+	endpoint string
+	headers  http.Header
+}
+
+// exportClient refuses redirects, which would carry the collector's header credentials to another host.
+var exportClient = &http.Client{CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }}
+
+func (*client) Start(context.Context) error { return nil }
+
+func (*client) Stop(context.Context) error { return nil }
+
+func (c *client) UploadTraces(ctx context.Context, spans []*tracepb.ResourceSpans) error {
+	body, err := proto.Marshal(&tracepb.TracesData{ResourceSpans: spans})
+	if err != nil {
+		return err
+	}
+	return retry.New(retry.Context(ctx), retry.Attempts(5), retry.LastErrorOnly(true)).Do(func() error {
+		req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.endpoint, bytes.NewReader(body))
+		if err != nil {
+			return retry.Unrecoverable(err)
+		}
+		req.Header = c.headers.Clone()
+		req.Header.Set("Content-Type", "application/x-protobuf")
+		resp, err := exportClient.Do(req)
+		if err != nil {
+			return err
+		}
+		_, _ = io.Copy(io.Discard, resp.Body)
+		_ = resp.Body.Close()
+		if resp.StatusCode < http.StatusMultipleChoices {
+			return nil
+		}
+		err = fmt.Errorf("OTLP export failed: %s", resp.Status)
+		if slices.Contains([]int{http.StatusTooManyRequests, http.StatusBadGateway, http.StatusServiceUnavailable, http.StatusGatewayTimeout}, resp.StatusCode) {
+			return err
+		}
+		return retry.Unrecoverable(err)
+	})
+}
+
+func StartJob(ctx context.Context, task *runnerv1.Task) (context.Context, func(runnerv1.Result)) {
+	field := func(key string) string { return task.GetContext().GetFields()[key].GetStringValue() }
+	pipeline, owner, repo := field("workflow"), field("repository_owner"), field("repository")
+	repoURL := strings.TrimSuffix(field("server_url"), "/") + "/" + repo
+	runURL := repoURL + "/actions/runs/" + field("run_id")
+	_, span := tracer.Start(ctx, "RUN "+pipeline, trace.WithSpanKind(trace.SpanKindServer), trace.WithAttributes(
+		semconv.CICDPipelineName(pipeline),
+		semconv.CICDPipelineActionNameRun,
+		semconv.CICDPipelineRunID(field("run_id")),
+		semconv.CICDPipelineRunURLFull(runURL),
+		semconv.VCSOwnerName(owner),
+		semconv.VCSRepositoryName(strings.TrimPrefix(repo, owner+"/")),
+		semconv.VCSRepositoryURLFull(repoURL),
+		semconv.VCSRefHeadName(field("ref_name")),
+		semconv.VCSRefHeadRevision(field("sha")),
+		attribute.String("gitea.job.key", field("job")),
+		attribute.String("gitea.run.number", field("run_number")),
+		attribute.String("gitea.run.attempt", field("run_attempt")),
+		attribute.Int64("gitea.task.id", task.Id),
+	))
+	if !span.SpanContext().IsValid() {
+		return ctx, func(runnerv1.Result) {}
+	}
+	ctx = context.WithValue(trace.ContextWithSpanContext(ctx, span.SpanContext()), taskRunKey{}, taskRun{strconv.FormatInt(task.Id, 10), runURL, pipeline, span})
+	return ctx, func(result runnerv1.Result) {
+		end(span, semconv.CICDPipelineResultKey, jobResult(ctx, result))
+	}
+}
+
+func SetJobName(ctx context.Context, job string) {
+	if run, ok := ctx.Value(taskRunKey{}).(taskRun); ok {
+		run.span.SetName("RUN " + run.pipeline + " " + job)
+	}
+}
+
+func AddJobEnv(ctx context.Context, envs map[string]string) {
+	carrier := propagation.MapCarrier{}
+	propagation.TraceContext{}.Inject(ctx, carrier)
+	for key, value := range carrier {
+		envs[strings.ToUpper(key)] = value
+	}
+}
+
+func StartStep(ctx context.Context, step *model.Step, stage string) (context.Context, func(error, *model.StepResult)) {
+	parent, ok := ctx.Value(taskRunKey{}).(taskRun)
+	if !ok {
+		return ctx, func(error, *model.StepResult) {}
+	}
+	name, id := step.Name, parent.id+"."+strconv.Itoa(step.Number)
+	if name == "" {
+		run, _, _ := strings.Cut(strings.TrimSpace(step.Run), "\n")
+		name = "Run " + cmp.Or(strings.TrimSpace(run), step.String())
+	}
+	if stage != "Main" {
+		name, id = stage+" "+name, id+"."+strings.ToLower(stage)
+	}
+	_, span := tracer.Start(ctx, name, trace.WithAttributes(
+		semconv.CICDPipelineTaskName(name),
+		semconv.CICDPipelineTaskRunID(id),
+		semconv.CICDPipelineTaskRunURLFull(parent.url),
+	))
+	ctx = context.WithValue(trace.ContextWithSpanContext(ctx, span.SpanContext()), taskRunKey{}, taskRun{id, parent.url, parent.pipeline, span})
+	return ctx, func(err error, result *model.StepResult) {
+		end(span, semconv.CICDPipelineTaskRunResultKey, stepResult(ctx, err, result))
+	}
+}
+
+func jobResult(ctx context.Context, result runnerv1.Result) string {
+	switch {
+	case result == runnerv1.Result_RESULT_SUCCESS:
+		return "success"
+	case result == runnerv1.Result_RESULT_SKIPPED:
+		return "skip"
+	case errors.Is(ctx.Err(), context.DeadlineExceeded):
+		return "timeout"
+	case result == runnerv1.Result_RESULT_CANCELLED:
+		return "cancellation"
+	case result == runnerv1.Result_RESULT_FAILURE:
+		return "failure"
+	}
+	return "error"
+}
+
+func stepResult(ctx context.Context, err error, result *model.StepResult) string {
+	switch {
+	case errors.Is(err, context.DeadlineExceeded) || errors.Is(ctx.Err(), context.DeadlineExceeded):
+		return "timeout"
+	case errors.Is(err, context.Canceled) || ctx.Err() != nil:
+		return "cancellation"
+	case err != nil || result.Conclusion == model.StepStatusFailure:
+		return "failure"
+	case result.Conclusion == model.StepStatusSkipped:
+		return "skip"
+	}
+	return "success"
+}
+
+func end(span trace.Span, key attribute.Key, result string) {
+	span.SetAttributes(key.String(result))
+	switch result {
+	case "timeout":
+		span.SetAttributes(semconv.ErrorTypeKey.String(result))
+		span.SetStatus(codes.Error, "")
+	case "failure", "error":
+		span.SetAttributes(semconv.ErrorTypeOther)
+		span.SetStatus(codes.Error, "")
+	}
+	span.End()
+}
@@ -0,0 +1,186 @@
+// Copyright 2026 The Gitea Authors. All rights reserved.
+// SPDX-License-Identifier: MIT
+
+package telemetry
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"gitea.dev/actionslib/pkg/model"
+	runnerv1 "gitea.dev/actionslib/runner/v1"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/codes"
+	sdktrace "go.opentelemetry.io/otel/sdk/trace"
+	"go.opentelemetry.io/otel/sdk/trace/tracetest"
+	semconv "go.opentelemetry.io/otel/semconv/v1.43.0"
+	"go.opentelemetry.io/otel/trace"
+	tracenoop "go.opentelemetry.io/otel/trace/noop"
+	tracepb "go.opentelemetry.io/proto/otlp/trace/v1"
+	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/types/known/structpb"
+)
+
+func setEnv(t *testing.T, env map[string]string) {
+	t.Cleanup(func() { SetTracerProvider(tracenoop.NewTracerProvider()) })
+	for _, key := range []string{
+		"OTEL_SDK_DISABLED", "OTEL_TRACES_EXPORTER", "OTEL_EXPORTER_OTLP_ENDPOINT", "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT",
+		"OTEL_EXPORTER_OTLP_PROTOCOL", "OTEL_EXPORTER_OTLP_TRACES_PROTOCOL", "OTEL_EXPORTER_OTLP_HEADERS", "OTEL_EXPORTER_OTLP_TRACES_HEADERS",
+	} {
+		t.Setenv(key, env[key])
+	}
+}
+
+func TestSetup(t *testing.T) {
+	for _, tc := range []struct {
+		env     map[string]string
+		enabled bool
+		err     string
+	}{
+		{},
+		{env: map[string]string{"OTEL_EXPORTER_OTLP_TRACES_ENDPOINT": "http://collector:4318/v1/traces"}, enabled: true},
+		{env: map[string]string{"OTEL_EXPORTER_OTLP_ENDPOINT": "http://collector:4318", "OTEL_TRACES_EXPORTER": "NONE"}},
+		{env: map[string]string{"OTEL_EXPORTER_OTLP_ENDPOINT": "http://collector:4318", "OTEL_SDK_DISABLED": "TRUE"}},
+		{env: map[string]string{"OTEL_EXPORTER_OTLP_ENDPOINT": "http://collector:4318", "OTEL_EXPORTER_OTLP_PROTOCOL": "grpc"}, err: `OTLP protocol "grpc" is not supported, only http/protobuf is`},
+	} {
+		t.Run(fmt.Sprint(tc.env), func(t *testing.T) {
+			setEnv(t, tc.env)
+			shutdown, err := Setup(t.Context(), "uuid", "runner")
+			if tc.err != "" {
+				require.EqualError(t, err, tc.err)
+			} else {
+				require.NoError(t, err)
+			}
+			_, disabled := tracer.(tracenoop.Tracer)
+			assert.Equal(t, tc.enabled, !disabled)
+			require.NoError(t, shutdown(t.Context()))
+		})
+	}
+}
+
+func TestExportRetriesAndMatchesOTLP(t *testing.T) {
+	bodies := make(chan []byte, 2)
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		assert.Equal(t, "/otlp/v1/traces", r.URL.Path)
+		assert.Equal(t, "Bearer token", r.Header.Get("Authorization"))
+		body, _ := io.ReadAll(r.Body)
+		bodies <- body
+		if len(bodies) == 1 {
+			w.WriteHeader(http.StatusServiceUnavailable)
+		}
+	}))
+	defer server.Close()
+	setEnv(t, map[string]string{"OTEL_EXPORTER_OTLP_ENDPOINT": server.URL + "/otlp/", "OTEL_EXPORTER_OTLP_HEADERS": "authorization=Bearer%20token"})
+
+	shutdown, err := Setup(t.Context(), "uuid", "runner")
+	require.NoError(t, err)
+	_, span := tracer.Start(t.Context(), "op")
+	span.End()
+	require.NoError(t, shutdown(t.Context()))
+
+	require.Len(t, bodies, 2)
+	<-bodies
+	var decoded tracepb.TracesData
+	require.NoError(t, proto.Unmarshal(<-bodies, &decoded))
+	assert.Equal(t, "op", decoded.GetResourceSpans()[0].GetScopeSpans()[0].GetSpans()[0].GetName())
+}
+
+func TestExportRefusesRedirects(t *testing.T) {
+	target := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { t.Error("redirect followed") }))
+	defer target.Close()
+	origin := httptest.NewServer(http.RedirectHandler(target.URL, http.StatusTemporaryRedirect))
+	defer origin.Close()
+	require.EqualError(t, (&client{origin.URL, http.Header{}}).UploadTraces(t.Context(), nil), "OTLP export failed: 307 Temporary Redirect")
+}
+
+func TestJob(t *testing.T) {
+	fields, err := structpb.NewStruct(map[string]any{
+		"workflow":         "ci.yml",
+		"repository":       "owner/repo",
+		"repository_owner": "owner",
+		"server_url":       "https://gitea.example/",
+		"run_id":           "42",
+		"run_number":       "40",
+		"run_attempt":      "2",
+		"job":              "build",
+	})
+	require.NoError(t, err)
+	task := &runnerv1.Task{Id: 7, Context: fields}
+
+	envs := map[string]string{}
+	ctx, endJob := StartJob(t.Context(), task)
+	AddJobEnv(ctx, envs)
+	stepCtx, endStep := StartStep(ctx, &model.Step{}, "Main")
+	endStep(nil, nil)
+	endJob(runnerv1.Result_RESULT_SUCCESS)
+	assert.Equal(t, ctx, stepCtx)
+	assert.Empty(t, envs)
+
+	spans := tracetest.NewSpanRecorder()
+	SetTracerProvider(sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(spans)))
+	t.Cleanup(func() { SetTracerProvider(tracenoop.NewTracerProvider()) })
+
+	ctx, endJob = StartJob(t.Context(), task)
+	SetJobName(ctx, "Build and test")
+	AddJobEnv(ctx, envs)
+	stepCtx, endStep = StartStep(ctx, &model.Step{Run: "make\nmake test", Number: 1}, "Main")
+	assert.False(t, trace.SpanFromContext(ctx).IsRecording())
+	assert.False(t, trace.SpanFromContext(stepCtx).IsRecording())
+	endStep(assert.AnError, nil)
+	endJob(runnerv1.Result_RESULT_FAILURE)
+
+	jobSpan := trace.SpanContextFromContext(ctx)
+	assert.Equal(t, "00-"+jobSpan.TraceID().String()+"-"+jobSpan.SpanID().String()+"-01", envs["TRACEPARENT"])
+
+	ended := spans.Ended()
+	require.Len(t, ended, 2)
+	step, job := ended[0], ended[1]
+	assert.Equal(t, "Run make", step.Name())
+	assert.Equal(t, jobSpan.SpanID(), step.Parent().SpanID())
+	assert.Equal(t, codes.Error, step.Status().Code)
+	assert.Subset(t, step.Attributes(), []any{
+		semconv.CICDPipelineTaskRunID("7.1"),
+		semconv.CICDPipelineTaskRunURLFull("https://gitea.example/owner/repo/actions/runs/42"),
+		semconv.ErrorTypeOther,
+	})
+	assert.Equal(t, "RUN ci.yml Build and test", job.Name())
+	assert.Equal(t, trace.SpanKindServer, job.SpanKind())
+	assert.Subset(t, job.Attributes(), []any{
+		semconv.VCSOwnerName("owner"),
+		semconv.VCSRepositoryName("repo"),
+		semconv.CICDPipelineRunID("42"),
+		attribute.String("gitea.run.number", "40"),
+		attribute.String("gitea.run.attempt", "2"),
+		attribute.String("gitea.job.key", "build"),
+		attribute.Int64("gitea.task.id", 7),
+	})
+}
+
+func TestResults(t *testing.T) {
+	cancelled, cancel := context.WithCancel(t.Context())
+	cancel()
+	expired, expire := context.WithDeadline(t.Context(), time.Time{})
+	defer expire()
+	for want, tc := range map[string]struct {
+		ctx    context.Context
+		err    error
+		result *model.StepResult
+	}{
+		"timeout":      {ctx: t.Context(), err: fmt.Errorf("exec: %w", context.DeadlineExceeded)},
+		"cancellation": {ctx: cancelled},
+		"failure":      {ctx: t.Context(), result: &model.StepResult{Outcome: model.StepStatusFailure, Conclusion: model.StepStatusFailure}},
+		"success":      {ctx: t.Context(), result: &model.StepResult{Outcome: model.StepStatusFailure, Conclusion: model.StepStatusSuccess}},
+		"skip":         {ctx: t.Context(), result: &model.StepResult{Outcome: model.StepStatusSkipped, Conclusion: model.StepStatusSkipped}},
+	} {
+		assert.Equal(t, want, stepResult(tc.ctx, tc.err, tc.result))
+	}
+	assert.Equal(t, "timeout", jobResult(expired, runnerv1.Result_RESULT_FAILURE))
+	assert.Equal(t, "cancellation", jobResult(t.Context(), runnerv1.Result_RESULT_CANCELLED))
+}