docklog: add go-plugin for forwarding of docker logs

drivers/docker/docklog/client.go

@@ -0,0 +1,35 @@
+package docklog
+
+import (
+	"context"
+
+	"github.com/hashicorp/nomad/drivers/docker/docklog/proto"
+)
+
+// docklogClient implements the Docklog interface for client side requests
+type docklogClient struct {
+	client proto.DocklogClient
+}
+
+// Start proxies the Start client side func to the protobuf interface
+func (c *docklogClient) Start(opts *StartOpts) error {
+	req := &proto.StartRequest{
+		Endpoint:    opts.Endpoint,
+		ContainerId: opts.ContainerID,
+		StdoutFifo:  opts.Stdout,
+		StderrFifo:  opts.Stderr,
+
+		TlsCert: opts.TLSCert,
+		TlsKey:  opts.TLSKey,
+		TlsCa:   opts.TLSCA,
+	}
+	_, err := c.client.Start(context.Background(), req)
+	return err
+}
+
+// Stop proxies the Stop client side func to the protobuf interface
+func (c *docklogClient) Stop() error {
+	req := &proto.StopRequest{}
+	_, err := c.client.Stop(context.Background(), req)
+	return err
+}

drivers/docker/docklog/docklog.go

@@ -0,0 +1,141 @@
+package docklog
+
+import (
+	"fmt"
+	"io"
+
+	docker "github.com/fsouza/go-dockerclient"
+	hclog "github.com/hashicorp/go-hclog"
+	multierror "github.com/hashicorp/go-multierror"
+	"github.com/hashicorp/nomad/client/lib/fifo"
+	"golang.org/x/net/context"
+)
+
+// Docklog is a small utility to forward logs from a docker container to a target
+// destination
+type Docklog interface {
+	Start(*StartOpts) error
+	Stop() error
+}
+
+// StartOpts are the options needed to start docker log monitoring
+type StartOpts struct {
+	// Endpoint sets the docker client endpoint, defaults to environment if not set
+	Endpoint string
+
+	// ContainerID of the container to monitor logs for
+	ContainerID string
+
+	// Stdout path to fifo
+	Stdout string
+	//Stderr path to fifo
+	Stderr string
+
+	// TLS settings for docker client
+	TLSCert string
+	TLSKey  string
+	TLSCA   string
+}
+
+// NewDocklog returns an implementation of the Docklog interface
+func NewDocklog(logger hclog.Logger) Docklog {
+	return &dockerLogger{logger: logger}
+}
+
+// dockerLogger implements the Docklog interface
+type dockerLogger struct {
+	logger hclog.Logger
+
+	stdout    io.WriteCloser
+	stderr    io.WriteCloser
+	cancelCtx context.CancelFunc
+}
+
+// Start log monitoring
+func (d *dockerLogger) Start(opts *StartOpts) error {
+	client, err := d.getDockerClient(opts)
+	if err != nil {
+		return fmt.Errorf("failed to open docker client: %v", err)
+	}
+
+	if d.stdout == nil {
+		stdout, err := fifo.Open(opts.Stdout)
+		if err != nil {
+			return fmt.Errorf("failed to open fifo for path %s: %v", opts.Stdout, err)
+		}
+		d.stdout = stdout
+	}
+	if d.stderr == nil {
+		stderr, err := fifo.Open(opts.Stderr)
+		if err != nil {
+			return fmt.Errorf("failed to open fifo for path %s: %v", opts.Stdout, err)
+		}
+		d.stderr = stderr
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	d.cancelCtx = cancel
+
+	logOpts := docker.LogsOptions{
+		Context:      ctx,
+		Container:    opts.ContainerID,
+		OutputStream: d.stdout,
+		ErrorStream:  d.stderr,
+		Since:        0,
+		Follow:       true,
+		Stdout:       true,
+		Stderr:       true,
+	}
+
+	go func() { client.Logs(logOpts) }()
+	return nil
+
+}
+
+// Stop log monitoring
+func (d *dockerLogger) Stop() error {
+	if d.cancelCtx != nil {
+		d.cancelCtx()
+	}
+	if d.stdout != nil {
+		d.stdout.Close()
+	}
+	if d.stderr != nil {
+		d.stderr.Close()
+	}
+	return nil
+}
+
+func (d *dockerLogger) getDockerClient(opts *StartOpts) (*docker.Client, error) {
+	var err error
+	var merr multierror.Error
+	var newClient *docker.Client
+
+	// Default to using whatever is configured in docker.endpoint. If this is
+	// not specified we'll fall back on NewClientFromEnv which reads config from
+	// the DOCKER_* environment variables DOCKER_HOST, DOCKER_TLS_VERIFY, and
+	// DOCKER_CERT_PATH. This allows us to lock down the config in production
+	// but also accept the standard ENV configs for dev and test.
+	if opts.Endpoint != "" {
+		if opts.TLSCert+opts.TLSKey+opts.TLSCA != "" {
+			d.logger.Debug("using TLS client connection to docker", "endpoint", opts.Endpoint)
+			newClient, err = docker.NewTLSClient(opts.Endpoint, opts.TLSCert, opts.TLSKey, opts.TLSCA)
+			if err != nil {
+				merr.Errors = append(merr.Errors, err)
+			}
+		} else {
+			d.logger.Debug("using plaintext client connection to docker", "endpoint", opts.Endpoint)
+			newClient, err = docker.NewClient(opts.Endpoint)
+			if err != nil {
+				merr.Errors = append(merr.Errors, err)
+			}
+		}
+	} else {
+		d.logger.Debug("using client connection initialized from environment")
+		newClient, err = docker.NewClientFromEnv()
+		if err != nil {
+			merr.Errors = append(merr.Errors, err)
+		}
+	}
+
+	return newClient, merr.ErrorOrNil()
+}

drivers/docker/docklog/docklog_test.go

@@ -0,0 +1,102 @@
+package docklog
+
+import (
+	"bytes"
+	"fmt"
+	"testing"
+	"time"
+
+	docker "github.com/fsouza/go-dockerclient"
+	"github.com/hashicorp/nomad/helper/testlog"
+	"github.com/hashicorp/nomad/testutil"
+	"github.com/stretchr/testify/require"
+	"golang.org/x/net/context"
+)
+
+func TestDocklog(t *testing.T) {
+	t.Parallel()
+	require := require.New(t)
+
+	client, err := docker.NewClientFromEnv()
+	if err != nil {
+		t.Skip("docker unavailable:", err)
+	}
+
+	containerConf := docker.CreateContainerOptions{
+		Config: &docker.Config{
+			Cmd: []string{
+				"/bin/ash", "-c", "touch /tmp/docklog; tail -f /tmp/docklog",
+			},
+			Image: "alpine",
+		},
+		Context: context.Background(),
+	}
+
+	container, err := client.CreateContainer(containerConf)
+	require.NoError(err)
+
+	defer client.RemoveContainer(docker.RemoveContainerOptions{
+		ID:    container.ID,
+		Force: true,
+	})
+
+	err = client.StartContainer(container.ID, nil)
+	require.NoError(err)
+
+	var count int
+	for !container.State.Running {
+		if count > 10 {
+			t.Fatal("timeout waiting for container to start")
+		}
+		time.Sleep(100 * time.Millisecond)
+		container, err = client.InspectContainer(container.ID)
+		count++
+	}
+
+	stdout := &noopCloser{bytes.NewBufferString("")}
+	stderr := &noopCloser{bytes.NewBufferString("")}
+
+	dl := NewDocklog(testlog.HCLogger(t)).(*dockerLogger)
+	dl.stdout = stdout
+	dl.stderr = stderr
+	require.NoError(dl.Start(&StartOpts{
+		ContainerID: container.ID,
+	}))
+
+	echoToContainer(t, client, container.ID, "abc")
+	echoToContainer(t, client, container.ID, "123")
+
+	time.Sleep(2000 * time.Millisecond)
+	testutil.WaitForResult(func() (bool, error) {
+		act := stdout.String()
+		if "abc\n123\n" != act {
+			return false, fmt.Errorf("expected abc\\n123\\n for stdout but got %s", act)
+		}
+
+		return true, nil
+	}, func(err error) {
+		require.NoError(err)
+	})
+}
+
+func echoToContainer(t *testing.T, client *docker.Client, id string, line string) {
+	op := docker.CreateExecOptions{
+		Container: id,
+		Cmd: []string{
+			"/bin/ash", "-c",
+			fmt.Sprintf("echo %s >>/tmp/docklog", line),
+		},
+	}
+
+	exec, err := client.CreateExec(op)
+	require.NoError(t, err)
+	require.NoError(t, client.StartExec(exec.ID, docker.StartExecOptions{Detach: true}))
+}
+
+type noopCloser struct {
+	*bytes.Buffer
+}
+
+func (*noopCloser) Close() error {
+	return nil
+}

drivers/docker/docklog/plugin.go

@@ -0,0 +1,68 @@
+package docklog
+
+import (
+	"context"
+	"os/exec"
+
+	hclog "github.com/hashicorp/go-hclog"
+	plugin "github.com/hashicorp/go-plugin"
+	"github.com/hashicorp/nomad/drivers/docker/docklog/proto"
+	"github.com/hashicorp/nomad/helper/discover"
+	"github.com/hashicorp/nomad/plugins/base"
+	"google.golang.org/grpc"
+)
+
+// LaunchDocklog launches an instance of docklog
+// TODO: Integrate with base plugin loader
+func LaunchDocklog(logger hclog.Logger) (Docklog, *plugin.Client, error) {
+	logger = logger.Named("docklog-launcher")
+	bin, err := discover.NomadExecutable()
+	if err != nil {
+		return nil, nil, err
+	}
+
+	client := plugin.NewClient(&plugin.ClientConfig{
+		HandshakeConfig: base.Handshake,
+		Plugins: map[string]plugin.Plugin{
+			"docklog": &Plugin{impl: NewDocklog(hclog.L().Named("docklog"))},
+		},
+		Cmd: exec.Command(bin, "docklog"),
+		AllowedProtocols: []plugin.Protocol{
+			plugin.ProtocolGRPC,
+		},
+	})
+
+	rpcClient, err := client.Client()
+	if err != nil {
+		return nil, nil, err
+	}
+
+	raw, err := rpcClient.Dispense("docklog")
+	if err != nil {
+		return nil, nil, err
+	}
+
+	l := raw.(Docklog)
+	return l, client, nil
+
+}
+
+// Plugin is the go-plugin implementation
+type Plugin struct {
+	plugin.NetRPCUnsupportedPlugin
+	impl Docklog
+}
+
+// GRPCServer registered the server side implementation with the grpc server
+func (p *Plugin) GRPCServer(broker *plugin.GRPCBroker, s *grpc.Server) error {
+	proto.RegisterDocklogServer(s, &docklogServer{
+		impl:   p.impl,
+		broker: broker,
+	})
+	return nil
+}
+
+// GRPCClient returns a client side implementation of the plugin
+func (p *Plugin) GRPCClient(ctx context.Context, broker *plugin.GRPCBroker, c *grpc.ClientConn) (interface{}, error) {
+	return &docklogClient{client: proto.NewDocklogClient(c)}, nil
+}