diff --git a/agent/agent_configuration.go b/agent/agent_configuration.go index 64a87e8d86..96395aff59 100644 --- a/agent/agent_configuration.go +++ b/agent/agent_configuration.go @@ -66,6 +66,7 @@ type AgentConfiguration struct { EnableJobLogTmpfile bool JobLogPath string WriteJobLogsToStdout bool + JobLogsOTLP bool LogFormat string Shell string HooksShell string diff --git a/agent/job_runner.go b/agent/job_runner.go index c95d55c8c8..e893548840 100644 --- a/agent/job_runner.go +++ b/agent/job_runner.go @@ -709,25 +709,30 @@ BUILDKITE_AGENT_JWKS_KEY_ID` setEnv("BUILDKITE_TRACING_BACKEND", r.conf.AgentConfiguration.TracingBackend) setEnv("BUILDKITE_TRACING_SERVICE_NAME", r.conf.AgentConfiguration.TracingServiceName) - // Buildkite backend can provide a traceparent property on the job - // which can be propagated to the job tracing if OpenTelemetry is used - // - // https://www.w3.org/TR/trace-context/#traceparent-header - if r.conf.Job.TraceParent != "" { - setEnv("BUILDKITE_TRACING_TRACEPARENT", r.conf.Job.TraceParent) - } - // Buildkite backend may also provide a tracestate property on the job, - // which carries vendor-specific trace context alongside the traceparent. - // - // https://www.w3.org/TR/trace-context/#tracestate-header - if r.conf.Job.TraceState != "" { - setEnv("BUILDKITE_TRACING_TRACESTATE", r.conf.Job.TraceState) - } if r.conf.AgentConfiguration.TracingPropagateTraceparent { + // Buildkite backend can provide a traceparent property on the job + // which can be propagated to the job tracing if OpenTelemetry is used. + // + // https://www.w3.org/TR/trace-context/#traceparent-header + if r.conf.Job.TraceParent != "" { + setEnv("BUILDKITE_TRACING_TRACEPARENT", r.conf.Job.TraceParent) + } + // Buildkite backend may also provide a tracestate property on the job, + // which carries vendor-specific trace context alongside traceparent. + // + // https://www.w3.org/TR/trace-context/#tracestate-header + if r.conf.Job.TraceState != "" { + setEnv("BUILDKITE_TRACING_TRACESTATE", r.conf.Job.TraceState) + } setEnv("BUILDKITE_TRACING_PROPAGATE_TRACEPARENT", "true") } } + // This is an agent-operator setting, not a pipeline setting. Always + // overwrite the job-provided value so a pipeline cannot enable OTLP export + // or choose its destination when the agent operator has not opted in. + setEnv("BUILDKITE_JOB_LOGS_OTLP", fmt.Sprint(r.conf.AgentConfiguration.JobLogsOTLP)) + setEnv("BUILDKITE_AGENT_DISABLE_WARNINGS_FOR", strings.Join(r.conf.AgentConfiguration.DisableWarningsFor, ",")) // see documentation for BuildkiteMessageMax diff --git a/clicommand/agent_start.go b/clicommand/agent_start.go index a040434911..3198ec9824 100644 --- a/clicommand/agent_start.go +++ b/clicommand/agent_start.go @@ -127,6 +127,7 @@ type AgentStartConfig struct { LogFormat string `cli:"log-format"` WriteJobLogsToStdout bool `cli:"write-job-logs-to-stdout"` + JobLogsOTLP bool `cli:"job-logs-otlp"` DisableWarningsFor []string `cli:"disable-warnings-for" normalize:"list"` BuildPath string `cli:"build-path" normalize:"filepath" validate:"required"` @@ -439,6 +440,11 @@ var AgentStartCommand = cli.Command{ Usage: "Writes job logs to the agent process' stdout. This simplifies log collection if running agents in Docker (default: false)", EnvVar: "BUILDKITE_WRITE_JOB_LOGS_TO_STDOUT", }, + cli.BoolFlag{ + Name: "job-logs-otlp", + Usage: "Export job logs directly as OpenTelemetry log records using the OTEL_EXPORTER_OTLP_LOGS_* / OTEL_EXPORTER_OTLP_* environment configuration (default: false)", + EnvVar: "BUILDKITE_JOB_LOGS_OTLP", + }, cli.StringFlag{ Name: "shell", Value: DefaultShell(), @@ -1109,6 +1115,7 @@ var AgentStartCommand = cli.Command{ EnableJobLogTmpfile: cfg.EnableJobLogTmpfile, JobLogPath: cfg.JobLogPath, WriteJobLogsToStdout: cfg.WriteJobLogsToStdout, + JobLogsOTLP: cfg.JobLogsOTLP, LogFormat: cfg.LogFormat, Shell: cfg.Shell, HooksShell: cfg.HooksShell, diff --git a/clicommand/bootstrap.go b/clicommand/bootstrap.go index e1c32ff30a..2bfdcdb7a7 100644 --- a/clicommand/bootstrap.go +++ b/clicommand/bootstrap.go @@ -117,6 +117,7 @@ type BootstrapConfig struct { TracingPropagateTraceparent bool `cli:"tracing-propagate-traceparent"` TraceContextEncoding string `cli:"trace-context-encoding"` NoJobAPI bool `cli:"no-job-api"` + JobLogsOTLP bool `cli:"job-logs-otlp"` DisableWarningsFor []string `cli:"disable-warnings-for" normalize:"list"` CheckoutAttempts int `cli:"checkout-attempts"` } @@ -367,12 +368,16 @@ var BootstrapCommand = cli.Command{ Usage: "Accept traceparent from Buildkite control plane (default: false)", EnvVar: "BUILDKITE_TRACING_PROPAGATE_TRACEPARENT", }, - cli.BoolFlag{ Name: "no-job-api", Usage: "Disables the Job API, which gives commands in jobs some abilities to introspect and mutate the state of the job (default: false)", EnvVar: "BUILDKITE_AGENT_NO_JOB_API", }, + cli.BoolFlag{ + Name: "job-logs-otlp", + EnvVar: "BUILDKITE_JOB_LOGS_OTLP", + Hidden: true, + }, cli.StringSliceFlag{ Name: "disable-warnings-for", Usage: "A list of warning IDs to disable", @@ -508,6 +513,7 @@ var BootstrapCommand = cli.Command{ TracingTraceState: cfg.TracingTraceState, TracingPropagateTraceparent: cfg.TracingPropagateTraceparent, JobAPI: !cfg.NoJobAPI, + JobLogsOTLP: cfg.JobLogsOTLP, DisabledWarnings: cfg.DisableWarningsFor, Secrets: cfg.Secrets, CheckoutAttempts: cfg.CheckoutAttempts, diff --git a/go.mod b/go.mod index ead6ec30f6..ab746a57c0 100644 --- a/go.mod +++ b/go.mod @@ -57,9 +57,13 @@ require ( go.opentelemetry.io/contrib/propagators/jaeger v1.44.0 go.opentelemetry.io/contrib/propagators/ot v1.44.0 go.opentelemetry.io/otel v1.44.0 + go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploggrpc v0.20.0 + go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploghttp v0.20.0 go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.44.0 go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.44.0 + go.opentelemetry.io/otel/log v0.20.0 go.opentelemetry.io/otel/sdk v1.44.0 + go.opentelemetry.io/otel/sdk/log v0.20.0 go.opentelemetry.io/otel/trace v1.44.0 golang.org/x/crypto v0.53.0 golang.org/x/net v0.56.0 diff --git a/go.sum b/go.sum index 6a6d90d4fc..d4d1b39171 100644 --- a/go.sum +++ b/go.sum @@ -477,16 +477,26 @@ go.opentelemetry.io/contrib/propagators/ot v1.44.0 h1:JLTPenzmPtLp5ODPntAA5JhxVu go.opentelemetry.io/contrib/propagators/ot v1.44.0/go.mod h1:8zr0bHgwkoQXucBK39/H4QphmLf1lSen1Z7FPDZD5Uc= go.opentelemetry.io/otel v1.44.0 h1:JjwHmHpA4iZ3wBxluu2fbbE7j4kqlE8jXyAyPXH7HqU= go.opentelemetry.io/otel v1.44.0/go.mod h1:BMgjTHL9WPRlRjL2oZCBTL4whCGtXch2H4BhOPIAyYc= +go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploggrpc v0.20.0 h1:rydZ9sxbcFdm/oWrVyfLTjHIygMgv0bEeMd+3B/BvoM= +go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploggrpc v0.20.0/go.mod h1:earQ25dooT0Hhspq59DZ8YCC50jWfOlFEeWoxy/P444= +go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploghttp v0.20.0 h1:owlhcJ3QO3X0YTDTCcDZ4V+6aVDkWbNmBoQ5NUp7Oww= +go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploghttp v0.20.0/go.mod h1:MP4eemTiI9zC8fgg+DYynhYDYf3ba72S376TvP+Ye0Q= go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.44.0 h1:4YsVu3B8+3qtWYYrsUYgn0OG78pN0rnNPRGX4SbokQI= go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.44.0/go.mod h1:+wnlSn0mD1ADVMe3v9Z/WIaiz6q6gL2J/ejaAmdmv80= go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.44.0 h1:qazEJlUOQzhCpzQpFETGby7EdqjI1wsd0W+6Gg1SCTU= go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.44.0/go.mod h1:fOD2Yefuxixkx3ahVNf0O/PERb6r4OlbxfATVnYvzCo= go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.44.0 h1:lgh3PiVrRUWMLOVSkQicxzZll5NjF1r+AtsX1XRIHw0= go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.44.0/go.mod h1:5Cnhth3m/AgOeTgE3ex12pPmiu/gGtZit03kSzx9X7s= +go.opentelemetry.io/otel/log v0.20.0 h1:/5i0vuHxCLWUfChWG41K9wkM0jafruPw9NU1/RCJirs= +go.opentelemetry.io/otel/log v0.20.0/go.mod h1:wOcMcjsZpG8x7Bak7IhSi/lg8wscV2C1VdrKCLPlt0E= go.opentelemetry.io/otel/metric v1.44.0 h1:1w0gILTcHdr3YI+ixLyjemwrVnsMURbTZFrSYCdDdmc= go.opentelemetry.io/otel/metric v1.44.0/go.mod h1:8O7hanEPBNgEMmybD3s2VBKcgWOCsA6tzHBPODAiquo= go.opentelemetry.io/otel/sdk v1.44.0 h1:nHYwb9lK+fJPU/dnT6s7W7Z8itMWyqrnVfbheVYrZ58= go.opentelemetry.io/otel/sdk v1.44.0/go.mod h1:Osuydd3Se74nqjAKxid74N5eC+jfEqfTegHRnq58oK0= +go.opentelemetry.io/otel/sdk/log v0.20.0 h1:vM3xI7TQgKPiSghe6urZtAkyFY7SodrSpC83CffDFuY= +go.opentelemetry.io/otel/sdk/log v0.20.0/go.mod h1:Knej2nmsTUzN79T2eeXdRsjjPcoxoq2pUyUHz9TFyyU= +go.opentelemetry.io/otel/sdk/log/logtest v0.20.0 h1:OqdRZ1guyzamK3M6LlRsmGqRrjkHWw6WZOKKli5ELpg= +go.opentelemetry.io/otel/sdk/log/logtest v0.20.0/go.mod h1:PuMIlm7zAt7c3z8zfOI5ox4iT1Z87We+PF6YoINux/M= go.opentelemetry.io/otel/sdk/metric v1.44.0 h1:3LlKgI+VjbVsjNRFZJZAJ30WjXC5VkNRks6si09iEfI= go.opentelemetry.io/otel/sdk/metric v1.44.0/go.mod h1:5B5pMARnXxKhltooO4xUuCBorl65a4EpnTalObqOigA= go.opentelemetry.io/otel/trace v1.44.0 h1:jxF5CsGYCe74MCRx2X4g7WsY/VBKRqqpNvXlX/6gtIk= diff --git a/internal/job/config.go b/internal/job/config.go index 3e88c572d0..68b3ee1dda 100644 --- a/internal/job/config.go +++ b/internal/job/config.go @@ -206,9 +206,9 @@ type ExecutorConfig struct { TracingTraceParent string // W3C tracestate accompanying TracingTraceParent. Plumbed through to the - // bootstrap environment whenever the server provides a value, but only - // attached to the OTel span context when TracingPropagateTraceparent is - // enabled (same opt-in gate as TracingTraceParent). + // bootstrap environment and attached to the OTel span context only when + // TracingPropagateTraceparent is enabled (same opt-in gate as + // TracingTraceParent). TracingTraceState string // Accept traceparent context from Buildkite control plane @@ -220,6 +220,9 @@ type ExecutorConfig struct { // Whether to start the JobAPI JobAPI bool + // Whether to emit visible job process output as OTLP log records. + JobLogsOTLP bool + // The warnings that have been disabled by the user DisabledWarnings []string diff --git a/internal/job/executor.go b/internal/job/executor.go index 71613e885d..da10d945e6 100644 --- a/internal/job/executor.go +++ b/internal/job/executor.go @@ -74,10 +74,54 @@ type Executor struct { // In order for the latter to happen, a reference is passed into the the Job API server as well redactors *replacer.Mux + // otlpJobLogger retains the OTLP redaction streams while job log export is + // enabled. Executor redactor flush boundaries must flush these streams too + // so OTLP and the customer-facing job log have identical stream semantics. + otlpJobLogger *otlpJobLogger + // jobAPI is the Job API server for this job. It's retained so the executor // can consume out-of-band requests made by hooks, such as a working // directory set by an unwrapped hook via the /workdir endpoint. jobAPI *jobapi.Server + + // loggerTee carries the redacted shell logger output (section headers, + // prompts, comments, warnings) to stderr and, when OTLP job logging is + // enabled, mirrors it into the OTLP exporter so the exported records match + // the customer-facing Buildkite job log. + loggerTee *teeWriter +} + +// teeWriter writes to a primary writer and, when configured, also to a +// secondary writer. The secondary sink can be attached after construction, +// which lets the executor mirror already-redacted shell logger output into the +// OTLP exporter once it has been initialised. +type teeWriter struct { + mu sync.Mutex + primary io.Writer + secondary io.Writer +} + +func (w *teeWriter) Write(p []byte) (int, error) { + // Hold the lock for the whole write so that detaching the secondary sink + // (setSecondary(nil), called before the OTLP exporter is flushed and shut + // down) cannot complete while a write is still in flight. This guarantees + // no control output lands on the OTLP emitter after it has been flushed. + w.mu.Lock() + defer w.mu.Unlock() + + n, err := w.primary.Write(p) + if w.secondary != nil { + // The secondary sink is best-effort: failures must not affect the + // primary (customer-facing) job log output. + _, _ = w.secondary.Write(p) + } + return n, err +} + +func (w *teeWriter) setSecondary(secondary io.Writer) { + w.mu.Lock() + w.secondary = secondary + w.mu.Unlock() } // New returns a new executor instance @@ -156,6 +200,36 @@ func (e *Executor) Run(ctx context.Context) (exitCode int) { // Create an empty env for us to keep track of our env changes in e.shell.Env = env.FromSlice(os.Environ()) + // OTLP job log export lives entirely in the bootstrap process, which is the + // single home for this feature. When OpenTelemetry tracing is enabled, the + // per-command emit context carries the active hook/command span, so log + // records are trace-correlated. With tracing disabled, records are still + // emitted but remain uncorrelated. + if e.JobLogsOTLP { + jobLogger, err := newOTLPJobLogger(ctx, e) + if err != nil { + e.shell.Warningf("Failed to initialize OTLP job log exporter: %v", err) + } else { + e.otlpJobLogger = jobLogger + e.shell.SetOutputInterceptor(jobLogger.Wrap) + // Mirror the redacted shell logger control output (section headers, + // prompts, comments, warnings) into OTLP so the exported records + // match the downloadable Buildkite job log and the UI stream. + if e.loggerTee != nil { + e.loggerTee.setSecondary(jobLogger.controlWriter()) + } + defer func() { + if e.loggerTee != nil { + e.loggerTee.setSecondary(nil) + } + if err := jobLogger.Close(); err != nil { + e.shell.Warningf("Failed to close OTLP job log exporter: %v", err) + } + e.otlpJobLogger = nil + }() + } + } + // Initialize the job API, iff the experiment is enabled. Noop otherwise if e.JobAPI { cleanup, err := e.startJobAPI() @@ -361,6 +435,18 @@ type HookConfig struct { PluginName string } +func hookLogAttributes(hookCfg HookConfig) map[string]string { + attrs := map[string]string{ + "buildkite.phase": "hook", + "buildkite.hook.name": hookCfg.Name, + "buildkite.hook.scope": hookCfg.Scope, + } + if hookCfg.PluginName != "" { + attrs["buildkite.hook.plugin"] = hookCfg.PluginName + } + return attrs +} + func (e *Executor) tracingImplementationSpecificHookScope(scope string) string { if e.TracingBackend != tracetools.BackendDatadog { return scope @@ -463,7 +549,7 @@ func (e *Executor) runUnwrappedHook(ctx context.Context, _ string, hookCfg HookC environ.Set("BUILDKITE_HOOK_PATH", hookCfg.Path) environ.Set("BUILDKITE_HOOK_SCOPE", hookCfg.Scope) - if err := e.shell.Command(hookCfg.Path).Run(ctx, shell.WithExtraEnv(environ)); err != nil { + if err := e.shell.Command(hookCfg.Path).Run(ctx, shell.WithExtraEnv(environ), shell.WithOutputAttributes(hookLogAttributes(hookCfg))); err != nil { return err } // Passing an empty env changes through because in polyglot hook we can't detect @@ -537,7 +623,7 @@ func logMissingHookInfo(l shell.Logger, hookName, wrapperPath string) { func (e *Executor) runWrappedShellScriptHook(ctx context.Context, hookName string, hookCfg HookConfig) error { defer func() { - if err := e.redactors.Flush(); err != nil { + if err := e.flushOutputRedactors(); err != nil { e.shell.Errorf("Error flushing redactors: %v", err) } }() @@ -572,7 +658,7 @@ func (e *Executor) runWrappedShellScriptHook(ctx context.Context, hookName strin if err != nil { return err } - return script.Run(ctx, shell.ShowPrompt(false), shell.WithExtraEnv(hookCfg.Env)) + return script.Run(ctx, shell.ShowPrompt(false), shell.WithExtraEnv(hookCfg.Env), shell.WithOutputAttributes(hookLogAttributes(hookCfg))) }() if err != nil { exitCode := shell.ExitCode(err) @@ -730,6 +816,14 @@ func (e *Executor) addOutputRedactors() { } } +func (e *Executor) flushOutputRedactors() error { + err := e.redactors.Flush() + if e.otlpJobLogger != nil { + e.otlpJobLogger.FlushRedactors() + } + return err +} + func (e *Executor) hasGlobalHook(name string) bool { _, err := hook.Find(nil, e.HooksPath, name) if err == nil { @@ -1167,7 +1261,7 @@ func (e *Executor) CommandPhase(ctx context.Context) (hookErr, commandErr error) // defaultCommandPhase is executed if there is no global or plugin command hook func (e *Executor) defaultCommandPhase(ctx context.Context) (retErr error) { defer func() { - if err := e.redactors.Flush(); err != nil { + if err := e.flushOutputRedactors(); err != nil { e.shell.Errorf("Error flushing redactors: %v", err) } }() @@ -1299,7 +1393,11 @@ func (e *Executor) defaultCommandPhase(ctx context.Context) (retErr error) { e.shell.Promptf("%s", cmdToExec) } - err = e.shell.Command(cmd[0], cmd[1:]...).Run(ctx, shell.ShowPrompt(false)) + err = e.shell.Command(cmd[0], cmd[1:]...).Run(ctx, shell.ShowPrompt(false), shell.WithOutputAttributes(map[string]string{ + "buildkite.phase": "command", + "buildkite.hook.name": "command", + "buildkite.hook.scope": "default", + })) return err } @@ -1398,7 +1496,12 @@ func (e *Executor) setupRedactors(log shell.Logger, environ *env.Environment, st stdoutRedactor := replacer.New(stdout, needles, redact.Redacted) e.redactors.Append(stdoutRedactor) - loggerRedactor := replacer.New(stderr, needles, redact.Redacted) + // The shell logger writes through this redactor into loggerTee, whose + // primary sink is stderr. When OTLP job logging is enabled, a secondary + // sink is attached so the same already-redacted control output is mirrored + // into the OTLP exporter. + e.loggerTee = &teeWriter{primary: stderr} + loggerRedactor := replacer.New(e.loggerTee, needles, redact.Redacted) e.redactors.Append(loggerRedactor) logger := shell.NewWriterLogger(loggerRedactor, true, e.DisabledWarnings) diff --git a/internal/job/executor_test.go b/internal/job/executor_test.go index d319ebfee8..6dd245db17 100644 --- a/internal/job/executor_test.go +++ b/internal/job/executor_test.go @@ -11,6 +11,7 @@ import ( "github.com/buildkite/agent/v3/tracetools" "github.com/google/go-cmp/cmp" "github.com/opentracing/opentracing-go" + "go.opentelemetry.io/otel/trace" "gopkg.in/DataDog/dd-trace-go.v1/ddtrace/opentracer" ) @@ -107,6 +108,47 @@ func TestStartTracing_Datadog(t *testing.T) { stopper() } +func TestContextWithTraceparentIfEnabledDoesNotAcceptServerTraceparentWithoutPropagation(t *testing.T) { + t.Parallel() + + sh, err := shell.New(shell.WithLogger(shell.DiscardLogger)) + if err != nil { + t.Fatalf("shell.New() error = %v", err) + } + e := New(ExecutorConfig{ + TracingTraceParent: "00-aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa-bbbbbbbbbbbbbbbb-01", + }) + e.shell = sh + + ctx := e.contextWithTraceparentIfEnabled(t.Context()) + if sc := trace.SpanContextFromContext(ctx); sc.IsValid() { + t.Fatalf("SpanContextFromContext(ctx).IsValid() = true, want false") + } +} + +func TestContextWithTraceparentIfEnabledAcceptsServerTraceparentWithPropagation(t *testing.T) { + t.Parallel() + + sh, err := shell.New(shell.WithLogger(shell.DiscardLogger)) + if err != nil { + t.Fatalf("shell.New() error = %v", err) + } + e := New(ExecutorConfig{ + TracingPropagateTraceparent: true, + TracingTraceParent: "00-aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa-bbbbbbbbbbbbbbbb-01", + }) + e.shell = sh + + ctx := e.contextWithTraceparentIfEnabled(t.Context()) + sc := trace.SpanContextFromContext(ctx) + if !sc.IsValid() { + t.Fatalf("SpanContextFromContext(ctx).IsValid() = false, want true") + } + if got, want := sc.TraceID().String(), "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"; got != want { + t.Fatalf("SpanContextFromContext(ctx).TraceID() = %q, want %q", got, want) + } +} + // newCancelTestExecutor returns an Executor whose shell.Env starts empty, // suitable for exercising Cancel without depending on the host environment. func newCancelTestExecutor(t *testing.T) *Executor { diff --git a/internal/job/otlp_job_logger.go b/internal/job/otlp_job_logger.go new file mode 100644 index 0000000000..8cec1f52d7 --- /dev/null +++ b/internal/job/otlp_job_logger.go @@ -0,0 +1,375 @@ +package job + +import ( + "bytes" + "context" + "fmt" + "io" + "os" + "reflect" + "sync" + "time" + + "github.com/buildkite/agent/v3/internal/redact" + "github.com/buildkite/agent/v3/internal/replacer" + "github.com/buildkite/agent/v3/version" + "go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploggrpc" + "go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploghttp" + otellog "go.opentelemetry.io/otel/log" + sdklog "go.opentelemetry.io/otel/sdk/log" + "go.opentelemetry.io/otel/sdk/resource" + semconv "go.opentelemetry.io/otel/semconv/v1.4.0" +) + +// Keep newline-free output bounded while retaining normal line-oriented +// records. This matches the maximum chunk size used by common log shippers and +// avoids buffering an arbitrarily large process write until command exit. +const otlpLogRecordMaxBytes = 64 * 1024 + +type otlpJobLogger struct { + log otellog.Logger + provider *sdklog.LoggerProvider + attrs []otellog.KeyValue + redactors *replacer.Mux + streamsMu sync.Mutex + streams []*otlpJobLogStream + + // control mirrors bootstrap-generated control output (section headers, + // prompts, comments, warnings) into OTLP so the exported records match the + // Buildkite job log stream. It is created lazily by controlWriter. + control *otlpLineEmitter +} + +func newOTLPJobLogger(ctx context.Context, e *Executor) (*otlpJobLogger, error) { + protocol := os.Getenv("OTEL_EXPORTER_OTLP_LOGS_PROTOCOL") + if protocol == "" { + protocol = os.Getenv("OTEL_EXPORTER_OTLP_PROTOCOL") + } + if protocol == "" { + protocol = "grpc" + } + + var exporter sdklog.Exporter + var err error + switch protocol { + case "grpc": + exporter, err = otlploggrpc.New(ctx) + case "http/protobuf", "http": + exporter, err = otlploghttp.New(ctx) + default: + return nil, fmt.Errorf("unsupported OTLP logs protocol: %s", protocol) + } + if err != nil { + return nil, fmt.Errorf("creating OTLP log exporter: %w", err) + } + + serviceName := e.TracingServiceName + if serviceName == "" { + serviceName = "buildkite-agent" + } + resources := resource.NewWithAttributes(semconv.SchemaURL, + semconv.ServiceNameKey.String(serviceName), + semconv.ServiceVersionKey.String(version.Version()), + semconv.DeploymentEnvironmentKey.String("ci"), + ) + provider := sdklog.NewLoggerProvider( + // Job logs must not be silently dropped when the SDK batch queue fills. + // Back-pressure the command output on the exporter instead. + sdklog.WithProcessor(sdklog.NewSimpleProcessor(exporter)), + sdklog.WithResource(resources), + ) + + return &otlpJobLogger{ + log: provider.Logger( + "buildkite-agent", + otellog.WithInstrumentationVersion(version.Version()), + otellog.WithSchemaURL(semconv.SchemaURL), + ), + provider: provider, + attrs: otlpJobAttributes(e), + redactors: e.redactors, + }, nil +} + +func (l *otlpJobLogger) Wrap(ctx context.Context, out io.Writer, attrs map[string]string) io.Writer { + emitter := &otlpLineEmitter{ + ctx: context.WithoutCancel(ctx), + log: l.log, + attrs: appendLogAttrs(l.attrs, attrs), + } + + // Keep one redaction stream for each shared downstream job-log writer. The + // visible job-log redactor also persists across command boundaries, so OTLP + // must retain partial matches across Wrap calls to avoid leaking fragments + // of a secret split between two commands. + l.streamsMu.Lock() + var stream *otlpJobLogStream + for _, candidate := range l.streams { + if sameOTLPWriter(candidate.out, out) { + stream = candidate + break + } + } + if stream == nil { + stream = &otlpJobLogStream{out: out, live: l.redactors} + stream.redactor = replacer.New( + otlpRedactedWriter{stream: stream}, + l.redactors.Needles(), + redact.Redacted, + ) + l.streams = append(l.streams, stream) + } + l.streamsMu.Unlock() + + return &otlpJobLogWriter{ + stream: stream, + emitter: emitter, + } +} + +// controlWriter returns an io.Writer that mirrors bootstrap control output +// (section headers, prompts, comments, warnings) into OTLP as log records, so +// the exported records contain the same lines a customer sees in the Buildkite +// UI. It is fed post-redaction bytes from the shell logger's redactor, so it +// does not redact again. Lines carry the base job attributes but no per-hook +// phase or span context (control output is bootstrap narration, not the output +// of a specific traced hook/command). +func (l *otlpJobLogger) controlWriter() io.Writer { + if l.control == nil { + l.control = &otlpLineEmitter{ + ctx: context.Background(), + log: l.log, + attrs: l.attrs, + } + } + return l.control +} + +func (l *otlpJobLogger) Close() error { + if l.control != nil { + l.control.flush() + } + l.FlushRedactors() + if l.provider == nil { + return nil + } + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + flushErr := l.provider.ForceFlush(ctx) + shutdownErr := l.provider.Shutdown(ctx) + if flushErr != nil { + return flushErr + } + return shutdownErr +} + +// FlushRedactors ends the current shared OTLP redaction streams. The executor +// calls this at the same explicit hook/default-command boundaries where it +// flushes the customer-facing redactors. Individual subprocesses within one of +// those phases deliberately do not flush this state. +func (l *otlpJobLogger) FlushRedactors() { + l.streamsMu.Lock() + streams := append([]*otlpJobLogStream(nil), l.streams...) + l.streamsMu.Unlock() + for _, stream := range streams { + stream.flush() + } +} + +// otlpJobLogWriter tees process output to the normal (already redacting) job-log +// writer and to a redacting OTLP line emitter, so OTLP records carry the same +// redacted content as the customer-facing job log. +type otlpJobLogWriter struct { + stream *otlpJobLogStream + emitter *otlpLineEmitter +} + +func (w *otlpJobLogWriter) Write(data []byte) (int, error) { + w.stream.mu.Lock() + defer w.stream.mu.Unlock() + + n, err := w.stream.out.Write(data) + // Keep the OTLP redactor's needles in sync with the live job redactor + // before redacting, so secrets added mid-command (e.g. via the Job API) + // are redacted in OTLP output too. Replacer.Add deduplicates, so re-adding + // the full needle set each write is safe. + if w.stream.live != nil { + w.stream.redactor.Add(w.stream.live.Needles()...) + } + // Feed the OTLP copy through the redactor, which streams redacted bytes to + // this command's line emitter. Errors here must not affect the primary job + // log. The stream retains partial redaction matches for the next command. + w.stream.emitter = w.emitter + _, _ = w.stream.redactor.Write(data) + return n, err +} + +func (w *otlpJobLogWriter) Flush() { + w.stream.mu.Lock() + defer w.stream.mu.Unlock() + // Flush this command's complete line data, but deliberately do not flush + // the shared redactor: it may be withholding the prefix of a secret that + // continues in the next command. + w.emitter.flush() +} + +// otlpJobLogStream owns redaction state shared by command writers that feed the +// same downstream job-log stream. +type otlpJobLogStream struct { + mu sync.Mutex + out io.Writer + live *replacer.Mux + redactor *replacer.Replacer + emitter *otlpLineEmitter +} + +func (s *otlpJobLogStream) flush() { + s.mu.Lock() + defer s.mu.Unlock() + _ = s.redactor.Flush() + if s.emitter != nil { + s.emitter.flush() + } +} + +// otlpRedactedWriter routes bytes released by the persistent redactor to the +// emitter for the command whose write released them. Its stream mutex is held +// by all callers. +type otlpRedactedWriter struct { + stream *otlpJobLogStream +} + +func (w otlpRedactedWriter) Write(data []byte) (int, error) { + if w.stream.emitter == nil { + return len(data), nil + } + return w.stream.emitter.Write(data) +} + +func sameOTLPWriter(a, b io.Writer) bool { + if a == nil || b == nil { + return a == nil && b == nil + } + av, bv := reflect.ValueOf(a), reflect.ValueOf(b) + return av.Type() == bv.Type() && av.Comparable() && av.Interface() == bv.Interface() +} + +// otlpLineEmitter line-buffers (already redacted) output and emits each line as +// an OpenTelemetry log record with a native timestamp. Write and flush are +// guarded by mu because control output may be written from multiple goroutines +// (e.g. the cancellation handler emitting a comment). +type otlpLineEmitter struct { + ctx context.Context + log otellog.Logger + attrs []otellog.KeyValue + mu sync.Mutex + buf []byte +} + +func (e *otlpLineEmitter) Write(data []byte) (int, error) { + e.mu.Lock() + defer e.mu.Unlock() + + origLen := len(data) + for len(data) > 0 { + i := bytes.IndexByte(data, '\n') + if i < 0 { + e.appendChunks(data) + return origLen, nil + } + + emittedChunk := e.appendChunks(data[:i]) + e.buf = bytes.TrimSuffix(e.buf, []byte{'\r'}) + if len(e.buf) > 0 || !emittedChunk { + e.emit(string(e.buf)) + } + e.buf = e.buf[:0] + data = data[i+1:] + } + + return origLen, nil +} + +func (e *otlpLineEmitter) appendChunks(data []byte) bool { + emitted := false + for len(data) > 0 { + // Keep one full chunk buffered until we see either more data or a line + // terminator. This lets Write strip a trailing carriage return when a + // CRLF sequence arrives across separate writes without exceeding the + // memory bound. + if len(e.buf) == otlpLogRecordMaxBytes { + e.emit(string(e.buf)) + e.buf = e.buf[:0] + emitted = true + } + remaining := otlpLogRecordMaxBytes - len(e.buf) + if remaining > len(data) { + remaining = len(data) + } + e.buf = append(e.buf, data[:remaining]...) + data = data[remaining:] + } + return emitted +} + +func (e *otlpLineEmitter) flush() { + e.mu.Lock() + defer e.mu.Unlock() + + if len(e.buf) == 0 { + return + } + e.emit(string(e.buf)) + e.buf = e.buf[:0] +} + +func (e *otlpLineEmitter) emit(line string) { + now := time.Now() + + var record otellog.Record + record.SetTimestamp(now) + record.SetObservedTimestamp(now) + record.SetSeverity(otellog.SeverityInfo) + record.SetSeverityText("INFO") + record.SetBody(otellog.StringValue(line)) + record.AddAttributes(e.attrs...) + e.log.Emit(e.ctx, record) +} + +func otlpJobAttributes(e *Executor) []otellog.KeyValue { + buildID, _ := e.shell.Env.Get("BUILDKITE_BUILD_ID") + buildNumber, _ := e.shell.Env.Get("BUILDKITE_BUILD_NUMBER") + branch, _ := e.shell.Env.Get("BUILDKITE_BRANCH") + label, _ := e.shell.Env.Get("BUILDKITE_LABEL") + stepKey, _ := e.shell.Env.Get("BUILDKITE_STEP_KEY") + agentID, _ := e.shell.Env.Get("BUILDKITE_AGENT_ID") + + return []otellog.KeyValue{ + otellog.String("source", "job"), + otellog.String("buildkite.organization.slug", e.OrganizationSlug), + otellog.String("buildkite.pipeline.slug", e.PipelineSlug), + otellog.String("buildkite.branch", branch), + otellog.String("buildkite.queue", e.Queue), + otellog.String("buildkite.agent", e.AgentName), + otellog.String("buildkite.agent.id", agentID), + otellog.String("buildkite.build.id", buildID), + otellog.String("buildkite.build.number", buildNumber), + otellog.String("buildkite.job.id", e.JobID), + otellog.String("buildkite.job.label", label), + otellog.String("buildkite.job.key", stepKey), + } +} + +func appendLogAttrs(base []otellog.KeyValue, attrs map[string]string) []otellog.KeyValue { + if len(attrs) == 0 { + return base + } + out := append([]otellog.KeyValue{}, base...) + for key, value := range attrs { + if value != "" { + out = append(out, otellog.String(key, value)) + } + } + return out +} diff --git a/internal/job/otlp_job_logger_test.go b/internal/job/otlp_job_logger_test.go new file mode 100644 index 0000000000..5bc3ae6e6a --- /dev/null +++ b/internal/job/otlp_job_logger_test.go @@ -0,0 +1,316 @@ +package job + +import ( + "bytes" + "context" + "strings" + "sync" + "testing" + + "github.com/buildkite/agent/v3/internal/redact" + "github.com/buildkite/agent/v3/internal/replacer" + otellog "go.opentelemetry.io/otel/log" + "go.opentelemetry.io/otel/log/embedded" +) + +// captureLogger is a minimal otellog.Logger that records emitted record bodies. +type captureLogger struct { + embedded.Logger + mu sync.Mutex + bodies []string + lastKVs map[string]string +} + +func (c *captureLogger) Emit(_ context.Context, r otellog.Record) { + c.mu.Lock() + defer c.mu.Unlock() + c.bodies = append(c.bodies, r.Body().AsString()) + c.lastKVs = map[string]string{} + r.WalkAttributes(func(kv otellog.KeyValue) bool { + c.lastKVs[string(kv.Key)] = kv.Value.AsString() + return true + }) +} + +func (c *captureLogger) Enabled(context.Context, otellog.EnabledParameters) bool { return true } + +func newTestOTLPJobLogger(log otellog.Logger, needles ...string) *otlpJobLogger { + mux := replacer.NewMux(replacer.New(nil, needles, redact.Redacted)) + return &otlpJobLogger{ + log: log, + attrs: []otellog.KeyValue{otellog.String("source", "job")}, + redactors: mux, + } +} + +// TestOTLPJobLoggerRedactsSecrets ensures OTLP log records never carry secret +// values, matching the redaction applied to the customer-facing job log. +func TestOTLPJobLoggerRedactsSecrets(t *testing.T) { + t.Parallel() + + const secret = "supersekret-value" + cap := &captureLogger{} + l := newTestOTLPJobLogger(cap, secret) + + var downstream bytes.Buffer + w := l.Wrap(t.Context(), &downstream, map[string]string{"buildkite.phase": "command"}) + + if _, err := w.Write([]byte("before " + secret + " after\n")); err != nil { + t.Fatalf("Write() error = %v", err) + } + if f, ok := w.(interface{ Flush() }); ok { + f.Flush() + } + + cap.mu.Lock() + defer cap.mu.Unlock() + + if len(cap.bodies) != 1 { + t.Fatalf("emitted %d records, want 1: %q", len(cap.bodies), cap.bodies) + } + body := cap.bodies[0] + if strings.Contains(body, secret) { + t.Errorf("OTLP record body leaked secret: %q", body) + } + if !strings.Contains(body, "[REDACTED]") { + t.Errorf("OTLP record body = %q, want it to contain [REDACTED]", body) + } + if got := cap.lastKVs["buildkite.phase"]; got != "command" { + t.Errorf("buildkite.phase attribute = %q, want %q", got, "command") + } +} + +// TestOTLPJobLoggerRedactsNeedlesAddedMidStream ensures a secret added to the +// live redactor Mux after the command writer is created (e.g. via the Job API) +// is still redacted in OTLP output, not just secrets known at writer creation. +func TestOTLPJobLoggerRedactsNeedlesAddedMidStream(t *testing.T) { + t.Parallel() + + cap := &captureLogger{} + // Start with no needles, matching a command that adds a secret at runtime. + l := newTestOTLPJobLogger(cap) + + w := l.Wrap(t.Context(), &bytes.Buffer{}, nil) + + const secret = "late-bound-secret" + // Secret added to the live Mux after the writer was created. + l.redactors.Add(secret) + + if _, err := w.Write([]byte("value " + secret + " end\n")); err != nil { + t.Fatalf("Write() error = %v", err) + } + if f, ok := w.(interface{ Flush() }); ok { + f.Flush() + } + + cap.mu.Lock() + defer cap.mu.Unlock() + + if len(cap.bodies) != 1 { + t.Fatalf("emitted %d records, want 1: %q", len(cap.bodies), cap.bodies) + } + if body := cap.bodies[0]; strings.Contains(body, secret) { + t.Errorf("OTLP record body leaked mid-stream secret: %q", body) + } +} + +// TestOTLPJobLoggerControlWriter ensures bootstrap control output (section +// headers, prompts, comments) written through the control writer is emitted as +// OTLP log records, giving parity with the downloadable Buildkite job log. +func TestOTLPJobLoggerControlWriter(t *testing.T) { + t.Parallel() + + cap := &captureLogger{} + l := newTestOTLPJobLogger(cap) + + w := l.controlWriter() + if _, err := w.Write([]byte("~~~ Running commands\n$ echo hello\n")); err != nil { + t.Fatalf("Write() error = %v", err) + } + // A line without a trailing newline should only be emitted on Close/flush. + if _, err := w.Write([]byte("# trailing comment")); err != nil { + t.Fatalf("Write() error = %v", err) + } + if err := l.Close(); err != nil { + t.Fatalf("Close() error = %v", err) + } + + cap.mu.Lock() + defer cap.mu.Unlock() + + want := []string{"~~~ Running commands", "$ echo hello", "# trailing comment"} + if len(cap.bodies) != len(want) { + t.Fatalf("emitted %d records, want %d: %q", len(cap.bodies), len(want), cap.bodies) + } + for i, line := range want { + if cap.bodies[i] != line { + t.Errorf("record[%d] = %q, want %q", i, cap.bodies[i], line) + } + } +} + +// TestOTLPJobLoggerControlWriterReused ensures repeated controlWriter calls +// return the same emitter so all control output shares one line buffer. +func TestOTLPJobLoggerControlWriterReused(t *testing.T) { + t.Parallel() + + l := newTestOTLPJobLogger(&captureLogger{}) + if l.controlWriter() != l.controlWriter() { + t.Error("controlWriter() returned different writers; want a single shared emitter") + } +} + +// TestOTLPJobLoggerRedactsSecretsSplitAcrossWrites ensures a secret that is +// split across multiple Write calls is still redacted in the OTLP output. +func TestOTLPJobLoggerRedactsSecretsSplitAcrossWrites(t *testing.T) { + t.Parallel() + + const secret = "supersekret-value" + cap := &captureLogger{} + l := newTestOTLPJobLogger(cap, secret) + + w := l.Wrap(t.Context(), &bytes.Buffer{}, nil) + + // Split the secret across writes and withhold the trailing newline so the + // line is only emitted on Flush. + if _, err := w.Write([]byte("start super")); err != nil { + t.Fatalf("Write() error = %v", err) + } + if _, err := w.Write([]byte("sekret-value end\n")); err != nil { + t.Fatalf("Write() error = %v", err) + } + if f, ok := w.(interface{ Flush() }); ok { + f.Flush() + } + + cap.mu.Lock() + defer cap.mu.Unlock() + + if len(cap.bodies) != 1 { + t.Fatalf("emitted %d records, want 1: %q", len(cap.bodies), cap.bodies) + } + if body := cap.bodies[0]; strings.Contains(body, secret) { + t.Errorf("OTLP record body leaked secret split across writes: %q", body) + } +} + +// TestOTLPJobLoggerRedactsSecretsSplitAcrossCommands ensures the OTLP redactor +// retains partial matches across wrappers for sequential commands that share +// the same downstream job-log stream. +func TestOTLPJobLoggerRedactsSecretsSplitAcrossCommands(t *testing.T) { + t.Parallel() + + const secret = "supersekret-value" + cap := &captureLogger{} + l := newTestOTLPJobLogger(cap, secret) + var downstream bytes.Buffer + + first := l.Wrap(t.Context(), &downstream, map[string]string{"buildkite.phase": "checkout"}) + if _, err := first.Write([]byte("start super")); err != nil { + t.Fatalf("first Write() error = %v", err) + } + if f, ok := first.(interface{ Flush() }); ok { + f.Flush() + } + + second := l.Wrap(t.Context(), &downstream, map[string]string{"buildkite.phase": "command"}) + if _, err := second.Write([]byte("sekret-value end\n")); err != nil { + t.Fatalf("second Write() error = %v", err) + } + if f, ok := second.(interface{ Flush() }); ok { + f.Flush() + } + if err := l.Close(); err != nil { + t.Fatalf("Close() error = %v", err) + } + + cap.mu.Lock() + defer cap.mu.Unlock() + body := strings.Join(cap.bodies, "") + if strings.Contains(body, secret) || strings.Contains(body, "super") || strings.Contains(body, "sekret-value") { + t.Errorf("OTLP records leaked secret fragments across commands: %q", cap.bodies) + } + if !strings.Contains(body, "[REDACTED]") { + t.Errorf("OTLP records = %q, want them to contain [REDACTED]", cap.bodies) + } + if got := downstream.String(); got != "start "+secret+" end\n" { + t.Errorf("downstream = %q, want unmodified command output", got) + } +} + +// TestOTLPJobLoggerFlushRedactorsMatchesExecutorBoundary ensures an explicit +// executor redactor flush ends a partial match and emits it using the phase +// that produced it, while ordinary per-command flushes do not. +func TestOTLPJobLoggerFlushRedactorsMatchesExecutorBoundary(t *testing.T) { + t.Parallel() + + const secret = "supersekret-value" + cap := &captureLogger{} + l := newTestOTLPJobLogger(cap, secret) + var downstream bytes.Buffer + + first := l.Wrap(t.Context(), &downstream, map[string]string{"buildkite.phase": "hook"}) + if _, err := first.Write([]byte("start super")); err != nil { + t.Fatalf("first Write() error = %v", err) + } + if f, ok := first.(interface{ Flush() }); ok { + f.Flush() + } + l.FlushRedactors() + + second := l.Wrap(t.Context(), &downstream, map[string]string{"buildkite.phase": "command"}) + if _, err := second.Write([]byte("sekret-value end\n")); err != nil { + t.Fatalf("second Write() error = %v", err) + } + if f, ok := second.(interface{ Flush() }); ok { + f.Flush() + } + + cap.mu.Lock() + defer cap.mu.Unlock() + // The first command flush has already emitted the safe prefix. The + // executor boundary then releases the withheld partial match before the + // next phase starts. + want := []string{"start ", "super", "sekret-value end"} + if len(cap.bodies) != len(want) { + t.Fatalf("emitted records = %q, want %q", cap.bodies, want) + } + for i := range want { + if cap.bodies[i] != want[i] { + t.Errorf("record[%d] = %q, want %q", i, cap.bodies[i], want[i]) + } + } +} + +func TestOTLPJobLoggerChunksUnterminatedOutput(t *testing.T) { + t.Parallel() + + cap := &captureLogger{} + l := newTestOTLPJobLogger(cap) + w := l.Wrap(t.Context(), &bytes.Buffer{}, nil) + line := strings.Repeat("x", otlpLogRecordMaxBytes*2+17) + + if _, err := w.Write([]byte(line)); err != nil { + t.Fatalf("Write() error = %v", err) + } + + cap.mu.Lock() + if got, want := len(cap.bodies), 2; got != want { + cap.mu.Unlock() + t.Fatalf("records before Flush = %d, want %d", got, want) + } + cap.mu.Unlock() + + if f, ok := w.(interface{ Flush() }); ok { + f.Flush() + } + + cap.mu.Lock() + defer cap.mu.Unlock() + if got, want := len(cap.bodies), 3; got != want { + t.Fatalf("records after Flush = %d, want %d", got, want) + } + if got := strings.Join(cap.bodies, ""); got != line { + t.Errorf("rejoined record bodies differ from input: got %d bytes, want %d", len(got), len(line)) + } +} diff --git a/internal/job/tracing.go b/internal/job/tracing.go index e89f3b3743..c33dee7c2a 100644 --- a/internal/job/tracing.go +++ b/internal/job/tracing.go @@ -211,17 +211,14 @@ func (e *Executor) contextWithTraceparentIfEnabled(ctx context.Context) context. return ctx } - carrier := propagation.MapCarrier{ - "traceparent": e.TracingTraceParent, - } + carrier := propagation.MapCarrier{"traceparent": e.TracingTraceParent} // W3C tracestate is optional and only meaningful alongside traceparent. // The OTel TraceContext propagator already tolerates a missing key, so // this guard is purely to keep the carrier minimal. if e.TracingTraceState != "" { carrier["tracestate"] = e.TracingTraceState } - - return otel.GetTextMapPropagator().Extract(ctx, carrier) + return propagation.TraceContext{}.Extract(ctx, carrier) } func GenericTracingExtras(e *Executor, env *env.Environment) map[string]any { diff --git a/internal/replacer/mux.go b/internal/replacer/mux.go index 28b7fa4305..3353798370 100644 --- a/internal/replacer/mux.go +++ b/internal/replacer/mux.go @@ -37,6 +37,16 @@ func (m *Mux) Append(r *Replacer) { m.underlying = append(m.underlying, r) } +// Needles returns the current set of needles (secrets). All replacers in a Mux +// are kept in sync by Add/Reset, so the needles of the first replacer are +// representative of the whole Mux. Returns nil if the Mux is empty. +func (m *Mux) Needles() []string { + if len(m.underlying) == 0 { + return nil + } + return m.underlying[0].Needles() +} + // Flush flushes all replacers. func (m *Mux) Flush() error { errs := make([]error, 0, len(m.underlying)) diff --git a/internal/shell/shell.go b/internal/shell/shell.go index bd3bd79c4b..1fbbd2e992 100644 --- a/internal/shell/shell.go +++ b/internal/shell/shell.go @@ -13,6 +13,7 @@ import ( "os/exec" "path" "path/filepath" + "reflect" "runtime" "strings" "sync/atomic" @@ -81,6 +82,8 @@ type Shell struct { // Defaults to [os.Stdout]. stdout io.Writer + outputInterceptor OutputInterceptor + // How to encode trace contexts. traceContextCodec tracetools.Codec @@ -90,14 +93,23 @@ type Shell struct { type NewShellOpt = func(*Shell) +type OutputInterceptor func(context.Context, io.Writer, map[string]string) io.Writer + func WithCommandLog(log *[][]string) NewShellOpt { return func(s *Shell) { s.commandLog = log } } func WithDebug(d bool) NewShellOpt { return func(s *Shell) { s.debug = d } } func WithDryRun(d bool) NewShellOpt { return func(s *Shell) { s.dryRun = d } } func WithEnv(e *env.Environment) NewShellOpt { return func(s *Shell) { s.Env = e } } func WithLogger(l Logger) NewShellOpt { return func(s *Shell) { s.Logger = l } } -func WithPTY(pty bool) NewShellOpt { return func(s *Shell) { s.pty = pty } } -func WithStdout(w io.Writer) NewShellOpt { return func(s *Shell) { s.stdout = w } } -func WithWD(wd string) NewShellOpt { return func(s *Shell) { s.wd = wd } } +func WithOutputInterceptor(i OutputInterceptor) NewShellOpt { + return func(s *Shell) { s.outputInterceptor = i } +} +func WithPTY(pty bool) NewShellOpt { return func(s *Shell) { s.pty = pty } } +func WithStdout(w io.Writer) NewShellOpt { return func(s *Shell) { s.stdout = w } } +func WithWD(wd string) NewShellOpt { return func(s *Shell) { s.wd = wd } } + +func (s *Shell) SetOutputInterceptor(i OutputInterceptor) { + s.outputInterceptor = i +} func WithInterruptSignal(sig process.Signal) NewShellOpt { return func(s *Shell) { s.interruptSignal = sig } @@ -161,6 +173,7 @@ func (s *Shell) CloneWithStdin(r io.Reader) *Shell { Env: s.Env, stdin: r, // our new stdin stdout: s.stdout, + outputInterceptor: s.outputInterceptor, wd: s.wd, interruptSignal: s.interruptSignal, signalGracePeriod: s.signalGracePeriod, @@ -461,8 +474,40 @@ func (c Command) Run(ctx context.Context, opts ...RunCommandOpt) error { stderr = io.Discard } - // If we're performing a string search, wrap the current stdout and stderr - // in olfactors, and report which ones were detected through the map. + var flushers []interface{ Flush() } + if c.shell.outputInterceptor != nil { + // stdout and stderr normally share the same downstream job-log writer. + // Preserve that shared stream through the interceptor as well so + // stateful processing such as redaction can match values split across + // writes to the two file descriptors. + if cfg.captureStdout == nil && sameWriter(stdout, stderr) && stdout != nil && stdout != io.Discard { + stdout = c.shell.outputInterceptor(ctx, stdout, cfg.outputAttrs) + stderr = stdout + if f, ok := stdout.(interface{ Flush() }); ok { + flushers = append(flushers, f) + } + } else { + if cfg.captureStdout == nil && stdout != nil && stdout != io.Discard { + stdout = c.shell.outputInterceptor(ctx, stdout, cfg.outputAttrs) + if f, ok := stdout.(interface{ Flush() }); ok { + flushers = append(flushers, f) + } + } + if stderr != nil && stderr != io.Discard { + stderr = c.shell.outputInterceptor(ctx, stderr, cfg.outputAttrs) + if f, ok := stderr.(interface{ Flush() }); ok { + flushers = append(flushers, f) + } + } + } + } + + // If we're performing a string search, wrap the intercepted stdout and + // stderr in olfactors and report which ones were detected through the map. + // Interception must happen first: olfactor uses separate wrappers for each + // stream, which would otherwise hide that both ultimately share the same + // job-log writer and incorrectly split stateful processing such as OTLP + // redaction. if cfg.smells != nil { smells := make([]string, 0, len(cfg.smells)) for s := range cfg.smells { @@ -481,6 +526,11 @@ func (c Command) Run(ctx context.Context, opts ...RunCommandOpt) error { } }() } + defer func() { + for _, f := range flushers { + f.Flush() + } + }() cmdCfg.Started = cfg.started cmdCfg.Done = cfg.done @@ -488,6 +538,14 @@ func (c Command) Run(ctx context.Context, opts ...RunCommandOpt) error { return c.shell.executeCommand(ctx, cmdCfg, stdout, stderr, pty) } +func sameWriter(a, b io.Writer) bool { + if a == nil || b == nil { + return a == nil && b == nil + } + av, bv := reflect.ValueOf(a), reflect.ValueOf(b) + return av.Type() == bv.Type() && av.Comparable() && av.Interface() == bv.Interface() +} + // RunAndCaptureStdout is Run, but automatically sets options: // * ShowPrompt(false) (overridable) // * CaptureStdout() (not overridable) @@ -507,6 +565,7 @@ type runConfig struct { started chan struct{} done chan struct{} extraEnv *env.Environment + outputAttrs map[string]string smells map[string]bool } @@ -530,6 +589,10 @@ func ShowPrompt(show bool) RunCommandOpt { return func(c *runConfig) { c.showPro // WithExtraEnv can be used to set additional env vars for this run. func WithExtraEnv(e *env.Environment) RunCommandOpt { return func(c *runConfig) { c.extraEnv = e } } +func WithOutputAttributes(attrs map[string]string) RunCommandOpt { + return func(c *runConfig) { c.outputAttrs = attrs } +} + // WithStarted provides a channel that is closed after the command has started. func WithStarted(started chan struct{}) RunCommandOpt { return func(c *runConfig) { c.started = started } diff --git a/internal/shell/shell_test.go b/internal/shell/shell_test.go index 9c9ff1271f..5c9108f217 100644 --- a/internal/shell/shell_test.go +++ b/internal/shell/shell_test.go @@ -11,6 +11,7 @@ import ( "path/filepath" "runtime" "strings" + "sync" "syscall" "testing" "time" @@ -22,6 +23,17 @@ import ( "github.com/google/go-cmp/cmp" ) +type synchronizedWriter struct { + mu sync.Mutex + out io.Writer +} + +func (w *synchronizedWriter) Write(p []byte) (int, error) { + w.mu.Lock() + defer w.mu.Unlock() + return w.out.Write(p) +} + func TestRunAndCaptureWithTTY(t *testing.T) { t.Parallel() @@ -143,6 +155,76 @@ func TestRunWithStdin(t *testing.T) { } } +func TestCloneWithStdinPreservesOutputInterceptor(t *testing.T) { + t.Parallel() + + out := &bytes.Buffer{} + intercepted := &bytes.Buffer{} + sh := newShellForTest(t, + shell.WithStdout(out), + shell.WithPTY(false), + shell.WithOutputInterceptor(func(_ context.Context, downstream io.Writer, _ map[string]string) io.Writer { + return io.MultiWriter(downstream, intercepted) + }), + ) + + cmd := sh.CloneWithStdin(strings.NewReader("hello stdin")).Command("tr", "hs", "HS") + if err := cmd.Run(t.Context()); err != nil { + t.Fatalf(`sh.CloneWithStdin("hello stdin").Command("tr", "hs", "HS").Run(ctx) error = %v`, err) + } + if got, want := out.String(), "Hello Stdin"; got != want { + t.Errorf("stdout = %q, want %q", got, want) + } + if got, want := intercepted.String(), "Hello Stdin"; got != want { + t.Errorf("intercepted stdout = %q, want %q", got, want) + } +} + +func TestRunSharesOutputInterceptorForCombinedStdoutAndStderr(t *testing.T) { + t.Parallel() + + proxy, err := bintest.CompileProxy("combined-output") + if err != nil { + t.Fatalf("bintest.CompileProxy(combined-output) error = %v", err) + } + defer func() { + if err := proxy.Close(); err != nil { + t.Errorf("proxy.Close() error = %v", err) + } + }() + + out := &bytes.Buffer{} + interceptorCalls := 0 + smelled := map[string]bool{"stdout": false, "stderr": false} + sh := newShellForTest(t, + shell.WithStdout(out), + shell.WithPTY(false), + shell.WithOutputInterceptor(func(_ context.Context, downstream io.Writer, _ map[string]string) io.Writer { + interceptorCalls++ + return &synchronizedWriter{out: downstream} + }), + ) + + go func() { + call := <-proxy.Ch + _, _ = io.WriteString(call.Stdout, "stdout") + _, _ = io.WriteString(call.Stderr, "stderr") + call.Exit(0) + }() + + if err := sh.Command(proxy.Path).Run(t.Context(), shell.WithStringSearch(smelled)); err != nil { + t.Fatalf("Command.Run() error = %v", err) + } + if got, want := interceptorCalls, 1; got != want { + t.Errorf("output interceptor calls = %d, want %d for shared stdout/stderr with string search", got, want) + } + for output, found := range smelled { + if !found { + t.Errorf("WithStringSearch did not find %q", output) + } + } +} + func TestContextCancelInterrupts(t *testing.T) { t.Parallel()