diff --git a/buildkitd/buildkitd.go b/buildkitd/buildkitd.go index c0ef3cc5ed..97012ee236 100644 --- a/buildkitd/buildkitd.go +++ b/buildkitd/buildkitd.go @@ -476,7 +476,7 @@ func RemoveExited(ctx context.Context, fe containerutil.ContainerFrontend, conta func Start( ctx context.Context, console conslogging.ConsoleLogger, - image, containerName, _ string, + image, containerName, installationName string, fe containerutil.ContainerFrontend, settings Settings, reset bool, @@ -507,6 +507,9 @@ func Start( "BUILDKIT_MAX_PARALLELISM": strconv.Itoa(settings.MaxParallelism), } + withDocker, _ := strconv.ParseBool(os.Getenv("EARTHLY_WITH_DOCKER")) + addBuildkitTelemetryEnv(envOpts, containerName, installationName, withDocker) + labelOpts := map[string]string{ "dev.earthly.settingshash": settingsHash, } @@ -532,8 +535,6 @@ func Start( const localhost = "127.0.0.1" - withDocker, _ := strconv.ParseBool(os.Getenv("EARTHLY_WITH_DOCKER")) - //nolint:nestif // TODO(jhorsts): simplify if withDocker { // Add /sys/fs/cgroup if it's earth-in-earth. @@ -686,6 +687,77 @@ func Start( return nil } +func addBuildkitTelemetryEnv(envOpts map[string]string, containerName, installationName string, withDocker bool) { + for _, key := range []string{ + "OTEL_EXPORTER_OTLP_ENDPOINT", + "OTEL_EXPORTER_OTLP_HEADERS", + "OTEL_EXPORTER_OTLP_METRICS_ENDPOINT", + "OTEL_EXPORTER_OTLP_METRICS_HEADERS", + "OTEL_EXPORTER_OTLP_METRICS_PROTOCOL", + "OTEL_EXPORTER_OTLP_PROTOCOL", + "OTEL_METRICS_EXPORTER", + } { + if value := os.Getenv(key); value != "" { + envOpts[key] = value + } + } + + if _, ok := envOpts["OTEL_METRICS_EXPORTER"]; !ok { + if envOpts["OTEL_EXPORTER_OTLP_ENDPOINT"] != "" || envOpts["OTEL_EXPORTER_OTLP_METRICS_ENDPOINT"] != "" { + envOpts["OTEL_METRICS_EXPORTER"] = "otlp" + } + } + + if envOpts["OTEL_METRICS_EXPORTER"] == "" { + return + } + + envOpts["OTEL_SERVICE_NAME"] = "EarthBuild-buildkitd" + + nesting := "outer" + if withDocker { + nesting = "inner" + } + + resourceAttrs := map[string]string{ + "earthbuild.process.role": "buildkitd", + "earthbuild.process.nesting": nesting, + "earthbuild.buildkit.container.name": containerName, + "earthbuild.installation.name": installationName, + } + envOpts["OTEL_RESOURCE_ATTRIBUTES"] = appendOTELResourceAttributes( + os.Getenv("OTEL_RESOURCE_ATTRIBUTES"), + resourceAttrs, + ) +} + +func appendOTELResourceAttributes(base string, attrs map[string]string) string { + parts := make([]string, 0, len(attrs)+1) + + for attr := range strings.SplitSeq(base, ",") { + attr = strings.TrimSpace(attr) + if attr == "" { + continue + } + + if _, value, ok := strings.Cut(attr, "="); !ok || strings.TrimSpace(value) == "" { + continue + } + + parts = append(parts, attr) + } + + for key, value := range attrs { + if value == "" { + continue + } + + parts = append(parts, key+"="+value) + } + + return strings.Join(parts, ",") +} + // Stop stops the buildkitd container. func Stop(ctx context.Context, containerName string, fe containerutil.ContainerFrontend) error { return fe.ContainerStop(ctx, 10, containerName) diff --git a/buildkitd/telemetry_test.go b/buildkitd/telemetry_test.go new file mode 100644 index 0000000000..940903adf2 --- /dev/null +++ b/buildkitd/telemetry_test.go @@ -0,0 +1,76 @@ +package buildkitd + +import ( + "strings" + "testing" +) + +func TestAddBuildkitTelemetryEnv(t *testing.T) { + t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "https://otel.example.test") + t.Setenv("OTEL_EXPORTER_OTLP_HEADERS", "authorization=Bearer token") + t.Setenv("OTEL_EXPORTER_OTLP_PROTOCOL", "http/protobuf") + t.Setenv("OTEL_RESOURCE_ATTRIBUTES", "cicd.pipeline.run.id=123,vcs.revision.id=abc") + + env := map[string]string{} + addBuildkitTelemetryEnv(env, "earthly-buildkitd", "earthly", true) + + if got := env["OTEL_SERVICE_NAME"]; got != "EarthBuild-buildkitd" { + t.Fatalf("OTEL_SERVICE_NAME = %q, want EarthBuild-buildkitd", got) + } + + if got := env["OTEL_METRICS_EXPORTER"]; got != "otlp" { + t.Fatalf("OTEL_METRICS_EXPORTER = %q, want otlp", got) + } + + if got := env["OTEL_EXPORTER_OTLP_ENDPOINT"]; got != "https://otel.example.test" { + t.Fatalf("OTEL_EXPORTER_OTLP_ENDPOINT = %q", got) + } + + if got := env["OTEL_EXPORTER_OTLP_HEADERS"]; got != "authorization=Bearer token" { + t.Fatalf("OTEL_EXPORTER_OTLP_HEADERS = %q", got) + } + + if got := env["OTEL_EXPORTER_OTLP_PROTOCOL"]; got != "http/protobuf" { + t.Fatalf("OTEL_EXPORTER_OTLP_PROTOCOL = %q", got) + } + + attrs := parseResourceAttrs(env["OTEL_RESOURCE_ATTRIBUTES"]) + wantAttrs := map[string]string{ + "cicd.pipeline.run.id": "123", + "vcs.revision.id": "abc", + "earthbuild.process.role": "buildkitd", + "earthbuild.process.nesting": "inner", + "earthbuild.buildkit.container.name": "earthly-buildkitd", + "earthbuild.installation.name": "earthly", + } + + for key, want := range wantAttrs { + if got := attrs[key]; got != want { + t.Fatalf("resource attr %s = %q, want %q", key, got, want) + } + } +} + +func TestAddBuildkitTelemetryEnvDoesNothingWithoutMetricsExporter(t *testing.T) { + t.Parallel() + + env := map[string]string{} + addBuildkitTelemetryEnv(env, "earthly-buildkitd", "earthly", false) + + if len(env) != 0 { + t.Fatalf("env = %#v, want empty", env) + } +} + +func parseResourceAttrs(value string) map[string]string { + attrs := map[string]string{} + + for part := range strings.SplitSeq(value, ",") { + key, value, ok := strings.Cut(part, "=") + if ok { + attrs[key] = value + } + } + + return attrs +} diff --git a/internal/telemetry/telemetry.go b/internal/telemetry/telemetry.go index 9e48b7ab8e..157e920102 100644 --- a/internal/telemetry/telemetry.go +++ b/internal/telemetry/telemetry.go @@ -9,14 +9,18 @@ import ( "net/http" "os" "path/filepath" + goruntime "runtime" + "strconv" "strings" "github.com/go-logr/stdr" "go.opentelemetry.io/contrib/exporters/autoexport" "go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp" - "go.opentelemetry.io/contrib/instrumentation/runtime" + otelruntime "go.opentelemetry.io/contrib/instrumentation/runtime" "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/log/global" + otelmetric "go.opentelemetry.io/otel/metric" "go.opentelemetry.io/otel/propagation" sdklog "go.opentelemetry.io/otel/sdk/log" "go.opentelemetry.io/otel/sdk/metric" @@ -52,6 +56,10 @@ func Setup(ctx context.Context) (ShutdownFunc, error) { shutdowns = nil + if shutdownErr != nil { + fmt.Fprintf(os.Stderr, "Warning: OpenTelemetry shutdown failed; continuing: %s\n", shutdownErr) + } + return shutdownErr } @@ -183,14 +191,170 @@ func setupMeterProvider(ctx context.Context, res *resource.Resource) (ShutdownFu ) otel.SetMeterProvider(mp) - err = runtime.Start() + err = otelruntime.Start() if err != nil { return errorf("initialize runtime metrics: %w", err) } + err = setupProcessMemoryMetrics() + if err != nil { + return errorf("initialize process memory metrics: %w", err) + } + return mp.Shutdown, nil } +func setupProcessMemoryMetrics() error { + meter := otel.Meter("go.earthbuild.dev/earthbuild/process") + attrs := processMemoryMetricAttributes() + + err := registerProcessMemoryGauge( + meter, + attrs, + "earthbuild_process_memory_alloc_bytes", + "Bytes allocated and still in use by this EarthBuild process.", + func(stats goruntime.MemStats) uint64 { return stats.Alloc }, + ) + if err != nil { + return err + } + + err = registerProcessMemoryGauge( + meter, + attrs, + "earthbuild_process_memory_heap_alloc_bytes", + "Heap bytes allocated and still in use by this EarthBuild process.", + func(stats goruntime.MemStats) uint64 { return stats.HeapAlloc }, + ) + if err != nil { + return err + } + + err = registerProcessMemoryGauge( + meter, + attrs, + "earthbuild_process_memory_heap_sys_bytes", + "Heap bytes obtained from the OS by this EarthBuild process.", + func(stats goruntime.MemStats) uint64 { return stats.HeapSys }, + ) + if err != nil { + return err + } + + return registerProcessMemoryGauge( + meter, + attrs, + "earthbuild_process_memory_sys_bytes", + "Total bytes obtained from the OS by this EarthBuild process.", + func(stats goruntime.MemStats) uint64 { return stats.Sys }, + ) +} + +func registerProcessMemoryGauge( + meter otelmetric.Meter, + attrs []attribute.KeyValue, + name string, + description string, + value func(goruntime.MemStats) uint64, +) error { + _, err := meter.Int64ObservableGauge( + name, + otelmetric.WithUnit("By"), + otelmetric.WithDescription(description), + otelmetric.WithInt64Callback(func(_ context.Context, observer otelmetric.Int64Observer) error { + var stats goruntime.MemStats + goruntime.ReadMemStats(&stats) + + observer.Observe(clampUint64ToInt64(value(stats)), otelmetric.WithAttributes(attrs...)) + + return nil + }), + ) + if err != nil { + return fmt.Errorf("create %s gauge: %w", name, err) + } + + return nil +} + +func clampUint64ToInt64(value uint64) int64 { + const maxInt64 = uint64(1<<63 - 1) + + if value > maxInt64 { + return int64(maxInt64) + } + + return int64(value) +} + +func processMemoryMetricAttributes() []attribute.KeyValue { + attrs := []attribute.KeyValue{ + attribute.Int("process.pid", os.Getpid()), + attribute.String("earthbuild.process.role", "earthbuild-cli"), + attribute.String("earthbuild.process.nesting", earthbuildProcessNesting()), + } + + for _, key := range []string{ + "cicd.pipeline.name", + "cicd.pipeline.run.id", + "cicd.pipeline.run.url.full", + "cicd.system.name", + "deployment.environment", + "user.id", + "vcs.ref.name", + "vcs.repository.change.id", + "vcs.repository.name", + "vcs.revision.id", + } { + if value, ok := otelResourceAttributeFromEnv(key); ok { + attrs = append(attrs, attribute.String(key, value)) + } + } + + if target := earthbuildTargetFromArgs(os.Args); target != "" { + attrs = append(attrs, attribute.String("earthbuild.target", target)) + } + + return attrs +} + +func earthbuildProcessNesting() string { + if value, _ := strconv.ParseBool(os.Getenv("EARTHLY_WITH_DOCKER")); value { + return "inner" + } + + return "outer" +} + +func otelResourceAttributeFromEnv(key string) (string, bool) { + for attr := range strings.SplitSeq(os.Getenv("OTEL_RESOURCE_ATTRIBUTES"), ",") { + attrKey, value, ok := strings.Cut(attr, "=") + if !ok || strings.TrimSpace(attrKey) != key { + continue + } + + value = strings.TrimSpace(value) + + return value, value != "" + } + + return "", false +} + +func earthbuildTargetFromArgs(args []string) string { + for _, arg := range args[1:] { + if strings.HasPrefix(arg, "-") { + continue + } + + if strings.Contains(arg, "+") { + return arg + } + } + + return "" +} + func setupLoggerProvider(ctx context.Context, res *resource.Resource) (ShutdownFunc, error) { errorf := func(format string, args ...any) (ShutdownFunc, error) { return nil, fmt.Errorf("create logger provider: "+format, args...) diff --git a/util/statsstreamparser/parser.go b/util/statsstreamparser/parser.go index 1746671810..66fdd9dd98 100644 --- a/util/statsstreamparser/parser.go +++ b/util/statsstreamparser/parser.go @@ -28,6 +28,16 @@ func New() *Parser { } } +// Reset discards any buffered partial frame and returns the parser to its +// initial state. Used to recover from a desynced or malformed stats stream +// (e.g. when the daemon's runc stats collector hits EOF and emits a partial +// or raw frame) without treating the decode failure as fatal. +func (ssp *Parser) Reset() { + ssp.buf.Reset() + ssp.bsr = binarystream.NewReader(ssp.buf, binary.LittleEndian) + ssp.readProtocolVersion = false +} + // Parse parses stream data containing execution statistics. func (ssp *Parser) Parse(b []byte) ([]*runc.Stats, error) { _, err := ssp.buf.Write(b) diff --git a/util/statsstreamparser/parser_test.go b/util/statsstreamparser/parser_test.go new file mode 100644 index 0000000000..cd41ba68e3 --- /dev/null +++ b/util/statsstreamparser/parser_test.go @@ -0,0 +1,58 @@ +package statsstreamparser + +import ( + "bytes" + "encoding/binary" + "encoding/json" + "testing" + + "github.com/containerd/go-runc" + "github.com/stretchr/testify/require" +) + +// frame builds a valid stats stream frame: [version=1][uint32 LE len][JSON]. +func frame(t *testing.T, s runc.Stats) []byte { + t.Helper() + + j, err := json.Marshal(s) + require.NoError(t, err) + + var b bytes.Buffer + b.WriteByte(0x01) + require.NoError(t, binary.Write(&b, binary.LittleEndian, uint32(len(j)))) //nolint:gosec // test frame length is tiny + b.Write(j) + + return b.Bytes() +} + +func TestParseValidFrame(t *testing.T) { + t.Parallel() + + p := New() + stats, err := p.Parse(frame(t, runc.Stats{})) + require.NoError(t, err) + require.Len(t, stats, 1) +} + +// A malformed frame (raw JSON, first byte '{' = 0x7B = 123) is the exact CI +// failure: the daemon's runc stats collector hit EOF and emitted raw bytes. +// It must be reported as an error, and Reset must let the parser recover so a +// transient desync does not permanently wedge stats collection. +func TestParserRecoversAfterGarbageFrame(t *testing.T) { + t.Parallel() + + p := New() + + _, err := p.Parse(frame(t, runc.Stats{})) + require.NoError(t, err) + + _, err = p.Parse([]byte(`{"malformed":true}`)) + require.Error(t, err) + require.Contains(t, err.Error(), "protocol version 123") + + p.Reset() + + stats, err := p.Parse(frame(t, runc.Stats{})) + require.NoError(t, err, "parser must recover after Reset") + require.Len(t, stats, 1) +}