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) {