@@ -100,7 +100,7 @@ github.com/docker/containerd 52ef1ceb4b660c42cf4ea9013180a5663968d4c7

 github.com/tonistiigi/fifo 8c56881ce5e63e19e2dfc495c8af0fb90916467d

 # cluster

-github.com/docker/swarmkit 4dfc88ccce14ced6f0a6ea82d46dca004c6de0e2

+github.com/docker/swarmkit 2eaae1ab6800f8521236e01bafb2667d2ec5371f

 github.com/golang/mock bd3c8e81be01eef76d4b503f5e687d2d1354d2d9

 github.com/gogo/protobuf v0.3

 github.com/cloudflare/cfssl 7fb22c8cba7ecaf98e4082d22d65800cf45e042a

@@ -7,6 +7,7 @@ import (

 	"sync"

 	"time"

+	"github.com/docker/swarmkit/agent/exec"

 	"github.com/docker/swarmkit/api"

 	"github.com/docker/swarmkit/log"

 	"golang.org/x/net/context"

@@ -48,11 +49,8 @@ func New(config *Config) (*Agent, error) {

 		return nil, err

 	}

-	worker := newWorker(config.DB, config.Executor)

-

 	a := &Agent{

 		config:   config,

-		worker:   worker,

 		sessionq: make(chan sessionOperation),

 		started:  make(chan struct{}),

 		stopped:  make(chan struct{}),

@@ -60,6 +58,7 @@ func New(config *Config) (*Agent, error) {

 		ready:    make(chan struct{}),

 	}

+	a.worker = newWorker(config.DB, config.Executor, a)

 	return a, nil

 }

@@ -147,11 +146,12 @@ func (a *Agent) run(ctx context.Context) {

 	defer nodeUpdateTicker.Stop()

 	var (

-		backoff    time.Duration

-		session    = newSession(ctx, a, backoff, "", nodeDescription) // start the initial session

-		registered = session.registered

-		ready      = a.ready // first session ready

-		sessionq   chan sessionOperation

+		backoff       time.Duration

+		session       = newSession(ctx, a, backoff, "", nodeDescription) // start the initial session

+		registered    = session.registered

+		ready         = a.ready // first session ready

+		sessionq      chan sessionOperation

+		subscriptions = map[string]context.CancelFunc{}

 	)

 	if err := a.worker.Init(ctx); err != nil {

@@ -159,6 +159,7 @@ func (a *Agent) run(ctx context.Context) {

 		a.err = err

 		return // fatal?

 	}

+	defer a.worker.Close()

 	// setup a reliable reporter to call back to us.

 	reporter := newStatusReporter(ctx, a)

@@ -186,6 +187,23 @@ func (a *Agent) run(ctx context.Context) {

 			if err := a.handleSessionMessage(ctx, msg); err != nil {

 				log.G(ctx).WithError(err).Error("session message handler failed")

 			}

+		case sub := <-session.subscriptions:

+			if sub.Close {

+				if cancel, ok := subscriptions[sub.ID]; ok {

+					cancel()

+				}

+				delete(subscriptions, sub.ID)

+				continue

+			}

+

+			if _, ok := subscriptions[sub.ID]; ok {

+				// Duplicate subscription

+				continue

+			}

+

+			subCtx, subCancel := context.WithCancel(ctx)

+			subscriptions[sub.ID] = subCancel

+			go a.worker.Subscribe(subCtx, sub)

 		case <-registered:

 			log.G(ctx).Debugln("agent: registered")

 			if ready != nil {

@@ -387,6 +405,40 @@ func (a *Agent) UpdateTaskStatus(ctx context.Context, taskID string, status *api

 	}

 }

+// Publisher returns a LogPublisher for the given subscription

+func (a *Agent) Publisher(ctx context.Context, subscriptionID string) (exec.LogPublisher, error) {

+	// TODO(stevvooe): The level of coordination here is WAY too much for logs.

+	// These should only be best effort and really just buffer until a session is

+	// ready. Ideally, they would use a separate connection completely.

+

+	var (

+		err    error

+		client api.LogBroker_PublishLogsClient

+	)

+

+	err = a.withSession(ctx, func(session *session) error {

+		client, err = api.NewLogBrokerClient(session.conn).PublishLogs(ctx)

+		return err

+	})

+	if err != nil {

+		return nil, err

+	}

+

+	return exec.LogPublisherFunc(func(ctx context.Context, message api.LogMessage) error {

+		select {

+		case <-ctx.Done():

+			client.CloseSend()

+			return ctx.Err()

+		default:

+		}

+

+		return client.Send(&api.PublishLogsMessage{

+			SubscriptionID: subscriptionID,

+			Messages:       []api.LogMessage{message},

+		})

+	}), nil

+}

+

 // nodeDescriptionWithHostname retrieves node description, and overrides hostname if available

 func (a *Agent) nodeDescriptionWithHostname(ctx context.Context) (*api.NodeDescription, error) {

 	desc, err := a.config.Executor.Describe(ctx)

@@ -45,6 +45,33 @@ type Controller interface {

 	Close() error

 }

+// ControllerLogs defines a component that makes logs accessible.

+//

+// Can usually be accessed on a controller instance via type assertion.

+type ControllerLogs interface {

+	// Logs will write publisher until the context is cancelled or an error

+	// occurs.

+	Logs(ctx context.Context, publisher LogPublisher, options api.LogSubscriptionOptions) error

+}

+

+// LogPublisher defines the protocol for receiving a log message.

+type LogPublisher interface {

+	Publish(ctx context.Context, message api.LogMessage) error

+}

+

+// LogPublisherFunc implements publisher with just a function.

+type LogPublisherFunc func(ctx context.Context, message api.LogMessage) error

+

+// Publish calls the wrapped function.

+func (fn LogPublisherFunc) Publish(ctx context.Context, message api.LogMessage) error {

+	return fn(ctx, message)

+}

+

+// LogPublisherProvider defines the protocol for receiving a log publisher

+type LogPublisherProvider interface {

+	Publisher(ctx context.Context, subscriptionID string) (LogPublisher, error)

+}

+

 // ContainerStatuser reports status of a container.

 //

 // This can be implemented by controllers or error types.

@@ -110,6 +110,100 @@ func (_mr *_MockControllerRecorder) Close() *gomock.Call {

 	return _mr.mock.ctrl.RecordCall(_mr.mock, "Close")

 }

+// Mock of ControllerLogs interface

+type MockControllerLogs struct {

+	ctrl     *gomock.Controller

+	recorder *_MockControllerLogsRecorder

+}

+

+// Recorder for MockControllerLogs (not exported)

+type _MockControllerLogsRecorder struct {

+	mock *MockControllerLogs

+}

+

+func NewMockControllerLogs(ctrl *gomock.Controller) *MockControllerLogs {

+	mock := &MockControllerLogs{ctrl: ctrl}

+	mock.recorder = &_MockControllerLogsRecorder{mock}

+	return mock

+}

+

+func (_m *MockControllerLogs) EXPECT() *_MockControllerLogsRecorder {

+	return _m.recorder

+}

+

+func (_m *MockControllerLogs) Logs(ctx context.Context, publisher LogPublisher, options api.LogSubscriptionOptions) error {

+	ret := _m.ctrl.Call(_m, "Logs", ctx, publisher, options)

+	ret0, _ := ret[0].(error)

+	return ret0

+}

+

+func (_mr *_MockControllerLogsRecorder) Logs(arg0, arg1, arg2 interface{}) *gomock.Call {

+	return _mr.mock.ctrl.RecordCall(_mr.mock, "Logs", arg0, arg1, arg2)

+}

+

+// Mock of LogPublisher interface

+type MockLogPublisher struct {

+	ctrl     *gomock.Controller

+	recorder *_MockLogPublisherRecorder

+}

+

+// Recorder for MockLogPublisher (not exported)

+type _MockLogPublisherRecorder struct {

+	mock *MockLogPublisher

+}

+

+func NewMockLogPublisher(ctrl *gomock.Controller) *MockLogPublisher {

+	mock := &MockLogPublisher{ctrl: ctrl}

+	mock.recorder = &_MockLogPublisherRecorder{mock}

+	return mock

+}

+

+func (_m *MockLogPublisher) EXPECT() *_MockLogPublisherRecorder {

+	return _m.recorder

+}

+

+func (_m *MockLogPublisher) Publish(ctx context.Context, message api.LogMessage) error {

+	ret := _m.ctrl.Call(_m, "Publish", ctx, message)

+	ret0, _ := ret[0].(error)

+	return ret0

+}

+

+func (_mr *_MockLogPublisherRecorder) Publish(arg0, arg1 interface{}) *gomock.Call {

+	return _mr.mock.ctrl.RecordCall(_mr.mock, "Publish", arg0, arg1)

+}

+

+// Mock of LogPublisherProvider interface

+type MockLogPublisherProvider struct {

+	ctrl     *gomock.Controller

+	recorder *_MockLogPublisherProviderRecorder

+}

+

+// Recorder for MockLogPublisherProvider (not exported)

+type _MockLogPublisherProviderRecorder struct {

+	mock *MockLogPublisherProvider

+}

+

+func NewMockLogPublisherProvider(ctrl *gomock.Controller) *MockLogPublisherProvider {

+	mock := &MockLogPublisherProvider{ctrl: ctrl}

+	mock.recorder = &_MockLogPublisherProviderRecorder{mock}

+	return mock

+}

+

+func (_m *MockLogPublisherProvider) EXPECT() *_MockLogPublisherProviderRecorder {

+	return _m.recorder

+}

+

+func (_m *MockLogPublisherProvider) Publisher(ctx context.Context, subscriptionID string) (LogPublisher, error) {

+	ret := _m.ctrl.Call(_m, "Publisher", ctx, subscriptionID)

+	ret0, _ := ret[0].(LogPublisher)

+	ret1, _ := ret[1].(error)

+	return ret0, ret1

+}

+

+func (_mr *_MockLogPublisherProviderRecorder) Publisher(arg0, arg1 interface{}) *gomock.Call {

+	return _mr.mock.ctrl.RecordCall(_mr.mock, "Publisher", arg0, arg1)

+}

+

 // Mock of ContainerStatuser interface

 type MockContainerStatuser struct {

 	ctrl     *gomock.Controller

@@ -33,12 +33,13 @@ type session struct {

 	conn *grpc.ClientConn

 	addr string

-	agent       *Agent

-	sessionID   string

-	session     api.Dispatcher_SessionClient

-	errs        chan error

-	messages    chan *api.SessionMessage

-	assignments chan *api.AssignmentsMessage

+	agent         *Agent

+	sessionID     string

+	session       api.Dispatcher_SessionClient

+	errs          chan error

+	messages      chan *api.SessionMessage

+	assignments   chan *api.AssignmentsMessage

+	subscriptions chan *api.SubscriptionMessage

 	registered chan struct{} // closed registration

 	closed     chan struct{}

@@ -47,14 +48,19 @@ type session struct {

 func newSession(ctx context.Context, agent *Agent, delay time.Duration, sessionID string, description *api.NodeDescription) *session {

 	s := &session{

-		agent:       agent,

-		sessionID:   sessionID,

-		errs:        make(chan error, 1),

-		messages:    make(chan *api.SessionMessage),

-		assignments: make(chan *api.AssignmentsMessage),

-		registered:  make(chan struct{}),

-		closed:      make(chan struct{}),

+		agent:         agent,

+		sessionID:     sessionID,

+		errs:          make(chan error, 1),

+		messages:      make(chan *api.SessionMessage),

+		assignments:   make(chan *api.AssignmentsMessage),

+		subscriptions: make(chan *api.SubscriptionMessage),

+		registered:    make(chan struct{}),

+		closed:        make(chan struct{}),

 	}

+

+	// TODO(stevvooe): Need to move connection management up a level or create

+	// independent connection for log broker client.

+

 	peer, err := agent.config.Managers.Select()

 	if err != nil {

 		s.errs <- err

@@ -98,6 +104,7 @@ func (s *session) run(ctx context.Context, delay time.Duration, description *api

 	go runctx(ctx, s.closed, s.errs, s.heartbeat)

 	go runctx(ctx, s.closed, s.errs, s.watch)

 	go runctx(ctx, s.closed, s.errs, s.listen)

+	go runctx(ctx, s.closed, s.errs, s.logSubscriptions)

 	close(s.registered)

 }

@@ -213,6 +220,33 @@ func (s *session) handleSessionMessage(ctx context.Context, msg *api.SessionMess

 	}

 }

+func (s *session) logSubscriptions(ctx context.Context) error {

+	log := log.G(ctx).WithFields(logrus.Fields{"method": "(*session).logSubscriptions"})

+	log.Debugf("")

+

+	client := api.NewLogBrokerClient(s.conn)

+	subscriptions, err := client.ListenSubscriptions(ctx, &api.ListenSubscriptionsRequest{})

+	if err != nil {

+		return err

+	}

+	defer subscriptions.CloseSend()

+

+	for {

+		resp, err := subscriptions.Recv()

+		if err != nil {

+			return err

+		}

+

+		select {

+		case s.subscriptions <- resp:

+		case <-s.closed:

+			return errSessionClosed

+		case <-ctx.Done():

+			return ctx.Err()

+		}

+	}

+}

+

 func (s *session) watch(ctx context.Context) error {

 	log := log.G(ctx).WithFields(logrus.Fields{"method": "(*session).watch"})

 	log.Debugf("")

@@ -64,6 +64,18 @@ func (tm *taskManager) Close() error {

 	}

 }

+func (tm *taskManager) Logs(ctx context.Context, options api.LogSubscriptionOptions, publisher exec.LogPublisher) {

+	ctx = log.WithModule(ctx, "taskmanager")

+

+	logCtlr, ok := tm.ctlr.(exec.ControllerLogs)

+	if !ok {

+		return // no logs available

+	}

+	if err := logCtlr.Logs(ctx, publisher, options); err != nil {

+		log.G(ctx).WithError(err).Errorf("logs call failed")

+	}

+}

+

 func (tm *taskManager) run(ctx context.Context) {

 	ctx, cancelAll := context.WithCancel(ctx)

 	defer cancelAll() // cancel all child operations on exit.

@@ -8,6 +8,7 @@ import (

 	"github.com/docker/swarmkit/agent/exec"

 	"github.com/docker/swarmkit/api"

 	"github.com/docker/swarmkit/log"

+	"github.com/docker/swarmkit/watch"

 	"golang.org/x/net/context"

 )

@@ -17,6 +18,11 @@ type Worker interface {

 	// Init prepares the worker for task assignment.

 	Init(ctx context.Context) error

+	// Close performs worker cleanup when no longer needed.

+	//

+	// It is not safe to call any worker function after that.

+	Close()

+

 	// Assign assigns a complete set of tasks and secrets to a worker. Any task or secrets not included in

 	// this set will be removed.

 	Assign(ctx context.Context, assignments []*api.AssignmentChange) error

@@ -31,6 +37,9 @@ type Worker interface {

 	//

 	// The listener will be removed if the context is cancelled.

 	Listen(ctx context.Context, reporter StatusReporter)

+

+	// Subscribe to log messages matching the subscription.

+	Subscribe(ctx context.Context, subscription *api.SubscriptionMessage) error

 }

 // statusReporterKey protects removal map from panic.

@@ -39,20 +48,25 @@ type statusReporterKey struct {

 }

 type worker struct {

-	db        *bolt.DB

-	executor  exec.Executor

-	listeners map[*statusReporterKey]struct{}

+	db                *bolt.DB

+	executor          exec.Executor

+	publisher         exec.LogPublisher

+	listeners         map[*statusReporterKey]struct{}

+	taskevents        *watch.Queue

+	publisherProvider exec.LogPublisherProvider

 	taskManagers map[string]*taskManager

 	mu           sync.RWMutex

 }

-func newWorker(db *bolt.DB, executor exec.Executor) *worker {

+func newWorker(db *bolt.DB, executor exec.Executor, publisherProvider exec.LogPublisherProvider) *worker {

 	return &worker{

-		db:           db,

-		executor:     executor,

-		listeners:    make(map[*statusReporterKey]struct{}),

-		taskManagers: make(map[string]*taskManager),

+		db:                db,

+		executor:          executor,

+		publisherProvider: publisherProvider,

+		taskevents:        watch.NewQueue(),

+		listeners:         make(map[*statusReporterKey]struct{}),

+		taskManagers:      make(map[string]*taskManager),

 	}

 }

@@ -90,6 +104,11 @@ func (w *worker) Init(ctx context.Context) error {

 	})

 }

+// Close performs worker cleanup when no longer needed.

+func (w *worker) Close() {

+	w.taskevents.Close()

+}

+

 // Assign assigns a full set of tasks and secrets to the worker.

 // Any tasks not previously known will be started. Any tasks that are in the task set

 // and already running will be updated, if possible. Any tasks currently running on

@@ -319,6 +338,7 @@ func (w *worker) Listen(ctx context.Context, reporter StatusReporter) {

 }

 func (w *worker) startTask(ctx context.Context, tx *bolt.Tx, task *api.Task) error {

+	w.taskevents.Publish(task.Copy())

 	_, err := w.taskManager(ctx, tx, task) // side-effect taskManager creation.

 	if err != nil {

@@ -381,3 +401,63 @@ func (w *worker) updateTaskStatus(ctx context.Context, tx *bolt.Tx, taskID strin

 	return nil

 }

+

+// Subscribe to log messages matching the subscription.

+func (w *worker) Subscribe(ctx context.Context, subscription *api.SubscriptionMessage) error {

+	log.G(ctx).Debugf("Received subscription %s (selector: %v)", subscription.ID, subscription.Selector)

+

+	publisher, err := w.publisherProvider.Publisher(ctx, subscription.ID)

+	if err != nil {

+		return err

+	}

+	// Send a close once we're done

+	defer publisher.Publish(ctx, api.LogMessage{})

+

+	match := func(t *api.Task) bool {

+		// TODO(aluzzardi): Consider using maps to limit the iterations.

+		for _, tid := range subscription.Selector.TaskIDs {

+			if t.ID == tid {

+				return true

+			}

+		}

+

+		for _, sid := range subscription.Selector.ServiceIDs {

+			if t.ServiceID == sid {

+				return true

+			}

+		}

+

+		for _, nid := range subscription.Selector.NodeIDs {

+			if t.NodeID == nid {

+				return true

+			}

+		}

+

+		return false

+	}

+

+	ch, cancel := w.taskevents.Watch()

+	defer cancel()

+

+	w.mu.Lock()

+	for _, tm := range w.taskManagers {

+		if match(tm.task) {

+			go tm.Logs(ctx, *subscription.Options, publisher)

+		}

+	}

+	w.mu.Unlock()

+

+	for {

+		select {

+		case v := <-ch:

+			w.mu.Lock()

+			task := v.(*api.Task)

+			if match(task) {

+				go w.taskManagers[task.ID].Logs(ctx, *subscription.Options, publisher)

+			}

+			w.mu.Unlock()

+		case <-ctx.Done():

+			return ctx.Err()

+		}

+	}

+}

@@ -1,3 +1,3 @@

 package api

-//go:generate protoc -I.:../protobuf:../vendor:../vendor/github.com/gogo/protobuf --gogoswarm_out=plugins=grpc+deepcopy+raftproxy+authenticatedwrapper,import_path=github.com/docker/swarmkit/api,Mgogoproto/gogo.proto=github.com/gogo/protobuf/gogoproto,Mtimestamp/timestamp.proto=github.com/docker/swarmkit/api/timestamp,Mduration/duration.proto=github.com/docker/swarmkit/api/duration,Mgoogle/protobuf/descriptor.proto=github.com/gogo/protobuf/protoc-gen-gogo/descriptor,Mplugin/plugin.proto=github.com/docker/swarmkit/protobuf/plugin:. types.proto specs.proto objects.proto control.proto dispatcher.proto ca.proto snapshot.proto raft.proto health.proto resource.proto

+//go:generate protoc -I.:../protobuf:../vendor:../vendor/github.com/gogo/protobuf --gogoswarm_out=plugins=grpc+deepcopy+raftproxy+authenticatedwrapper,import_path=github.com/docker/swarmkit/api,Mgogoproto/gogo.proto=github.com/gogo/protobuf/gogoproto,Mtimestamp/timestamp.proto=github.com/docker/swarmkit/api/timestamp,Mduration/duration.proto=github.com/docker/swarmkit/api/duration,Mgoogle/protobuf/descriptor.proto=github.com/gogo/protobuf/protoc-gen-gogo/descriptor,Mplugin/plugin.proto=github.com/docker/swarmkit/protobuf/plugin:. types.proto specs.proto objects.proto control.proto dispatcher.proto ca.proto snapshot.proto raft.proto health.proto resource.proto logbroker.proto

new file mode 100644

@@ -0,0 +1,3130 @@

+// Code generated by protoc-gen-gogo.

+// source: logbroker.proto

+// DO NOT EDIT!

+

+package api

+

+import proto "github.com/gogo/protobuf/proto"

+import fmt "fmt"

+import math "math"

+import _ "github.com/gogo/protobuf/gogoproto"

+import docker_swarmkit_v1 "github.com/docker/swarmkit/api/timestamp"

+import _ "github.com/docker/swarmkit/protobuf/plugin"

+

+import strings "strings"

+import github_com_gogo_protobuf_proto "github.com/gogo/protobuf/proto"

+import sort "sort"

+import strconv "strconv"

+import reflect "reflect"

+

+import (

+	context "golang.org/x/net/context"

+	grpc "google.golang.org/grpc"

+)

+

+import raftselector "github.com/docker/swarmkit/manager/raftselector"

+import codes "google.golang.org/grpc/codes"

+import metadata "google.golang.org/grpc/metadata"

+import transport "google.golang.org/grpc/transport"

+import time "time"

+

+import io "io"

+

+// Reference imports to suppress errors if they are not otherwise used.

+var _ = proto.Marshal

+var _ = fmt.Errorf

+var _ = math.Inf

+

+// LogStream defines the stream from which the log message came.

+type LogStream int32

+

+const (

+	LogStreamUnknown LogStream = 0

+	LogStreamStdout  LogStream = 1

+	LogStreamStderr  LogStream = 2

+)

+

+var LogStream_name = map[int32]string{

+	0: "LOG_STREAM_UNKNOWN",

+	1: "LOG_STREAM_STDOUT",

+	2: "LOG_STREAM_STDERR",

+}

+var LogStream_value = map[string]int32{

+	"LOG_STREAM_UNKNOWN": 0,

+	"LOG_STREAM_STDOUT":  1,

+	"LOG_STREAM_STDERR":  2,

+}

+

+func (x LogStream) String() string {

+	return proto.EnumName(LogStream_name, int32(x))

+}

+func (LogStream) EnumDescriptor() ([]byte, []int) { return fileDescriptorLogbroker, []int{0} }

+

+type LogSubscriptionOptions struct {

+	// Streams defines which log streams should be sent from the task source.

+	// Empty means send all the messages.

+	Streams []LogStream `protobuf:"varint,1,rep,name=streams,enum=docker.swarmkit.v1.LogStream" json:"streams,omitempty"`

+	// Follow instructs the publisher to continue sending log messages as they

+	// are produced, after satisfying the initial query.

+	Follow bool `protobuf:"varint,2,opt,name=follow,proto3" json:"follow,omitempty"`

+	// Tail defines how many messages relative to the log stream to send when

+	// starting the stream.

+	//

+	// Positive values will skip that number of messages from the start of the

+	// stream before publishing.

+	//

+	// Negative values will specify messages relative to the end of the stream,

+	// offset by one. We can say that the last (-n-1) lines are returned when n

+	// < 0. As reference, -1 would mean send no log lines (typically used with

+	// follow), -2 would return the last log line, -11 would return the last 10

+	// and so on.

+	//

+	// The default value of zero will return all logs.

+	//

+	// Note that this is very different from the Docker API.

+	Tail int64 `protobuf:"varint,3,opt,name=tail,proto3" json:"tail,omitempty"`

+	// Since indicates that only log messages produced after this timestamp

+	// should be sent.

+	Since *docker_swarmkit_v1.Timestamp `protobuf:"bytes,4,opt,name=since" json:"since,omitempty"`

+}

+

+func (m *LogSubscriptionOptions) Reset()                    { *m = LogSubscriptionOptions{} }

+func (*LogSubscriptionOptions) ProtoMessage()               {}

+func (*LogSubscriptionOptions) Descriptor() ([]byte, []int) { return fileDescriptorLogbroker, []int{0} }

+

+// LogSelector will match logs from ANY of the defined parameters.

+//

+// For the best effect, the client should use the least specific parameter

+// possible. For example, if they want to listen to all the tasks of a service,

+// they should use the service id, rather than specifying the individual tasks.

+type LogSelector struct {

+	ServiceIDs []string `protobuf:"bytes,1,rep,name=service_ids,json=serviceIds" json:"service_ids,omitempty"`

+	NodeIDs    []string `protobuf:"bytes,2,rep,name=node_ids,json=nodeIds" json:"node_ids,omitempty"`

+	TaskIDs    []string `protobuf:"bytes,3,rep,name=task_ids,json=taskIds" json:"task_ids,omitempty"`

+}

+

+func (m *LogSelector) Reset()                    { *m = LogSelector{} }

+func (*LogSelector) ProtoMessage()               {}

+func (*LogSelector) Descriptor() ([]byte, []int) { return fileDescriptorLogbroker, []int{1} }

+

+// LogContext marks the context from which a log message was generated.

+type LogContext struct {

+	ServiceID string `protobuf:"bytes,1,opt,name=service_id,json=serviceId,proto3" json:"service_id,omitempty"`

+	NodeID    string `protobuf:"bytes,2,opt,name=node_id,json=nodeId,proto3" json:"node_id,omitempty"`

+	TaskID    string `protobuf:"bytes,3,opt,name=task_id,json=taskId,proto3" json:"task_id,omitempty"`

+}

+

+func (m *LogContext) Reset()                    { *m = LogContext{} }

+func (*LogContext) ProtoMessage()               {}

+func (*LogContext) Descriptor() ([]byte, []int) { return fileDescriptorLogbroker, []int{2} }

+

+// LogMessage

+type LogMessage struct {

+	// Context identifies the source of the log message.

+	Context LogContext `protobuf:"bytes,1,opt,name=context" json:"context"`

+	// Timestamp is the time at which the message was generated.

+	Timestamp *docker_swarmkit_v1.Timestamp `protobuf:"bytes,2,opt,name=timestamp" json:"timestamp,omitempty"`

+	// Stream identifies the stream of the log message, stdout or stderr.

+	Stream LogStream `protobuf:"varint,3,opt,name=stream,proto3,enum=docker.swarmkit.v1.LogStream" json:"stream,omitempty"`

+	// Data is the raw log message, as generated by the application.

+	Data []byte `protobuf:"bytes,4,opt,name=data,proto3" json:"data,omitempty"`

+}

+

+func (m *LogMessage) Reset()                    { *m = LogMessage{} }

+func (*LogMessage) ProtoMessage()               {}

+func (*LogMessage) Descriptor() ([]byte, []int) { return fileDescriptorLogbroker, []int{3} }

+

+type SubscribeLogsRequest struct {

+	// LogSelector describes the logs to which the subscriber is

+	Selector *LogSelector            `protobuf:"bytes,1,opt,name=selector" json:"selector,omitempty"`

+	Options  *LogSubscriptionOptions `protobuf:"bytes,2,opt,name=options" json:"options,omitempty"`

+}

+

+func (m *SubscribeLogsRequest) Reset()                    { *m = SubscribeLogsRequest{} }

+func (*SubscribeLogsRequest) ProtoMessage()               {}

+func (*SubscribeLogsRequest) Descriptor() ([]byte, []int) { return fileDescriptorLogbroker, []int{4} }

+

+type SubscribeLogsMessage struct {

+	Messages []LogMessage `protobuf:"bytes,1,rep,name=messages" json:"messages"`

+}

+

+func (m *SubscribeLogsMessage) Reset()                    { *m = SubscribeLogsMessage{} }

+func (*SubscribeLogsMessage) ProtoMessage()               {}

+func (*SubscribeLogsMessage) Descriptor() ([]byte, []int) { return fileDescriptorLogbroker, []int{5} }

+

+// ListenSubscriptionsRequest is a placeholder to begin listening for

+// subscriptions.

+type ListenSubscriptionsRequest struct {

+}

+

+func (m *ListenSubscriptionsRequest) Reset()      { *m = ListenSubscriptionsRequest{} }

+func (*ListenSubscriptionsRequest) ProtoMessage() {}

+func (*ListenSubscriptionsRequest) Descriptor() ([]byte, []int) {

+	return fileDescriptorLogbroker, []int{6}

+}

+

+// SubscriptionMessage instructs the listener to start publishing messages for

+// the stream or end a subscription.

+//

+// If Options.Follow == false, the worker should end the subscription on its own.

+type SubscriptionMessage struct {

+	// ID identifies the subscription.

+	ID string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"`

+	// Selector defines which sources should be sent for the subscription.

+	Selector *LogSelector `protobuf:"bytes,2,opt,name=selector" json:"selector,omitempty"`

+	// Options specify how the subscription should be satisfied.

+	Options *LogSubscriptionOptions `protobuf:"bytes,3,opt,name=options" json:"options,omitempty"`

+	// Close will be true if the node should shutdown the subscription with the

+	// provided identifier.

+	Close bool `protobuf:"varint,4,opt,name=close,proto3" json:"close,omitempty"`

+}

+

+func (m *SubscriptionMessage) Reset()                    { *m = SubscriptionMessage{} }

+func (*SubscriptionMessage) ProtoMessage()               {}

+func (*SubscriptionMessage) Descriptor() ([]byte, []int) { return fileDescriptorLogbroker, []int{7} }

+

+type PublishLogsMessage struct {

+	// SubscriptionID identifies which subscription the set of messages should

+	// be sent to. We can think of this as a "mail box" for the subscription.

+	SubscriptionID string `protobuf:"bytes,1,opt,name=subscription_id,json=subscriptionId,proto3" json:"subscription_id,omitempty"`

+	// Messages is the log message for publishing.

+	Messages []LogMessage `protobuf:"bytes,2,rep,name=messages" json:"messages"`

+}

+

+func (m *PublishLogsMessage) Reset()                    { *m = PublishLogsMessage{} }

+func (*PublishLogsMessage) ProtoMessage()               {}

+func (*PublishLogsMessage) Descriptor() ([]byte, []int) { return fileDescriptorLogbroker, []int{8} }

+

+type PublishLogsResponse struct {

+}

+

+func (m *PublishLogsResponse) Reset()                    { *m = PublishLogsResponse{} }

+func (*PublishLogsResponse) ProtoMessage()               {}

+func (*PublishLogsResponse) Descriptor() ([]byte, []int) { return fileDescriptorLogbroker, []int{9} }

+

+func init() {

+	proto.RegisterType((*LogSubscriptionOptions)(nil), "docker.swarmkit.v1.LogSubscriptionOptions")

+	proto.RegisterType((*LogSelector)(nil), "docker.swarmkit.v1.LogSelector")

+	proto.RegisterType((*LogContext)(nil), "docker.swarmkit.v1.LogContext")

+	proto.RegisterType((*LogMessage)(nil), "docker.swarmkit.v1.LogMessage")

+	proto.RegisterType((*SubscribeLogsRequest)(nil), "docker.swarmkit.v1.SubscribeLogsRequest")

+	proto.RegisterType((*SubscribeLogsMessage)(nil), "docker.swarmkit.v1.SubscribeLogsMessage")

+	proto.RegisterType((*ListenSubscriptionsRequest)(nil), "docker.swarmkit.v1.ListenSubscriptionsRequest")

+	proto.RegisterType((*SubscriptionMessage)(nil), "docker.swarmkit.v1.SubscriptionMessage")

+	proto.RegisterType((*PublishLogsMessage)(nil), "docker.swarmkit.v1.PublishLogsMessage")

+	proto.RegisterType((*PublishLogsResponse)(nil), "docker.swarmkit.v1.PublishLogsResponse")

+	proto.RegisterEnum("docker.swarmkit.v1.LogStream", LogStream_name, LogStream_value)

+}

+

+type authenticatedWrapperLogsServer struct {

+	local     LogsServer

+	authorize func(context.Context, []string) error

+}

+

+func NewAuthenticatedWrapperLogsServer(local LogsServer, authorize func(context.Context, []string) error) LogsServer {

+	return &authenticatedWrapperLogsServer{

+		local:     local,

+		authorize: authorize,

+	}

+}

+

+func (p *authenticatedWrapperLogsServer) SubscribeLogs(r *SubscribeLogsRequest, stream Logs_SubscribeLogsServer) error {

+

+	if err := p.authorize(stream.Context(), []string{"swarm-manager"}); err != nil {

+		return err

+	}

+	return p.local.SubscribeLogs(r, stream)

+}

+

+type authenticatedWrapperLogBrokerServer struct {

+	local     LogBrokerServer

+	authorize func(context.Context, []string) error

+}

+

+func NewAuthenticatedWrapperLogBrokerServer(local LogBrokerServer, authorize func(context.Context, []string) error) LogBrokerServer {

+	return &authenticatedWrapperLogBrokerServer{

+		local:     local,

+		authorize: authorize,

+	}

+}

+

+func (p *authenticatedWrapperLogBrokerServer) ListenSubscriptions(r *ListenSubscriptionsRequest, stream LogBroker_ListenSubscriptionsServer) error {

+

+	if err := p.authorize(stream.Context(), []string{"swarm-worker", "swarm-manager"}); err != nil {

+		return err

+	}

+	return p.local.ListenSubscriptions(r, stream)

+}

+

+func (p *authenticatedWrapperLogBrokerServer) PublishLogs(stream LogBroker_PublishLogsServer) error {

+

+	if err := p.authorize(stream.Context(), []string{"swarm-worker", "swarm-manager"}); err != nil {

+		return err

+	}

+	return p.local.PublishLogs(stream)

+}

+

+func (m *LogSubscriptionOptions) Copy() *LogSubscriptionOptions {

+	if m == nil {

+		return nil

+	}

+

+	o := &LogSubscriptionOptions{

+		Follow: m.Follow,

+		Tail:   m.Tail,

+		Since:  m.Since.Copy(),

+	}

+

+	if m.Streams != nil {

+		o.Streams = make([]LogStream, 0, len(m.Streams))

+		o.Streams = append(o.Streams, m.Streams...)

+	}

+

+	return o

+}

+

+func (m *LogSelector) Copy() *LogSelector {

+	if m == nil {

+		return nil

+	}

+

+	o := &LogSelector{}

+

+	if m.ServiceIDs != nil {

+		o.ServiceIDs = make([]string, 0, len(m.ServiceIDs))

+		o.ServiceIDs = append(o.ServiceIDs, m.ServiceIDs...)

+	}

+

+	if m.NodeIDs != nil {

+		o.NodeIDs = make([]string, 0, len(m.NodeIDs))

+		o.NodeIDs = append(o.NodeIDs, m.NodeIDs...)

+	}

+

+	if m.TaskIDs != nil {

+		o.TaskIDs = make([]string, 0, len(m.TaskIDs))

+		o.TaskIDs = append(o.TaskIDs, m.TaskIDs...)

+	}

+

+	return o

+}

+

+func (m *LogContext) Copy() *LogContext {

+	if m == nil {

+		return nil

+	}

+

+	o := &LogContext{

+		ServiceID: m.ServiceID,

+		NodeID:    m.NodeID,

+		TaskID:    m.TaskID,

+	}

+

+	return o

+}

+

+func (m *LogMessage) Copy() *LogMessage {

+	if m == nil {

+		return nil

+	}

+

+	o := &LogMessage{

+		Context:   *m.Context.Copy(),

+		Timestamp: m.Timestamp.Copy(),

+		Stream:    m.Stream,

+		Data:      m.Data,

+	}

+

+	return o

+}

+

+func (m *SubscribeLogsRequest) Copy() *SubscribeLogsRequest {

+	if m == nil {

+		return nil

+	}