Add tail and since to service logs

This change adds the ability to do --tail and --since on docker service
logs. It wires up the API endpoints to each other and fixes some older
bugs. It adds integration tests for these new features.

Signed-off-by: Drew Erny <drew.erny@docker.com>

cli/command/service/logs.go

@@ -45,7 +45,7 @@ func newLogsCommand(dockerCli *command.DockerCli) *cobra.Command {
 	flags := cmd.Flags()
 	flags.BoolVar(&opts.noResolve, "no-resolve", false, "Do not map IDs to Names")
 	flags.BoolVarP(&opts.follow, "follow", "f", false, "Follow log output")
-	flags.StringVar(&opts.since, "since", "", "Show logs since timestamp")
+	flags.StringVar(&opts.since, "since", "", "Show logs since timestamp (e.g. 2013-01-02T13:23:37) or relative (e.g. 42m for 42 minutes)")
 	flags.BoolVarP(&opts.timestamps, "timestamps", "t", false, "Show timestamps")
 	flags.BoolVar(&opts.details, "details", false, "Show extra details provided to logs")
 	flags.StringVar(&opts.tail, "tail", "all", "Number of lines to show from the end of the logs")

daemon/cluster/executor/container/adapter.go

@@ -417,7 +417,10 @@ func (c *containerAdapter) logs(ctx context.Context, options api.LogSubscription
 		if err != nil {
 			return nil, err
 		}
-		apiOptions.Since = since.Format(time.RFC3339Nano)
+		// print since as this formatted string because the docker container
+		// logs interface expects it like this.
+		// see github.com/docker/docker/api/types/time.ParseTimestamps
+		apiOptions.Since = fmt.Sprintf("%d.%09d", since.Unix(), int64(since.Nanosecond()))
 	}
 
 	if options.Tail < 0 {

daemon/cluster/services.go

@@ -6,7 +6,9 @@ import (
 	"fmt"
 	"io"
 	"os"
+	"strconv"
 	"strings"
+	"time"
 
 	"github.com/Sirupsen/logrus"
 	"github.com/docker/distribution/reference"
@@ -14,6 +16,7 @@ import (
 	apitypes "github.com/docker/docker/api/types"
 	"github.com/docker/docker/api/types/backend"
 	types "github.com/docker/docker/api/types/swarm"
+	timetypes "github.com/docker/docker/api/types/time"
 	"github.com/docker/docker/daemon/cluster/convert"
 	"github.com/docker/docker/daemon/logger"
 	"github.com/docker/docker/pkg/ioutils"
@@ -272,6 +275,44 @@ func (c *Cluster) ServiceLogs(ctx context.Context, input string, config *backend
 		stdStreams = append(stdStreams, swarmapi.LogStreamStderr)
 	}
 
+	// Get tail value squared away - the number of previous log lines we look at
+	var tail int64
+	if config.Tail == "all" {
+		// tail of 0 means send all logs on the swarmkit side
+		tail = 0
+	} else {
+		t, err := strconv.Atoi(config.Tail)
+		if err != nil {
+			return errors.New("tail value must be a positive integer or \"all\"")
+		}
+		if t < 0 {
+			return errors.New("negative tail values not supported")
+		}
+		// we actually use negative tail in swarmkit to represent messages
+		// backwards starting from the beginning. also, -1 means no logs. so,
+		// basically, for api compat with docker container logs, add one and
+		// flip the sign. we error above if you try to negative tail, which
+		// isn't supported by docker (and would error deeper in the stack
+		// anyway)
+		//
+		// See the logs protobuf for more information
+		tail = int64(-(t + 1))
+	}
+
+	// get the since value - the time in the past we're looking at logs starting from
+	var sinceProto *gogotypes.Timestamp
+	if config.Since != "" {
+		s, n, err := timetypes.ParseTimestamps(config.Since, 0)
+		if err != nil {
+			return errors.Wrap(err, "could not parse since timestamp")
+		}
+		since := time.Unix(s, n)
+		sinceProto, err = gogotypes.TimestampProto(since)
+		if err != nil {
+			return errors.Wrap(err, "could not parse timestamp to proto")
+		}
+	}
+
 	stream, err := state.logsClient.SubscribeLogs(ctx, &swarmapi.SubscribeLogsRequest{
 		Selector: &swarmapi.LogSelector{
 			ServiceIDs: []string{service.ID},
@@ -279,6 +320,8 @@ func (c *Cluster) ServiceLogs(ctx context.Context, input string, config *backend
 		Options: &swarmapi.LogSubscriptionOptions{
 			Follow:  config.Follow,
 			Streams: stdStreams,
+			Tail:    tail,
+			Since:   sinceProto,
 		},
 	})
 	if err != nil {

integration-cli/docker_cli_service_logs_experimental_test.go

@@ -8,8 +8,10 @@ import (
 	"io"
 	"os/exec"
 	"strings"
+	"time"
 
 	"github.com/docker/docker/integration-cli/checker"
+	"github.com/docker/docker/integration-cli/daemon"
 	"github.com/go-check/check"
 )
 
@@ -49,6 +51,122 @@ func (s *DockerSwarmSuite) TestServiceLogs(c *check.C) {
 	}
 }
 
+// countLogLines returns a closure that can be used with waitAndAssert to
+// verify that a minimum number of expected container log messages have been
+// output.
+func countLogLines(d *daemon.Swarm, name string) func(*check.C) (interface{}, check.CommentInterface) {
+	return func(c *check.C) (interface{}, check.CommentInterface) {
+		out, err := d.Cmd("service", "logs", "-t", name)
+		c.Assert(err, checker.IsNil)
+		lines := strings.Split(strings.TrimSpace(out), "\n")
+		return len(lines), check.Commentf("output, %q", string(out))
+	}
+}
+
+func (s *DockerSwarmSuite) TestServiceLogsCompleteness(c *check.C) {
+	testRequires(c, ExperimentalDaemon)
+	d := s.AddDaemon(c, true, true)
+
+	name := "TestServiceLogsCompleteness"
+
+	// make a service that prints 6 lines
+	out, err := d.Cmd("service", "create", "--name", name, "busybox", "sh", "-c", "for line in $(seq 1 6); do echo log test $line; done; sleep 100000")
+	c.Assert(err, checker.IsNil)
+	c.Assert(strings.TrimSpace(out), checker.Not(checker.Equals), "")
+
+	// make sure task has been deployed.
+	waitAndAssert(c, defaultReconciliationTimeout, d.CheckActiveContainerCount, checker.Equals, 1)
+	// and make sure we have all the log lines
+	waitAndAssert(c, defaultReconciliationTimeout, countLogLines(d, name), checker.Equals, 6)
+
+	args := []string{"service", "logs", name}
+	cmd := exec.Command(dockerBinary, d.PrependHostArg(args)...)
+	r, w := io.Pipe()
+	cmd.Stdout = w
+	cmd.Stderr = w
+	c.Assert(cmd.Start(), checker.IsNil)
+
+	reader := bufio.NewReader(r)
+	// i have heard anecdotal reports that logs may come back from the engine
+	// mis-ordered. if this tests fails, consider the possibility that that
+	// might be occurring
+	for i := 1; i <= 6; i++ {
+		msg := &logMessage{}
+		msg.data, _, msg.err = reader.ReadLine()
+		c.Assert(msg.err, checker.IsNil)
+		c.Assert(string(msg.data), checker.Contains, fmt.Sprintf("log test %v", i))
+	}
+}
+
+func (s *DockerSwarmSuite) TestServiceLogsTail(c *check.C) {
+	testRequires(c, ExperimentalDaemon)
+	d := s.AddDaemon(c, true, true)
+
+	name := "TestServiceLogsTail"
+
+	// make a service that prints 6 lines
+	out, err := d.Cmd("service", "create", "--name", name, "busybox", "sh", "-c", "for line in $(seq 1 6); do echo log test $line; done; sleep 100000")
+	c.Assert(err, checker.IsNil)
+	c.Assert(strings.TrimSpace(out), checker.Not(checker.Equals), "")
+
+	// make sure task has been deployed.
+	waitAndAssert(c, defaultReconciliationTimeout, d.CheckActiveContainerCount, checker.Equals, 1)
+	waitAndAssert(c, defaultReconciliationTimeout, countLogLines(d, name), checker.Equals, 6)
+
+	args := []string{"service", "logs", "--tail=2", name}
+	cmd := exec.Command(dockerBinary, d.PrependHostArg(args)...)
+	r, w := io.Pipe()
+	cmd.Stdout = w
+	cmd.Stderr = w
+	c.Assert(cmd.Start(), checker.IsNil)
+
+	reader := bufio.NewReader(r)
+	// see TestServiceLogsCompleteness for comments about logs being well-
+	// ordered, if this flakes
+	for i := 5; i <= 6; i++ {
+		msg := &logMessage{}
+		msg.data, _, msg.err = reader.ReadLine()
+		c.Assert(msg.err, checker.IsNil)
+		c.Assert(string(msg.data), checker.Contains, fmt.Sprintf("log test %v", i))
+	}
+}
+
+func (s *DockerSwarmSuite) TestServiceLogsSince(c *check.C) {
+	// See DockerSuite.TestLogsSince, which is where this comes from
+	testRequires(c, ExperimentalDaemon)
+	d := s.AddDaemon(c, true, true)
+
+	name := "TestServiceLogsSince"
+
+	out, err := d.Cmd("service", "create", "--name", name, "busybox", "sh", "-c", "for i in $(seq 1 3); do sleep .1; echo log$i; done; sleep 10000000")
+	c.Assert(err, checker.IsNil)
+	c.Assert(strings.TrimSpace(out), checker.Not(checker.Equals), "")
+	waitAndAssert(c, defaultReconciliationTimeout, d.CheckActiveContainerCount, checker.Equals, 1)
+	// wait a sec for the logs to come in
+	waitAndAssert(c, defaultReconciliationTimeout, countLogLines(d, name), checker.Equals, 3)
+
+	out, err = d.Cmd("service", "logs", "-t", name)
+	c.Assert(err, checker.IsNil)
+
+	log2Line := strings.Split(strings.Split(out, "\n")[1], " ")
+	t, err := time.Parse(time.RFC3339Nano, log2Line[0]) // timestamp log2 is written
+	c.Assert(err, checker.IsNil)
+	u := t.Add(50 * time.Millisecond) // add .05s so log1 & log2 don't show up
+	since := u.Format(time.RFC3339Nano)
+
+	out, err = d.Cmd("service", "logs", "-t", fmt.Sprintf("--since=%v", since), name)
+	c.Assert(err, checker.IsNil)
+
+	unexpected := []string{"log1", "log2"}
+	expected := []string{"log3"}
+	for _, v := range unexpected {
+		c.Assert(out, checker.Not(checker.Contains), v, check.Commentf("unexpected log message returned, since=%v", u))
+	}
+	for _, v := range expected {
+		c.Assert(out, checker.Contains, v, check.Commentf("expected log message %v, was not present, since=%v", u))
+	}
+}
+
 func (s *DockerSwarmSuite) TestServiceLogsFollow(c *check.C) {
 	testRequires(c, ExperimentalDaemon)