From 86e91e010df80e9f617a038d7478afe938723ef2 Mon Sep 17 00:00:00 2001 From: Guillaume Lours <705411+glours@users.noreply.github.com> Date: Fri, 24 Oct 2025 16:32:45 +0200 Subject: [PATCH] Add streamOverrideWrapper to intercepts command.Cli stream methods and transparently returns custom streams when provided via options Add new GetConfiguredStreams function to Compose API definition Signed-off-by: Guillaume Lours <705411+glours@users.noreply.github.com> --- cmd/compose/logs.go | 3 +- cmd/compose/up.go | 3 +- cmd/compose/watch.go | 3 +- pkg/api/api.go | 3 ++ pkg/compose/build.go | 2 +- pkg/compose/build_bake.go | 2 +- pkg/compose/compose.go | 75 +++++++++++++++++++++++++++++++-------- 7 files changed, 72 insertions(+), 19 deletions(-) diff --git a/cmd/compose/logs.go b/cmd/compose/logs.go index f0704540c..4931d9dca 100644 --- a/cmd/compose/logs.go +++ b/cmd/compose/logs.go @@ -86,11 +86,12 @@ func runLogs(ctx context.Context, dockerCli command.Cli, backendOptions *Backend } } - consumer := formatter.NewLogConsumer(ctx, dockerCli.Out(), dockerCli.Err(), !opts.noColor, !opts.noPrefix, false) backend, err := compose.NewComposeService(dockerCli, backendOptions.Options...) if err != nil { return err } + outStream, errStream, _ := backend.GetConfiguredStreams() + consumer := formatter.NewLogConsumer(ctx, outStream, errStream, !opts.noColor, !opts.noPrefix, false) return backend.Logs(ctx, name, consumer, api.LogOptions{ Project: project, Services: services, diff --git a/cmd/compose/up.go b/cmd/compose/up.go index a63aed46f..08ea18372 100644 --- a/cmd/compose/up.go +++ b/cmd/compose/up.go @@ -293,7 +293,8 @@ func runUp( var consumer api.LogConsumer var attach []string if !upOptions.Detach { - consumer = formatter.NewLogConsumer(ctx, dockerCli.Out(), dockerCli.Err(), !upOptions.noColor, !upOptions.noPrefix, upOptions.timestamp) + outStream, errStream, _ := backend.GetConfiguredStreams() + consumer = formatter.NewLogConsumer(ctx, outStream, errStream, !upOptions.noColor, !upOptions.noPrefix, upOptions.timestamp) var attachSet utils.Set[string] if len(upOptions.attach) != 0 { diff --git a/cmd/compose/watch.go b/cmd/compose/watch.go index 580ddf5a7..68d9004e0 100644 --- a/cmd/compose/watch.go +++ b/cmd/compose/watch.go @@ -121,11 +121,12 @@ func runWatch(ctx context.Context, dockerCli command.Cli, backendOptions *Backen } } - consumer := formatter.NewLogConsumer(ctx, dockerCli.Out(), dockerCli.Err(), false, false, false) backend, err := compose.NewComposeService(dockerCli, backendOptions.Options...) if err != nil { return err } + outStream, errStream, _ := backend.GetConfiguredStreams() + consumer := formatter.NewLogConsumer(ctx, outStream, errStream, false, false, false) return backend.Watch(ctx, project, api.WatchOptions{ Build: &build, LogTo: consumer, diff --git a/pkg/api/api.go b/pkg/api/api.go index ebb0a767b..ec800c6cc 100644 --- a/pkg/api/api.go +++ b/pkg/api/api.go @@ -99,6 +99,9 @@ type Compose interface { Generate(ctx context.Context, options GenerateOptions) (*types.Project, error) // Volumes executes the equivalent to a `docker volume ls` Volumes(ctx context.Context, project string, options VolumesOptions) ([]VolumesSummary, error) + // GetConfiguredStreams returns the configured I/O streams (stdout, stderr, stdin). + // If no custom streams were configured, it returns the dockerCli streams. + GetConfiguredStreams() (stdout io.Writer, stderr io.Writer, stdin io.Reader) } type VolumesOptions struct { diff --git a/pkg/compose/build.go b/pkg/compose/build.go index 8e2b3e471..986761cd4 100644 --- a/pkg/compose/build.go +++ b/pkg/compose/build.go @@ -177,7 +177,7 @@ func (s *composeService) build(ctx context.Context, project *types.Project, opti if options.Progress == progress.ModeAuto { options.Progress = os.Getenv("BUILDKIT_PROGRESS") } - w, err = xprogress.NewPrinter(progressCtx, os.Stdout, progressui.DisplayMode(options.Progress), + w, err = xprogress.NewPrinter(progressCtx, s.stdout(), progressui.DisplayMode(options.Progress), xprogress.WithDesc( fmt.Sprintf("building with %q instance using %s driver", b.Name, b.Driver), fmt.Sprintf("%s:%s", b.Driver, b.Name), diff --git a/pkg/compose/build_bake.go b/pkg/compose/build_bake.go index a793151f3..04fa8e6c5 100644 --- a/pkg/compose/build_bake.go +++ b/pkg/compose/build_bake.go @@ -142,7 +142,7 @@ func (s *composeService) doBuildBake(ctx context.Context, project *types.Project if displayMode == progress.ModeAuto && !s.stdout().IsTerminal() { displayMode = progressui.PlainMode } - out = os.Stdout // should be s.stdout(), but NewDisplay require access to the underlying *File + out = s.stdout() } display, err := progressui.NewDisplay(out, displayMode) if err != nil { diff --git a/pkg/compose/compose.go b/pkg/compose/compose.go index fd096f06c..c44849d06 100644 --- a/pkg/compose/compose.go +++ b/pkg/compose/compose.go @@ -94,6 +94,12 @@ func NewComposeService(dockerCli command.Cli, options ...Option) (api.Compose, e return defaultValue, nil } } + + // If custom streams were provided, wrap the Docker CLI to use them + if s.outStream != nil || s.errStream != nil || s.inStream != nil { + s.dockerCli = s.wrapDockerCliWithStreams(dockerCli) + } + return s, nil } @@ -141,7 +147,7 @@ func WithContextInfo(info api.ContextInfo) Option { // WithProxyConfig sets custom HTTP proxy configuration for builds func WithProxyConfig(config map[string]string) Option { - return func(s *composeService) error{ + return func(s *composeService) error { s.proxyConfig = config return nil } @@ -238,28 +244,15 @@ func (s *composeService) getProxyConfig() map[string]string { return storeutil.GetProxyConfig(s.dockerCli) } - func (s *composeService) stdout() *streams.Out { - // If stream overrides are provided, use them - if s.outStream != nil { - return streams.NewOut(s.outStream) - } return s.dockerCli.Out() } func (s *composeService) stdin() *streams.In { - // If stream overrides are provided, use them - if s.inStream != nil { - return streams.NewIn(&readCloserAdapter{r: s.inStream}) - } return s.dockerCli.In() } func (s *composeService) stderr() *streams.Out { - // If stream overrides are provided, use them - if s.errStream != nil { - return streams.NewOut(s.errStream) - } return s.dockerCli.Err() } @@ -270,6 +263,11 @@ func (s *composeService) stdinfo() *streams.Out { return s.stderr() } +// GetConfiguredStreams returns the configured I/O streams (implements api.Compose interface) +func (s *composeService) GetConfiguredStreams() (io.Writer, io.Writer, io.Reader) { + return s.stdout(), s.stderr(), s.stdin() +} + // readCloserAdapter adapts io.Reader to io.ReadCloser type readCloserAdapter struct { r io.Reader @@ -283,6 +281,55 @@ func (r *readCloserAdapter) Close() error { return nil } +// wrapDockerCliWithStreams wraps the Docker CLI to intercept and override stream methods +func (s *composeService) wrapDockerCliWithStreams(baseCli command.Cli) command.Cli { + wrapper := &streamOverrideWrapper{ + Cli: baseCli, + } + + // Wrap custom streams in Docker CLI's stream types + if s.outStream != nil { + wrapper.outStream = streams.NewOut(s.outStream) + } + if s.errStream != nil { + wrapper.errStream = streams.NewOut(s.errStream) + } + if s.inStream != nil { + wrapper.inStream = streams.NewIn(&readCloserAdapter{r: s.inStream}) + } + + return wrapper +} + +// streamOverrideWrapper wraps command.Cli to override streams with custom implementations +type streamOverrideWrapper struct { + command.Cli + outStream *streams.Out + errStream *streams.Out + inStream *streams.In +} + +func (w *streamOverrideWrapper) Out() *streams.Out { + if w.outStream != nil { + return w.outStream + } + return w.Cli.Out() +} + +func (w *streamOverrideWrapper) Err() *streams.Out { + if w.errStream != nil { + return w.errStream + } + return w.Cli.Err() +} + +func (w *streamOverrideWrapper) In() *streams.In { + if w.inStream != nil { + return w.inStream + } + return w.Cli.In() +} + func getCanonicalContainerName(c container.Summary) string { if len(c.Names) == 0 { // corner case, sometime happens on removal. return short ID as a safeguard value