feat: process-memory + buildkitd OTEL telemetry for CI memory diagnos…

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)

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
+}

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