PoC implementation

I want to address moby#1895 - Please beaware that this is just an experiment
to get my hands dirty with the codebase and to understand how things
works.

At the current stage this basically is a gRPC service on the managers
that connects the client to the client to the container executing the
task of a specific service.

It does not have TTYs.

Feedback and What I need to go on:
- I need some advice in designing this and if possible code review.
- Since this is just a PoC I'm not yet writing tests, this is just
something I need to understand a codebase where I'm not even that
familiar to do TDD.
- There are probably, I already see some of them contexts that are not
closed.

Try it:
  swarmctl service exec <service-id>

(Yes I know that one expects to have `sh`) in the end or something like
that. It's too early

Signed-off-by: Lorenzo Fontana <lo@linux.com>

agent/session.go

@@ -1,10 +1,15 @@
 package agent
 
 import (
+	"bytes"
 	"errors"
+	"io"
+	"strings"
 	"sync"
 	"time"
 
+	"github.com/docker/docker/api/types"
+	engineapi "github.com/docker/docker/client"
 	"github.com/docker/swarmkit/api"
 	"github.com/docker/swarmkit/connectionbroker"
 	"github.com/docker/swarmkit/log"
@@ -99,6 +104,7 @@ func (s *session) run(ctx context.Context, delay time.Duration, description *api
 	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)
+	go runctx(ctx, s.closed, s.errs, s.taskExecutions)
 
 	close(s.registered)
 }
@@ -217,6 +223,58 @@ func (s *session) handleSessionMessage(ctx context.Context, msg *api.SessionMess
 	}
 }
 
+func (s *session) taskExecutions(ctx context.Context) error {
+	// TODO(fntlnz): REMOVE THE CLIENT HERE find a way to have here the docker client instantiated in cmd/swarmd
+	// instead of creating a new one or at least use the same creation logic bcause the other spports containerd
+	dclient, err := engineapi.NewEnvClient()
+	client := api.NewTaskExecClient(s.conn.ClientConn)
+	attachment, err := client.Attach(ctx)
+	if err != nil {
+		return err
+	}
+
+	// TODO(fntlnz): rework this to support streaming over the same exec and not just sending/receiveing
+	// TODO(fntlnz): pass out errors instead of just conintuing
+
+	for {
+		data, err := attachment.Recv()
+
+		if err != nil {
+			return err
+		}
+		execCfg := types.ExecConfig{
+			User:         "root",
+			AttachStdout: true,
+			AttachStderr: true,
+			AttachStdin:  true,
+			Cmd:          strings.Split(string(data.Message), " "),
+		}
+
+		res, err := dclient.ContainerExecCreate(ctx, data.Containerid, execCfg)
+
+		if err != nil {
+			return err
+		}
+
+		resp, err := dclient.ContainerExecAttach(ctx, res.ID, execCfg)
+		if err != nil {
+			return err
+		}
+		defer resp.Close()
+
+		buf := new(bytes.Buffer)
+		io.Copy(buf, resp.Reader)
+
+		// sending the result back (in future we will have a tty)
+		attachment.Send(&api.TaskExecStream{
+			Message:     buf.Bytes(),
+			Containerid: data.Containerid,
+		})
+
+	}
+
+}
+
 func (s *session) logSubscriptions(ctx context.Context) error {
 	log := log.G(ctx).WithFields(logrus.Fields{"method": "(*session).logSubscriptions"})
 	log.Debugf("")

api/api.pb.txt

@@ -4630,6 +4630,14 @@ file {
       type_name: ".docker.swarmkit.v1.HealthConfig"
       json_name: "healthcheck"
     }
+    field {
+      name: "init_value"
+      number: 23
+      label: LABEL_OPTIONAL
+      type: TYPE_BOOL
+      oneof_index: 0
+      json_name: "initValue"
+    }
     nested_type {
       name: "LabelsEntry"
       field {
@@ -4684,6 +4692,9 @@ file {
         json_name: "options"
       }
     }
+    oneof_decl {
+      name: "init"
+    }
   }
   message_type {
     name: "EndpointSpec"
@@ -6155,6 +6166,23 @@ file {
   dependency: "github.com/docker/swarmkit/api/types.proto"
   dependency: "gogoproto/gogo.proto"
   dependency: "github.com/docker/swarmkit/protobuf/plugin/plugin.proto"
+  message_type {
+    name: "TaskExecStream"
+    field {
+      name: "message"
+      number: 1
+      label: LABEL_OPTIONAL
+      type: TYPE_BYTES
+      json_name: "message"
+    }
+    field {
+      name: "containerid"
+      number: 2
+      label: LABEL_OPTIONAL
+      type: TYPE_STRING
+      json_name: "containerid"
+    }
+  }
   message_type {
     name: "GetNodeRequest"
     field {
@@ -7321,6 +7349,21 @@ file {
   message_type {
     name: "RemoveConfigResponse"
   }
+  service {
+    name: "TaskExec"
+    method {
+      name: "Attach"
+      input_type: ".docker.swarmkit.v1.TaskExecStream"
+      output_type: ".docker.swarmkit.v1.TaskExecStream"
+      options {
+        73626345 {
+          1: "swarm-manager"
+        }
+      }
+      client_streaming: true
+      server_streaming: true
+    }
+  }
   service {
     name: "Control"
     method {
@@ -7383,6 +7426,18 @@ file {
         }
       }
     }
+    method {
+      name: "Attach"
+      input_type: ".docker.swarmkit.v1.TaskExecStream"
+      output_type: ".docker.swarmkit.v1.TaskExecStream"
+      options {
+        73626345 {
+          1: "swarm-manager"
+        }
+      }
+      client_streaming: true
+      server_streaming: true
+    }
     method {
       name: "RemoveTask"
       input_type: ".docker.swarmkit.v1.RemoveTaskRequest"