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 <[email protected]>
Co-authored-by: silverwind <[email protected]>
Co-authored-by: bircni <[email protected]>
Reviewed-on: https://gitea.com/gitea/runner/pulls/1248
Reviewed-by: silverwind <[email protected]>
Reviewed-by: bircni <[email protected]>
Co-authored-by: wmTJc9IK0Q <[email protected]>
This commit is contained in:
authored and bircni committed 2026-10-01 12:55:16 +00:00
1 parent d93b7d278a
commit 568f89d186
7 files changed
+482 -72

No files matched your search

+9 -3
View File
@@ -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)
}
+157
View File
@@ -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}}}
}
+161 -27
View File
@@ -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")
+132 -25
View File
@@ -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) {