stack: add 'docker stack logs' command

cli/command/stack/client_test.go

@@ -2,6 +2,7 @@ package stack
 
 import (
 	"context"
+	"io"
 	"strings"
 
 	"github.com/docker/cli/cli/compose/convert"
@@ -24,6 +25,7 @@ type fakeClient struct {
 	removedConfigs  []string
 
 	serviceListFunc   func(options client.ServiceListOptions) (client.ServiceListResult, error)
+	serviceLogsFunc   func(serviceID string, options client.ServiceLogsOptions) (client.ServiceLogsResult, error)
 	networkListFunc   func(options client.NetworkListOptions) (client.NetworkListResult, error)
 	secretListFunc    func(options client.SecretListOptions) (client.SecretListResult, error)
 	configListFunc    func(options client.ConfigListOptions) (client.ConfigListResult, error)
@@ -185,6 +187,13 @@ func (*fakeClient) ServiceInspect(_ context.Context, serviceID string, _ client.
 	}, nil
 }
 
+func (cli *fakeClient) ServiceLogs(_ context.Context, serviceID string, options client.ServiceLogsOptions) (client.ServiceLogsResult, error) {
+	if cli.serviceLogsFunc != nil {
+		return cli.serviceLogsFunc(serviceID, options)
+	}
+	return io.NopCloser(strings.NewReader("")), nil
+}
+
 func serviceFromName(name string) swarm.Service {
 	return swarm.Service{
 		ID: "ID-" + name,

cli/command/stack/cmd.go

@@ -38,6 +38,7 @@ func newStackCommand(dockerCLI command.Cli) *cobra.Command {
 	cmd.AddCommand(
 		newDeployCommand(dockerCLI),
 		newListCommand(dockerCLI),
+		newLogsCommand(dockerCLI),
 		newPsCommand(dockerCLI),
 		newRemoveCommand(dockerCLI),
 		newServicesCommand(dockerCLI),

cli/command/stack/logs.go

@@ -0,0 +1,118 @@
+package stack
+
+import (
+	"context"
+	"fmt"
+	"sync"
+
+	"github.com/docker/cli/cli"
+	"github.com/docker/cli/cli/command"
+	"github.com/moby/moby/api/pkg/stdcopy"
+	"github.com/moby/moby/api/types/swarm"
+	"github.com/moby/moby/client"
+	"github.com/spf13/cobra"
+)
+
+// logsOptions holds docker stack logs options
+type logsOptions struct {
+	namespace  string
+	follow     bool
+	since      string
+	tail       string
+	timestamps bool
+	noColor    bool
+}
+
+func newLogsCommand(dockerCLI command.Cli) *cobra.Command {
+	var opts logsOptions
+
+	cmd := &cobra.Command{
+		Use:   "logs [OPTIONS] STACK",
+		Short: "Fetch aggregated logs of all services in the stack",
+		Args:  cli.ExactArgs(1),
+		RunE: func(cmd *cobra.Command, args []string) error {
+			opts.namespace = args[0]
+			if err := validateStackName(opts.namespace); err != nil {
+				return err
+			}
+			return runLogs(cmd.Context(), dockerCLI, opts)
+		},
+		Annotations:           map[string]string{"version": "1.29"},
+		ValidArgsFunction:     completeNames(dockerCLI),
+		DisableFlagsInUseLine: true,
+	}
+	flags := cmd.Flags()
+	flags.BoolVarP(&opts.follow, "follow", "f", false, "Follow log output")
+	flags.StringVar(&opts.since, "since", "", `Show logs since timestamp (e.g. "2013-01-02T13:23:37Z") or relative (e.g. "42m" for 42 minutes)`)
+	flags.StringVarP(&opts.tail, "tail", "n", "all", "Number of lines to show from the end of the logs (per service)")
+	flags.BoolVarP(&opts.timestamps, "timestamps", "t", false, "Show timestamps")
+	flags.BoolVar(&opts.noColor, "no-color", false, "Produce monochrome output")
+	return cmd
+}
+
+// runLogs is the swarm implementation of docker stack logs. It streams the
+// logs of every service in the stack concurrently, prefixing each line with
+// the (colored) service name, similar to "docker compose logs".
+func runLogs(ctx context.Context, dockerCLI command.Cli, opts logsOptions) error {
+	apiClient := dockerCLI.Client()
+
+	res, err := getStackServices(ctx, apiClient, opts.namespace)
+	if err != nil {
+		return err
+	}
+	if len(res.Items) == 0 {
+		return fmt.Errorf("nothing found in stack: %s", opts.namespace)
+	}
+
+	maxLen := 0
+	for _, s := range res.Items {
+		if n := len(s.Spec.Name); n > maxLen {
+			maxLen = n
+		}
+	}
+
+	var (
+		wg sync.WaitGroup
+		mu sync.Mutex // serializes output lines across services
+	)
+	errs := make([]error, len(res.Items))
+	for i, service := range res.Items {
+		wg.Add(1)
+		go func(idx int, s swarm.Service) {
+			defer wg.Done()
+			if err := streamServiceLogs(ctx, apiClient, dockerCLI, s, idx, maxLen, &mu, opts); err != nil {
+				errs[idx] = fmt.Errorf("%s: %w", s.Spec.Name, err)
+			}
+		}(i, service)
+	}
+	wg.Wait()
+	for _, err := range errs {
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func streamServiceLogs(ctx context.Context, apiClient client.APIClient, dockerCLI command.Cli, s swarm.Service, idx, maxLen int, mu *sync.Mutex, opts logsOptions) error {
+	body, err := apiClient.ServiceLogs(ctx, s.ID, client.ServiceLogsOptions{
+		ShowStdout: true,
+		ShowStderr: true,
+		Follow:     opts.follow,
+		Since:      opts.since,
+		Tail:       opts.tail,
+		Timestamps: opts.timestamps,
+	})
+	if err != nil {
+		return err
+	}
+	defer body.Close()
+
+	prefix := logPrefix(s.Spec.Name, idx, maxLen, opts.noColor)
+	stdout := newPrefixWriter(dockerCLI.Out(), prefix, mu)
+	stderr := newPrefixWriter(dockerCLI.Err(), prefix, mu)
+	// Service logs are always multiplexed (services with a TTY are not
+	// supported by the service logs endpoint without --raw).
+	_, err = stdcopy.StdCopy(stdout, stderr, body)
+	return err
+}

cli/command/stack/logs_test.go

@@ -0,0 +1,165 @@
+package stack
+
+import (
+	"bytes"
+	"encoding/binary"
+	"errors"
+	"io"
+	"sync"
+	"testing"
+
+	"github.com/docker/cli/internal/test"
+	"github.com/moby/moby/api/types/swarm"
+	"github.com/moby/moby/client"
+	"gotest.tools/v3/assert"
+	is "gotest.tools/v3/assert/cmp"
+)
+
+func TestStackLogsErrors(t *testing.T) {
+	testCases := []struct {
+		doc             string
+		args            []string
+		serviceListFunc func(options client.ServiceListOptions) (client.ServiceListResult, error)
+		serviceLogsFunc func(serviceID string, options client.ServiceLogsOptions) (client.ServiceLogsResult, error)
+		expectedError   string
+	}{
+		{
+			doc:           "no args",
+			args:          []string{},
+			expectedError: "requires 1 argument",
+		},
+		{
+			doc:           "too many args",
+			args:          []string{"foo", "bar"},
+			expectedError: "requires 1 argument",
+		},
+		{
+			doc:           "invalid stack name",
+			args:          []string{"  "},
+			expectedError: "invalid stack name",
+		},
+		{
+			doc:  "service list error",
+			args: []string{"foo"},
+			serviceListFunc: func(client.ServiceListOptions) (client.ServiceListResult, error) {
+				return client.ServiceListResult{}, errors.New("error getting services")
+			},
+			expectedError: "error getting services",
+		},
+		{
+			doc:           "empty stack",
+			args:          []string{"emptystack"},
+			expectedError: "nothing found in stack",
+		},
+		{
+			doc:  "service logs error",
+			args: []string{"foo"},
+			serviceListFunc: func(client.ServiceListOptions) (client.ServiceListResult, error) {
+				return client.ServiceListResult{
+					Items: []swarm.Service{serviceFromName("foo_web")},
+				}, nil
+			},
+			serviceLogsFunc: func(string, client.ServiceLogsOptions) (client.ServiceLogsResult, error) {
+				return nil, errors.New("error getting logs")
+			},
+			expectedError: "error getting logs",
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.doc, func(t *testing.T) {
+			cmd := newLogsCommand(test.NewFakeCli(&fakeClient{
+				serviceListFunc: tc.serviceListFunc,
+				serviceLogsFunc: tc.serviceLogsFunc,
+			}))
+			cmd.SetArgs(tc.args)
+			cmd.SetOut(io.Discard)
+			cmd.SetErr(io.Discard)
+			assert.ErrorContains(t, cmd.Execute(), tc.expectedError)
+		})
+	}
+}
+
+// muxStdout wraps payload in the stdcopy stream format (stdout frame), as
+// returned by the service logs endpoint for services without a TTY.
+func muxStdout(payload string) io.ReadCloser {
+	var buf bytes.Buffer
+	hdr := make([]byte, 8)
+	hdr[0] = 1 // stdout
+	binary.BigEndian.PutUint32(hdr[4:], uint32(len(payload)))
+	buf.Write(hdr)
+	buf.WriteString(payload)
+	return io.NopCloser(&buf)
+}
+
+func TestStackLogsPrefixesOutput(t *testing.T) {
+	logs := map[string]string{
+		"ID-mystack_web": "hello from web\n",
+		"ID-mystack_db":  "hello from db\n",
+	}
+
+	cli := test.NewFakeCli(&fakeClient{
+		services: []string{"mystack_web", "mystack_db"},
+		serviceLogsFunc: func(serviceID string, _ client.ServiceLogsOptions) (client.ServiceLogsResult, error) {
+			return muxStdout(logs[serviceID]), nil
+		},
+	})
+	cmd := newLogsCommand(cli)
+	cmd.SetArgs([]string{"--no-color", "mystack"})
+	cmd.SetOut(io.Discard)
+	cmd.SetErr(io.Discard)
+	assert.NilError(t, cmd.Execute())
+
+	out := cli.OutBuffer().String()
+	assert.Check(t, is.Contains(out, "mystack_web | hello from web\n"))
+	assert.Check(t, is.Contains(out, "mystack_db  | hello from db\n"))
+}
+
+func TestStackLogsPassesOptions(t *testing.T) {
+	var got client.ServiceLogsOptions
+	cli := test.NewFakeCli(&fakeClient{
+		services: []string{"mystack_web"},
+		serviceLogsFunc: func(_ string, options client.ServiceLogsOptions) (client.ServiceLogsResult, error) {
+			got = options
+			return io.NopCloser(bytes.NewReader(nil)), nil
+		},
+	})
+	cmd := newLogsCommand(cli)
+	cmd.SetArgs([]string{"--since", "42m", "--tail", "10", "--timestamps", "mystack"})
+	cmd.SetOut(io.Discard)
+	cmd.SetErr(io.Discard)
+	assert.NilError(t, cmd.Execute())
+
+	assert.Check(t, got.ShowStdout)
+	assert.Check(t, got.ShowStderr)
+	assert.Check(t, is.Equal(got.Since, "42m"))
+	assert.Check(t, is.Equal(got.Tail, "10"))
+	assert.Check(t, got.Timestamps)
+	assert.Check(t, !got.Follow)
+}
+
+func TestPrefixWriter(t *testing.T) {
+	var out bytes.Buffer
+	var mu sync.Mutex
+	w := newPrefixWriter(&out, []byte("P | "), &mu)
+
+	// partial writes must be buffered until a newline arrives
+	_, err := w.Write([]byte("a"))
+	assert.NilError(t, err)
+	assert.Check(t, is.Equal(out.String(), ""))
+
+	_, err = w.Write([]byte("b\nc"))
+	assert.NilError(t, err)
+	assert.Check(t, is.Equal(out.String(), "P | ab\n"))
+
+	_, err = w.Write([]byte("\n"))
+	assert.NilError(t, err)
+	assert.Check(t, is.Equal(out.String(), "P | ab\nP | c\n"))
+}
+
+func TestLogPrefix(t *testing.T) {
+	assert.Check(t, is.Equal(string(logPrefix("web", 0, 5, true)), "web   | "))
+	colored := string(logPrefix("web", 0, 3, false))
+	assert.Check(t, is.Contains(colored, "\x1b["))
+	assert.Check(t, is.Contains(colored, "web | "))
+}

cli/command/stack/logs_writer.go

@@ -0,0 +1,55 @@
+package stack
+
+import (
+	"bytes"
+	"fmt"
+	"io"
+	"sync"
+)
+
+// logColors is the rotating palette used to distinguish services, in the
+// same spirit as "docker compose logs".
+var logColors = []string{"36", "33", "32", "35", "34", "96", "93", "92", "95", "94"}
+
+// logPrefix returns the (optionally colored) per-line prefix for a service,
+// padded so that log lines of all services align.
+func logPrefix(name string, idx, maxLen int, noColor bool) []byte {
+	padded := fmt.Sprintf("%-*s | ", maxLen, name)
+	if noColor {
+		return []byte(padded)
+	}
+	c := logColors[idx%len(logColors)]
+	return []byte(fmt.Sprintf("\x1b[%sm%s\x1b[0m", c, padded))
+}
+
+// prefixWriter prefixes every complete line written to it. Partial lines are
+// buffered until a newline arrives, so a line is never split between two
+// writes, and the shared mutex keeps lines from concurrent services intact.
+type prefixWriter struct {
+	out    io.Writer
+	prefix []byte
+	mu     *sync.Mutex
+	buf    bytes.Buffer
+}
+
+func newPrefixWriter(out io.Writer, prefix []byte, mu *sync.Mutex) *prefixWriter {
+	return &prefixWriter{out: out, prefix: prefix, mu: mu}
+}
+
+func (w *prefixWriter) Write(p []byte) (int, error) {
+	w.buf.Write(p)
+	for {
+		line, err := w.buf.ReadBytes('\n')
+		if err != nil {
+			// no complete line yet: keep the remainder buffered
+			w.buf.Write(line)
+			return len(p), nil
+		}
+		w.mu.Lock()
+		_, werr := w.out.Write(append(w.prefix, line...))
+		w.mu.Unlock()
+		if werr != nil {
+			return len(p), werr
+		}
+	}
+}

docs/reference/commandline/stack.md

@@ -9,6 +9,7 @@ Manage Swarm stacks
 |:--------------------------------|:---------------------------------------------------------------------|
 | [`config`](stack_config.md)     | Outputs the final config file, after doing merges and interpolations |
 | [`deploy`](stack_deploy.md)     | Deploy a new stack or update an existing stack                       |
+| [`logs`](stack_logs.md)         | Fetch aggregated logs of all services in the stack                   |
 | [`ls`](stack_ls.md)             | List stacks                                                          |
 | [`ps`](stack_ps.md)             | List the tasks in the stack                                          |
 | [`rm`](stack_rm.md)             | Remove one or more stacks                                            |