From 568f89d186e941b9a50a866c4c73051a1f955acf Mon Sep 17 00:00:00 2001 From: wmTJc9IK0Q <231276+wmtjc9ik0q@noreply.gitea.com> Date: Thu, 1 Oct 2026 12:55:16 +0000 Subject: [PATCH] enhance: export the Prometheus metrics over OpenTelemetry (#1248) Push the runner's Prometheus metrics to the OTLP endpoint next to the job traces, so both land in the same backend. The `/metrics` registry is converted as the [Prometheus compatibility spec](https://opentelemetry.io/docs/specs/otel/compatibility/prometheus_and_openmetrics/) describes and sent through the existing gRPC-free client, so names stay unchanged and no second metrics stack is linked. Metrics are on by default and pushed every interval and once on shutdown, so ephemeral runners report too. New standard variables, each `OTEL_EXPORTER_OTLP_*` one with `TRACES` and `METRICS` variants: - `OTEL_METRICS_EXPORTER`, `none` turns metrics off - `OTEL_METRIC_EXPORT_INTERVAL`, `OTEL_METRIC_EXPORT_TIMEOUT` - `OTEL_EXPORTER_OTLP_TIMEOUT`, `OTEL_EXPORTER_OTLP_COMPRESSION` - `OTEL_EXPORTER_OTLP_CERTIFICATE`, `OTEL_EXPORTER_OTLP_CLIENT_CERTIFICATE`, `OTEL_EXPORTER_OTLP_CLIENT_KEY` --------- Co-authored-by: silverwind Co-authored-by: silverwind <2021+silverwind@noreply.gitea.com> Co-authored-by: bircni Reviewed-on: https://gitea.com/gitea/runner/pulls/1248 Reviewed-by: silverwind <2021+silverwind@noreply.gitea.com> Reviewed-by: bircni Co-authored-by: wmTJc9IK0Q <231276+wmtjc9ik0q@noreply.gitea.com> --- README.md | 2 +- docs/telemetry.md | 26 ++-- internal/app/cmd/daemon.go | 12 +- internal/pkg/metrics/metrics.go | 12 +- internal/pkg/telemetry/metrics.go | 157 +++++++++++++++++++ internal/pkg/telemetry/telemetry.go | 188 +++++++++++++++++++---- internal/pkg/telemetry/telemetry_test.go | 157 ++++++++++++++++--- 7 files changed, 482 insertions(+), 72 deletions(-) create mode 100644 internal/pkg/telemetry/metrics.go diff --git a/README.md b/README.md index 2667f8ee..9db9fecd 100644 --- a/README.md +++ b/README.md @@ -382,7 +382,7 @@ See **[docs/job-hooks.md](docs/job-hooks.md)** for the execution order, environm #### OpenTelemetry (`OTEL_*`) -Set `OTEL_EXPORTER_OTLP_ENDPOINT` in the runner's environment to send a trace of every job to an OpenTelemetry collector. See **[docs/telemetry.md](docs/telemetry.md)**. +Set `OTEL_EXPORTER_OTLP_ENDPOINT` in the runner's environment to send a trace of every job, and the runner's metrics, to an OpenTelemetry collector. See **[docs/telemetry.md](docs/telemetry.md)**. #### Local job logs (`log.job.dir`) diff --git a/docs/telemetry.md b/docs/telemetry.md index 9518cf70..9f3951d0 100644 --- a/docs/telemetry.md +++ b/docs/telemetry.md @@ -1,6 +1,6 @@ -# OpenTelemetry tracing +# OpenTelemetry -The runner can send a trace of every job it runs to an OpenTelemetry collector or any backend that accepts OTLP. Each job becomes a trace with a span per step, so you can see where CI time goes and which steps fail. Tracing stays off until you set an endpoint. +The runner can send a trace of every job it runs, and its metrics, to an OpenTelemetry collector or any backend that accepts OTLP. Each job becomes a trace with a span per step, so you can see where CI time goes and which steps fail. Export stays off until you set an endpoint. ## Setup @@ -12,14 +12,20 @@ OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4318 | Variable | Purpose | | --- | --- | -| `OTEL_EXPORTER_OTLP_ENDPOINT` | Collector base URL, traces go to `/v1/traces` under it | -| `OTEL_EXPORTER_OTLP_TRACES_ENDPOINT` | Full traces URL, takes precedence over the base URL | -| `OTEL_EXPORTER_OTLP_HEADERS`, `OTEL_EXPORTER_OTLP_TRACES_HEADERS` | Request headers such as an API key, as comma-separated `key=value` pairs with URL-encoded values | +| `OTEL_EXPORTER_OTLP_ENDPOINT` | Collector base URL, traces go to `/v1/traces` and metrics to `/v1/metrics` under it | +| `OTEL_EXPORTER_OTLP_TRACES_ENDPOINT`, `OTEL_EXPORTER_OTLP_METRICS_ENDPOINT` | Full URL per signal, takes precedence over the base URL | +| `OTEL_EXPORTER_OTLP_HEADERS` | Request headers such as an API key, as comma-separated `key=value` pairs with URL-encoded values | +| `OTEL_EXPORTER_OTLP_TIMEOUT` | Export timeout in milliseconds, default 10000, 0 for none | +| `OTEL_EXPORTER_OTLP_COMPRESSION` | `gzip` or `none`, default `none` | +| `OTEL_EXPORTER_OTLP_CERTIFICATE` | PEM file of the CA to trust instead of the system ones | +| `OTEL_EXPORTER_OTLP_CLIENT_CERTIFICATE`, `OTEL_EXPORTER_OTLP_CLIENT_KEY` | PEM files of the client certificate and key for mTLS | | `OTEL_SERVICE_NAME`, `OTEL_RESOURCE_ATTRIBUTES` | Resource attributes, the service name defaults to `gitea-runner` | | `OTEL_TRACES_SAMPLER`, `OTEL_BSP_*` | Sampling and batching, see the [SDK variables](https://opentelemetry.io/docs/specs/otel/configuration/sdk-environment-variables/) | -| `OTEL_SDK_DISABLED=true`, `OTEL_TRACES_EXPORTER=none` | Turn tracing off | +| `OTEL_METRIC_EXPORT_INTERVAL`, `OTEL_METRIC_EXPORT_TIMEOUT` | Metrics export interval and timeout in milliseconds, default 60000 and 30000, a timeout of 0 for none | +| `OTEL_SDK_DISABLED=true` | Turn all export off | +| `OTEL_TRACES_EXPORTER=none`, `OTEL_METRICS_EXPORTER=none` | Turn traces or metrics off | -Only the `http/protobuf` protocol is supported. +Except for the endpoint, each `OTEL_EXPORTER_OTLP_*` variable has `TRACES` and `METRICS` variants such as `OTEL_EXPORTER_OTLP_METRICS_HEADERS`, which take precedence for that signal. Only the `otlp` exporter over the `http/protobuf` protocol is supported. ## What is sent @@ -35,12 +41,12 @@ Jobs receive `TRACEPARENT`, so tools in a job that use OpenTelemetry can add the ## Metrics -Metrics stay on the Prometheus endpoint enabled by `metrics.enabled`. To get them into OpenTelemetry, scrape `/metrics` with the collector's `prometheus` receiver. +The metrics of the Prometheus endpoint are also exported, under the same names and whether or not `metrics.enabled` is set. They are sent every export interval and once more on shutdown. If your collector only accepts traces, set `OTEL_METRICS_EXPORTER=none`. ## Security -- OTLP has no authentication of its own. Send to a collector on the runner's host that holds the backend credentials, as the [OpenTelemetry security guide](https://opentelemetry.io/docs/security/config-best-practices/) recommends, or use `https` with a token in `OTEL_EXPORTER_OTLP_HEADERS`. Over plain `http`, headers travel unencrypted. -- TLS certificates are verified against the system CA bundle, set `SSL_CERT_FILE` to use your own. Client certificates are not supported, use a local collector for mTLS. Redirects are not followed, so headers only reach the configured endpoint. +- OTLP has no authentication of its own. Send to a collector on the runner's host that holds the backend credentials, as the [OpenTelemetry security guide](https://opentelemetry.io/docs/security/config-best-practices/) recommends, or use `https` with a token in `OTEL_EXPORTER_OTLP_HEADERS` or a client certificate. Over plain `http`, headers travel unencrypted. +- TLS certificates are verified against the system CA bundle, or only against `OTEL_EXPORTER_OTLP_CERTIFICATE` when set. Redirects are not followed, so headers only reach the configured endpoint. - Jobs in containers never see the runner's `OTEL_*` variables. Jobs on the host inherit the runner's whole environment, headers included, and can read its files anyway since they run as the same user. - Every workflow can read `runner.envs`, so never put credentials there. Point jobs at a collector that holds them instead, keeping in mind that any workflow can then send data through it. - Traces carry the repository and step names of every job, private repositories included, so give the collector the same access controls as the runner's logs. diff --git a/internal/app/cmd/daemon.go b/internal/app/cmd/daemon.go index 13257c56..c86e7941 100644 --- a/internal/app/cmd/daemon.go +++ b/internal/app/cmd/daemon.go @@ -72,7 +72,7 @@ func runDaemon(ctx context.Context, daemArgs *daemonArgs, configFile *string) fu defer func() { _ = releaseLock() }() } - shutdownTelemetry, err := telemetry.Setup(ctx, reg.UUID, reg.Name) + shutdownTelemetry, err := telemetry.Setup(ctx, reg.UUID, reg.Name, metrics.Registry) if err != nil { log.WithError(err).Warn("OpenTelemetry export failed to start") } @@ -203,12 +203,12 @@ func runDaemon(ctx context.Context, daemArgs *daemonArgs, configFile *string) fu poller := poll.New(cfg, cli, runner) + metrics.Init() // OTLP exports the registry even without the server + metrics.RunnerInfo.WithLabelValues(ver.Version(), resp.Msg.Runner.Name).Set(1) + metrics.RunnerCapacity.Set(float64(cfg.Runner.Capacity)) + metrics.RegisterUptimeFunc(time.Now()) + metrics.RegisterRunningJobsFunc(runner.RunningCount, cfg.Runner.Capacity) if cfg.Metrics.Enabled { - metrics.Init() - metrics.RunnerInfo.WithLabelValues(ver.Version(), resp.Msg.Runner.Name).Set(1) - metrics.RunnerCapacity.Set(float64(cfg.Runner.Capacity)) - metrics.RegisterUptimeFunc(time.Now()) - metrics.RegisterRunningJobsFunc(runner.RunningCount, cfg.Runner.Capacity) metrics.StartServer(ctx, cfg.Metrics.Addr, func() (bool, string) { return poller.Ready(cfg.Metrics.ReadinessGrace) }) diff --git a/internal/pkg/metrics/metrics.go b/internal/pkg/metrics/metrics.go index 1a193a1f..1603d77c 100644 --- a/internal/pkg/metrics/metrics.go +++ b/internal/pkg/metrics/metrics.go @@ -176,7 +176,7 @@ func Init() { // RegisterUptimeFunc registers a GaugeFunc that reports seconds since startTime. func RegisterUptimeFunc(startTime time.Time) { - Registry.MustRegister(prometheus.NewGaugeFunc( + register(prometheus.NewGaugeFunc( prometheus.GaugeOpts{ Namespace: Namespace, Name: "uptime_seconds", @@ -190,7 +190,7 @@ func RegisterUptimeFunc(startTime time.Time) { // capacity utilisation ratio, evaluated lazily at Prometheus scrape time. func RegisterRunningJobsFunc(countFn func() int64, capacity int) { capF := float64(capacity) - Registry.MustRegister(prometheus.NewGaugeFunc( + register(prometheus.NewGaugeFunc( prometheus.GaugeOpts{ Namespace: Namespace, Subsystem: "job", @@ -199,7 +199,7 @@ func RegisterRunningJobsFunc(countFn func() int64, capacity int) { }, func() float64 { return float64(countFn()) }, )) - Registry.MustRegister(prometheus.NewGaugeFunc( + register(prometheus.NewGaugeFunc( prometheus.GaugeOpts{ Namespace: Namespace, Subsystem: "job", @@ -214,3 +214,9 @@ func RegisterRunningJobsFunc(countFn func() int64, capacity int) { }, )) } + +// register replaces an earlier registration, as the daemon can run more than once per process. +func register(collector prometheus.Collector) { + Registry.Unregister(collector) + Registry.MustRegister(collector) +} diff --git a/internal/pkg/telemetry/metrics.go b/internal/pkg/telemetry/metrics.go new file mode 100644 index 00000000..5943debb --- /dev/null +++ b/internal/pkg/telemetry/metrics.go @@ -0,0 +1,157 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +package telemetry + +import ( + "cmp" + "context" + "errors" + "strings" + "time" + + "github.com/prometheus/client_golang/prometheus" + dto "github.com/prometheus/client_model/go" + log "github.com/sirupsen/logrus" + "go.opentelemetry.io/otel/sdk/resource" + commonpb "go.opentelemetry.io/proto/otlp/common/v1" + metricpb "go.opentelemetry.io/proto/otlp/metrics/v1" + resourcepb "go.opentelemetry.io/proto/otlp/resource/v1" + "google.golang.org/protobuf/types/known/timestamppb" +) + +const cumulative = metricpb.AggregationTemporality_AGGREGATION_TEMPORALITY_CUMULATIVE + +// pushMetrics also pushes on shutdown so short-lived runners report. +func pushMetrics(exporter *client, res *resource.Resource, gatherer prometheus.Gatherer) func(context.Context) error { + attrs := make([]*commonpb.KeyValue, 0, res.Len()) + for _, kv := range res.Attributes() { + attrs = append(attrs, keyValue(string(kv.Key), kv.Value.String())) + } + interval, timeout := millis("OTEL_METRIC_EXPORT_INTERVAL", time.Minute, time.Minute), millis("OTEL_METRIC_EXPORT_TIMEOUT", 30*time.Second, noLimit) + push := func(ctx context.Context) error { + ctx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + families, err := gatherer.Gather() + return errors.Join(err, exporter.upload(ctx, &metricpb.MetricsData{ResourceMetrics: []*metricpb.ResourceMetrics{{ + Resource: &resourcepb.Resource{Attributes: attrs}, + ScopeMetrics: []*metricpb.ScopeMetrics{{Scope: &commonpb.InstrumentationScope{Name: scope}, Metrics: convert(families)}}, + SchemaUrl: res.SchemaURL(), + }}})) + } + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + defer close(done) + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + if err := push(ctx); err != nil && ctx.Err() == nil { + log.WithError(err).Warn("OTLP metrics export failed, set OTEL_METRICS_EXPORTER=none to disable it") + } + } + } + }() + return func(ctx context.Context) error { + cancel() + <-done + return push(ctx) + } +} + +var ucum = map[string]string{ + "days": "d", "hours": "h", "minutes": "min", "seconds": "s", "milliseconds": "ms", "microseconds": "us", "nanoseconds": "ns", + "bytes": "By", "kibibytes": "KiBy", "mebibytes": "MiBy", "gibibytes": "GiBy", "tebibytes": "TiBy", + "kilobytes": "kBy", "megabytes": "MBy", "gigabytes": "GBy", "terabytes": "TBy", + "meters": "m", "volts": "V", "amperes": "A", "joules": "J", "watts": "W", "grams": "g", "celsius": "Cel", "hertz": "Hz", "percent": "%", +} + +// convert follows https://opentelemetry.io/docs/specs/otel/compatibility/prometheus_and_openmetrics/#prometheus-metric-points-to-otlp +func convert(families []*dto.MetricFamily) []*metricpb.Metric { + now := uint64(time.Now().UnixNano()) + metrics := make([]*metricpb.Metric, 0, len(families)) + for _, family := range families { + metric := &metricpb.Metric{ + Name: family.GetName(), Description: family.GetHelp(), Unit: cmp.Or(ucum[family.GetUnit()], family.GetUnit()), + Metadata: []*commonpb.KeyValue{keyValue("prometheus.type", strings.ToLower(family.GetType().String()))}, + } + switch family.GetType() { + case dto.MetricType_COUNTER: + sum := &metricpb.Sum{AggregationTemporality: cumulative, IsMonotonic: true} + for _, series := range family.GetMetric() { + sum.DataPoints = append(sum.DataPoints, number(series, unixNano(series.GetCounter().GetCreatedTimestamp()), now, series.GetCounter().GetValue())) + } + metric.Data = &metricpb.Metric_Sum{Sum: sum} + case dto.MetricType_GAUGE: + gauge := &metricpb.Gauge{} + for _, series := range family.GetMetric() { + gauge.DataPoints = append(gauge.DataPoints, number(series, 0, now, series.GetGauge().GetValue())) + } + metric.Data = &metricpb.Metric_Gauge{Gauge: gauge} + case dto.MetricType_HISTOGRAM: + histogram := &metricpb.Histogram{AggregationTemporality: cumulative} + for _, series := range family.GetMetric() { + histogram.DataPoints = append(histogram.DataPoints, histogramPoint(series, now)) + } + metric.Data = &metricpb.Metric_Histogram{Histogram: histogram} + case dto.MetricType_SUMMARY: + summary := &metricpb.Summary{} + for _, series := range family.GetMetric() { + point := &metricpb.SummaryDataPoint{ + Attributes: labels(series), StartTimeUnixNano: unixNano(series.GetSummary().GetCreatedTimestamp()), TimeUnixNano: now, + Count: series.GetSummary().GetSampleCount(), Sum: series.GetSummary().GetSampleSum(), + } + for _, quantile := range series.GetSummary().GetQuantile() { + point.QuantileValues = append(point.QuantileValues, &metricpb.SummaryDataPoint_ValueAtQuantile{Quantile: quantile.GetQuantile(), Value: quantile.GetValue()}) + } + summary.DataPoints = append(summary.DataPoints, point) + } + metric.Data = &metricpb.Metric_Summary{Summary: summary} + default: + continue + } + metrics = append(metrics, metric) + } + return metrics +} + +func number(series *dto.Metric, start, now uint64, value float64) *metricpb.NumberDataPoint { + return &metricpb.NumberDataPoint{Attributes: labels(series), StartTimeUnixNano: start, TimeUnixNano: now, Value: &metricpb.NumberDataPoint_AsDouble{AsDouble: value}} +} + +// unixNano maps a nil timestamp to 0, leaving the start time unset as the spec asks. +func unixNano(timestamp *timestamppb.Timestamp) uint64 { + return uint64(timestamp.AsTime().UnixNano()) +} + +func histogramPoint(series *dto.Metric, now uint64) *metricpb.HistogramDataPoint { + histogram := series.GetHistogram() + point := &metricpb.HistogramDataPoint{ + Attributes: labels(series), StartTimeUnixNano: unixNano(histogram.GetCreatedTimestamp()), TimeUnixNano: now, + Count: histogram.GetSampleCount(), Sum: histogram.SampleSum, + BucketCounts: make([]uint64, len(histogram.GetBucket())+1), + } + var below uint64 + for i, bucket := range histogram.GetBucket() { + point.ExplicitBounds = append(point.ExplicitBounds, bucket.GetUpperBound()) + point.BucketCounts[i], below = bucket.GetCumulativeCount()-below, bucket.GetCumulativeCount() + } + point.BucketCounts[len(histogram.GetBucket())] = histogram.GetSampleCount() - below + return point +} + +func labels(series *dto.Metric) []*commonpb.KeyValue { + attrs := make([]*commonpb.KeyValue, 0, len(series.GetLabel())) + for _, label := range series.GetLabel() { + attrs = append(attrs, keyValue(label.GetName(), label.GetValue())) + } + return attrs +} + +func keyValue(key, value string) *commonpb.KeyValue { + return &commonpb.KeyValue{Key: key, Value: &commonpb.AnyValue{Value: &commonpb.AnyValue_StringValue{StringValue: value}}} +} diff --git a/internal/pkg/telemetry/telemetry.go b/internal/pkg/telemetry/telemetry.go index 3af83d2d..909a1c35 100644 --- a/internal/pkg/telemetry/telemetry.go +++ b/internal/pkg/telemetry/telemetry.go @@ -6,22 +6,29 @@ package telemetry import ( "bytes" "cmp" + "compress/gzip" "context" "errors" "fmt" "io" + "math" "net/http" "net/url" "os" "slices" "strconv" "strings" + "sync" + "time" "gitea.com/gitea/runner/internal/pkg/ver" "gitea.dev/actionslib/pkg/model" runnerv1 "gitea.dev/actionslib/runner/v1" "github.com/avast/retry-go/v5" + "github.com/docker/go-connections/tlsconfig" + "github.com/prometheus/client_golang/prometheus" + log "github.com/sirupsen/logrus" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/codes" "go.opentelemetry.io/otel/exporters/otlp/otlptrace" @@ -32,6 +39,7 @@ import ( "go.opentelemetry.io/otel/trace" tracenoop "go.opentelemetry.io/otel/trace/noop" tracepb "go.opentelemetry.io/proto/otlp/trace/v1" + "golang.org/x/sync/errgroup" "google.golang.org/protobuf/proto" ) @@ -51,17 +59,48 @@ func SetTracerProvider(provider trace.TracerProvider) { tracer = provider.Tracer(scope) } -func Setup(ctx context.Context, uuid, name string) (func(context.Context) error, error) { +func Setup(ctx context.Context, uuid, name string, gatherer prometheus.Gatherer) (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") { + if strings.EqualFold(os.Getenv("OTEL_SDK_DISABLED"), "true") { 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) + clients := map[string]*client{} + for _, signal := range []string{"TRACES", "METRICS"} { + endpoint := os.Getenv("OTEL_EXPORTER_OTLP_" + signal + "_ENDPOINT") + if base := os.Getenv("OTEL_EXPORTER_OTLP_ENDPOINT"); endpoint == "" && base != "" { + endpoint = strings.TrimSuffix(base, "/") + "/v1/" + strings.ToLower(signal) + } + if endpoint == "" || !exportsOTLP(signal) { + continue + } + if protocol := os.Getenv(otlpKey(signal, "PROTOCOL")); protocol != "" && !strings.EqualFold(protocol, "http/protobuf") { + return noop, fmt.Errorf("OTLP protocol %q is not supported, only http/protobuf is", protocol) + } + headers := http.Header{} + for _, key := range []string{"OTEL_EXPORTER_OTLP_HEADERS", "OTEL_EXPORTER_OTLP_" + signal + "_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) + } + } + } + compressionKey := otlpKey(signal, "COMPRESSION") + compression := os.Getenv(compressionKey) + if !slices.Contains([]string{"", "none", "gzip"}, strings.ToLower(compression)) { + log.Warnf("ignoring unsupported %s=%q", compressionKey, compression) + } + httpClient, err := newHTTPClient(signal) + if err != nil { + return noop, err + } + clients[signal] = &client{ + endpoint: endpoint, headers: headers, timeout: millis(otlpKey(signal, "TIMEOUT"), 10*time.Second, noLimit), + gzip: strings.EqualFold(compression, "gzip"), httpClient: httpClient, + } + } + if len(clients) == 0 { + return noop, nil } res, err := resource.New(ctx, resource.WithTelemetrySDK(), resource.WithAttributes( semconv.ServiceName("gitea-runner"), @@ -73,26 +112,90 @@ func Setup(ctx context.Context, uuid, name string) (func(context.Context) error, 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) - } - } + var shutdown []func(context.Context) error + if c := clients["TRACES"]; c != nil { + provider := sdktrace.NewTracerProvider(sdktrace.WithBatcher(otlptrace.NewUnstarted(c)), sdktrace.WithResource(res)) + SetTracerProvider(provider) + shutdown = append(shutdown, provider.Shutdown) } - provider := sdktrace.NewTracerProvider(sdktrace.WithBatcher(otlptrace.NewUnstarted(&client{endpoint, headers})), sdktrace.WithResource(res)) - SetTracerProvider(provider) - return provider.Shutdown, nil + if c := clients["METRICS"]; c != nil { + shutdown = append(shutdown, pushMetrics(c, res, gatherer)) + } + return func(ctx context.Context) error { + var group errgroup.Group + for _, fn := range shutdown { + group.Go(func() error { return fn(ctx) }) + } + return group.Wait() + }, 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 +func exportsOTLP(signal string) bool { + key := "OTEL_" + signal + "_EXPORTER" + otlp, known := false, false + for value := range strings.SplitSeq(os.Getenv(key), ",") { + switch value = strings.TrimSpace(value); strings.ToLower(value) { + case "": + case "otlp": + otlp, known = true, true + case "none": + known = true + default: + log.Warnf("ignoring unsupported %s value %q", key, value) + } + } + return otlp || !known } +func otlpKey(signal, name string) string { + if key := "OTEL_EXPORTER_OTLP_" + signal + "_" + name; os.Getenv(key) != "" { + return key + } + return "OTEL_EXPORTER_OTLP_" + name +} + +const noLimit = math.MaxInt32 * time.Millisecond // the spec's stand-in for a 0 timeout + +func millis(key string, fallback, zero time.Duration) time.Duration { + value := os.Getenv(key) + if value == "" { + return fallback + } + ms, err := strconv.ParseInt(value, 10, 32) + if err != nil || ms < 0 { + log.Warnf("ignoring %s=%q, it must be a non-negative number of milliseconds", key, value) + return fallback + } + return cmp.Or(time.Duration(ms)*time.Millisecond, zero) +} + +// newHTTPClient applies the OTLP certificate variables, a configured CA replaces the system roots like in the Go SDK. +func newHTTPClient(signal string) (*http.Client, error) { + caFile, certFile, keyFile := os.Getenv(otlpKey(signal, "CERTIFICATE")), os.Getenv(otlpKey(signal, "CLIENT_CERTIFICATE")), os.Getenv(otlpKey(signal, "CLIENT_KEY")) + if caFile == "" && certFile == "" && keyFile == "" { + return exportClient, nil + } + config, err := tlsconfig.Client(tlsconfig.Options{CAFile: caFile, CertFile: certFile, KeyFile: keyFile, ExclusiveRootPools: true}) + if err != nil { + return nil, err + } + transport, _ := http.DefaultTransport.(*http.Transport) + transport = transport.Clone() + transport.TLSClientConfig = config + return &http.Client{Transport: transport, CheckRedirect: exportClient.CheckRedirect}, nil +} + +// client replaces the otlp*http exporters, which link gRPC (+7 MB): https://github.com/open-telemetry/opentelemetry-go/issues/2579 +type client struct { + endpoint string + headers http.Header + timeout time.Duration + gzip bool + httpClient *http.Client +} + +var gzipWriters = sync.Pool{New: func() any { return gzip.NewWriter(io.Discard) }} + // 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 }} @@ -101,18 +204,36 @@ 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}) + return c.upload(ctx, &tracepb.TracesData{ResourceSpans: spans}) +} + +func (c *client) upload(ctx context.Context, data proto.Message) error { + body, err := proto.Marshal(data) if err != nil { return err } - return retry.New(retry.Context(ctx), retry.Attempts(5), retry.LastErrorOnly(true)).Do(func() error { + if c.gzip { + var buffer bytes.Buffer + writer, _ := gzipWriters.Get().(*gzip.Writer) + writer.Reset(&buffer) + _, _ = writer.Write(body) + _ = writer.Close() + gzipWriters.Put(writer) + body = buffer.Bytes() + } + ctx, cancel := context.WithTimeout(ctx, c.timeout) + defer cancel() + return retry.New(retry.Context(ctx), retry.Attempts(5), retry.LastErrorOnly(true), retry.DelayType(retryDelay)).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 c.gzip { + req.Header.Set("Content-Encoding", "gzip") + } + resp, err := c.httpClient.Do(req) if err != nil { return err } @@ -121,7 +242,7 @@ func (c *client) UploadTraces(ctx context.Context, spans []*tracepb.ResourceSpan if resp.StatusCode < http.StatusMultipleChoices { return nil } - err = fmt.Errorf("OTLP export failed: %s", resp.Status) + err = &statusError{resp.Status, resp.Header.Get("Retry-After")} if slices.Contains([]int{http.StatusTooManyRequests, http.StatusBadGateway, http.StatusServiceUnavailable, http.StatusGatewayTimeout}, resp.StatusCode) { return err } @@ -129,6 +250,19 @@ func (c *client) UploadTraces(ctx context.Context, spans []*tracepb.ResourceSpan }) } +type statusError struct{ status, retryAfter string } + +func (e *statusError) Error() string { return "OTLP export failed: " + e.status } + +func retryDelay(attempt uint, err error, config retry.DelayContext) time.Duration { + if status := (*statusError)(nil); errors.As(err, &status) { + if seconds, err := strconv.ParseUint(status.retryAfter, 10, 31); err == nil { + return time.Duration(seconds) * time.Second + } + } + return retry.CombineDelay(retry.BackOffDelay, retry.RandomDelay)(attempt, err, config) +} + 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") diff --git a/internal/pkg/telemetry/telemetry_test.go b/internal/pkg/telemetry/telemetry_test.go index 556e6ab8..46d65d7a 100644 --- a/internal/pkg/telemetry/telemetry_test.go +++ b/internal/pkg/telemetry/telemetry_test.go @@ -4,16 +4,27 @@ package telemetry import ( + "bytes" + "compress/gzip" "context" + "crypto/tls" + "crypto/x509" + "encoding/pem" "fmt" "io" "net/http" "net/http/httptest" + "os" + "path/filepath" + "slices" + "strings" "testing" "time" "gitea.dev/actionslib/pkg/model" runnerv1 "gitea.dev/actionslib/runner/v1" + "github.com/prometheus/client_golang/prometheus" + dto "github.com/prometheus/client_model/go" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "go.opentelemetry.io/otel/attribute" @@ -23,6 +34,8 @@ import ( semconv "go.opentelemetry.io/otel/semconv/v1.43.0" "go.opentelemetry.io/otel/trace" tracenoop "go.opentelemetry.io/otel/trace/noop" + commonpb "go.opentelemetry.io/proto/otlp/common/v1" + metricpb "go.opentelemetry.io/proto/otlp/metrics/v1" tracepb "go.opentelemetry.io/proto/otlp/trace/v1" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/structpb" @@ -30,66 +43,153 @@ import ( 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]) + for _, variable := range os.Environ() { + if key, _, _ := strings.Cut(variable, "="); strings.HasPrefix(key, "OTEL_") { + t.Setenv(key, "") + } + } + for key, value := range env { + t.Setenv(key, value) } } func TestSetup(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) {})) + defer server.Close() for _, tc := range []struct { - env map[string]string - enabled bool - err string + env map[string]string + traces, metrics 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_TRACES_ENDPOINT": "http://collector:4318/v1/traces"}, traces: true}, + {env: map[string]string{"OTEL_EXPORTER_OTLP_ENDPOINT": server.URL, "OTEL_TRACES_EXPORTER": "NONE"}, metrics: true}, + {env: map[string]string{"OTEL_EXPORTER_OTLP_ENDPOINT": server.URL, "OTEL_TRACES_EXPORTER": "zipkin, OTLP", "OTEL_METRICS_EXPORTER": "console"}, traces: true, metrics: true}, + {env: map[string]string{"OTEL_EXPORTER_OTLP_ENDPOINT": "http://collector:4318", "OTEL_TRACES_EXPORTER": "NONE", "OTEL_METRICS_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") + gathered := false + shutdown, err := Setup(t.Context(), "uuid", "runner", prometheus.GathererFunc(func() ([]*dto.MetricFamily, error) { + gathered = true + return nil, nil + })) if tc.err != "" { require.EqualError(t, err, tc.err) } else { require.NoError(t, err) } _, disabled := tracer.(tracenoop.Tracer) - assert.Equal(t, tc.enabled, !disabled) + assert.Equal(t, tc.traces, !disabled) require.NoError(t, shutdown(t.Context())) + assert.Equal(t, tc.metrics, gathered) }) } } 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) + traceBodies, metricBodies := make(chan []byte, 2), make(chan []byte, 1) + server := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + assert.Len(t, r.TLS.PeerCertificates, 1) assert.Equal(t, "Bearer token", r.Header.Get("Authorization")) + assert.Equal(t, "gzip", r.Header.Get("Content-Encoding")) body, _ := io.ReadAll(r.Body) - bodies <- body - if len(bodies) == 1 { + if r.URL.Path == "/otlp/v1/metrics" { + select { + case metricBodies <- body: + default: + } + return + } + assert.Equal(t, "/otlp/v1/traces", r.URL.Path) + traceBodies <- body + if len(traceBodies) == 1 { w.WriteHeader(http.StatusServiceUnavailable) } })) + server.TLS = &tls.Config{ClientAuth: tls.RequireAnyClientCert} + server.StartTLS() defer server.Close() - setEnv(t, map[string]string{"OTEL_EXPORTER_OTLP_ENDPOINT": server.URL + "/otlp/", "OTEL_EXPORTER_OTLP_HEADERS": "authorization=Bearer%20token"}) + certificate := server.TLS.Certificates[0] + key, err := x509.MarshalPKCS8PrivateKey(certificate.PrivateKey) + require.NoError(t, err) + certificatePath, keyPath := filepath.Join(t.TempDir(), "cert.pem"), filepath.Join(t.TempDir(), "key.pem") + require.NoError(t, os.WriteFile(certificatePath, pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: certificate.Certificate[0]}), 0o600)) + require.NoError(t, os.WriteFile(keyPath, pem.EncodeToMemory(&pem.Block{Type: "PRIVATE KEY", Bytes: key}), 0o600)) + setEnv(t, map[string]string{ + "OTEL_EXPORTER_OTLP_ENDPOINT": server.URL + "/otlp/", "OTEL_EXPORTER_OTLP_HEADERS": "authorization=Bearer%20token", + "OTEL_EXPORTER_OTLP_COMPRESSION": "gzip", "OTEL_METRIC_EXPORT_INTERVAL": "1", "OTEL_EXPORTER_OTLP_CERTIFICATE": certificatePath, + "OTEL_EXPORTER_OTLP_CLIENT_CERTIFICATE": certificatePath, "OTEL_EXPORTER_OTLP_CLIENT_KEY": keyPath, + }) + registry := prometheus.NewRegistry() + registry.MustRegister(prometheus.NewGaugeFunc(prometheus.GaugeOpts{Name: "running"}, func() float64 { return 2 })) - shutdown, err := Setup(t.Context(), "uuid", "runner") + shutdown, err := Setup(t.Context(), "uuid", "runner", registry) require.NoError(t, err) _, span := tracer.Start(t.Context(), "op") span.End() + gunzip := func(body []byte) []byte { + reader, err := gzip.NewReader(bytes.NewReader(body)) + require.NoError(t, err) + decoded, err := io.ReadAll(reader) + require.NoError(t, err) + return decoded + } + var metricsData metricpb.MetricsData + select { + case body := <-metricBodies: + require.NoError(t, proto.Unmarshal(gunzip(body), &metricsData)) + case <-time.After(30 * time.Second): + require.FailNow(t, "no periodic metrics export") + } 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()) + require.Len(t, traceBodies, 2) + <-traceBodies + var tracesData tracepb.TracesData + require.NoError(t, proto.Unmarshal(gunzip(<-traceBodies), &tracesData)) + assert.Equal(t, 2*time.Second, retryDelay(1, &statusError{"429 Too Many Requests", "2"}, nil)) + assert.Equal(t, "op", tracesData.GetResourceSpans()[0].GetScopeSpans()[0].GetSpans()[0].GetName()) + + resourceMetrics := metricsData.GetResourceMetrics()[0] + assert.True(t, slices.ContainsFunc(resourceMetrics.GetResource().GetAttributes(), func(kv *commonpb.KeyValue) bool { + return proto.Equal(kv, keyValue("service.name", "gitea-runner")) + })) + assert.InDelta(t, 2, resourceMetrics.GetScopeMetrics()[0].GetMetrics()[0].GetGauge().GetDataPoints()[0].GetAsDouble(), 0) +} + +func TestConvert(t *testing.T) { + registry := prometheus.NewRegistry() + counter := prometheus.NewCounterVec(prometheus.CounterOpts{Name: "jobs_total", Help: "Jobs."}, []string{"status"}) + histogram := prometheus.NewHistogram(prometheus.HistogramOpts{Name: "duration_seconds", Unit: "seconds", Buckets: []float64{1, 2}}) + summary := prometheus.NewSummary(prometheus.SummaryOpts{Name: "gc_seconds", Objectives: map[float64]float64{0.5: 0.05}}) + registry.MustRegister(counter, histogram, summary) + counter.WithLabelValues("success").Add(3) + for _, value := range []float64{0.5, 1.5, 3} { + histogram.Observe(value) + } + summary.Observe(4) + families, err := registry.Gather() + require.NoError(t, err) + + exported := convert(families) + assert.Equal(t, "s", exported[0].GetUnit()) + buckets := exported[0].GetHistogram().GetDataPoints()[0] + assert.Equal(t, []float64{1, 2}, buckets.GetExplicitBounds()) + assert.Equal(t, []uint64{1, 1, 1}, buckets.GetBucketCounts()) + assert.Equal(t, uint64(3), buckets.GetCount()) + assert.InDelta(t, 5, buckets.GetSum(), 0) + assert.InDelta(t, 4, exported[1].GetSummary().GetDataPoints()[0].GetQuantileValues()[0].GetValue(), 0) + sum := exported[2].GetSum() + assert.Equal(t, "Jobs.", exported[2].GetDescription()) + assert.True(t, proto.Equal(keyValue("prometheus.type", "counter"), exported[2].GetMetadata()[0])) + assert.NotZero(t, sum.GetDataPoints()[0].GetStartTimeUnixNano()) + assert.True(t, sum.GetIsMonotonic()) + assert.Equal(t, cumulative, sum.GetAggregationTemporality()) + assert.InDelta(t, 3, sum.GetDataPoints()[0].GetAsDouble(), 0) + assert.True(t, proto.Equal(keyValue("status", "success"), sum.GetDataPoints()[0].GetAttributes()[0])) } func TestExportRefusesRedirects(t *testing.T) { @@ -97,7 +197,14 @@ func TestExportRefusesRedirects(t *testing.T) { 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") + require.EqualError(t, (&client{endpoint: origin.URL, headers: http.Header{}, timeout: time.Second, httpClient: exportClient}).UploadTraces(t.Context(), nil), "OTLP export failed: 307 Temporary Redirect") +} + +func TestMillis(t *testing.T) { + for value, want := range map[string]time.Duration{"": time.Minute, "1500": 1500 * time.Millisecond, "0": noLimit, "-1": time.Minute, "1s": time.Minute, "2147483648": time.Minute} { + t.Setenv("OTEL_METRIC_EXPORT_TIMEOUT", value) + assert.Equal(t, want, millis("OTEL_METRIC_EXPORT_TIMEOUT", time.Minute, noLimit), value) + } } func TestJob(t *testing.T) {