Move server => cli

Signed-off-by: Guillaume Tardif <guillaume.tardif@gmail.com>

cli/server/proxy/compose.go

@@ -0,0 +1,102 @@
+/*
+   Copyright 2020 Docker Compose CLI authors
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package proxy
+
+import (
+	"context"
+
+	"github.com/docker/compose-cli/api/compose"
+
+	"github.com/compose-spec/compose-go/cli"
+	"github.com/compose-spec/compose-go/types"
+
+	composev1 "github.com/docker/compose-cli/protos/compose/v1"
+)
+
+func (p *proxy) Up(ctx context.Context, request *composev1.ComposeUpRequest) (*composev1.ComposeUpResponse, error) {
+	project, err := getComposeProject(request.Files, request.WorkDir, request.ProjectName)
+	if err != nil {
+		return nil, err
+	}
+	err = Client(ctx).ComposeService().Up(ctx, project, compose.UpOptions{Detach: true})
+	return &composev1.ComposeUpResponse{ProjectName: project.Name}, err
+}
+
+func (p *proxy) Down(ctx context.Context, request *composev1.ComposeDownRequest) (*composev1.ComposeDownResponse, error) {
+	projectName := request.GetProjectName()
+	if projectName == "" {
+		project, err := getComposeProject(request.Files, request.WorkDir, request.ProjectName)
+		if err != nil {
+			return nil, err
+		}
+		projectName = project.Name
+	}
+	err := Client(ctx).ComposeService().Down(ctx, projectName, compose.DownOptions{})
+	return &composev1.ComposeDownResponse{ProjectName: projectName}, err
+}
+
+func (p *proxy) Services(ctx context.Context, request *composev1.ComposeServicesRequest) (*composev1.ComposeServicesResponse, error) {
+	projectName := request.GetProjectName()
+	if projectName == "" {
+		project, err := getComposeProject(request.Files, request.WorkDir, request.ProjectName)
+		if err != nil {
+			return nil, err
+		}
+		projectName = project.Name
+	}
+	response := []*composev1.Service{}
+	_, err := Client(ctx).ComposeService().Ps(ctx, projectName)
+	if err != nil {
+		return nil, err
+	}
+	/* FIXME need to create `docker service ls` command to re-introduce this feature
+	for _, service := range services {
+		response = append(response, &composev1.Service{
+			Id:              service.ID,
+			ProjectName:     service.ProjectName,
+			Replicas:        uint32(service.Replicas),
+			Desired:         uint32(service.Desired),
+			Ports:           service.Ports,
+		})
+	}*/
+	return &composev1.ComposeServicesResponse{Services: response}, nil
+}
+
+func (p *proxy) Stacks(ctx context.Context, request *composev1.ComposeStacksRequest) (*composev1.ComposeStacksResponse, error) {
+	stacks, err := Client(ctx).ComposeService().List(ctx, request.ProjectName)
+	if err != nil {
+		return nil, err
+	}
+	response := []*composev1.Stack{}
+	for _, stack := range stacks {
+		response = append(response, &composev1.Stack{
+			Id:     stack.ID,
+			Name:   stack.Name,
+			Status: stack.Status,
+			Reason: stack.Reason,
+		})
+	}
+	return &composev1.ComposeStacksResponse{Stacks: response}, nil
+}
+
+func getComposeProject(files []string, workingDir string, projectName string) (*types.Project, error) {
+	options, err := cli.NewProjectOptions(files, cli.WithWorkingDirectory(workingDir), cli.WithName(projectName))
+	if err != nil {
+		return nil, err
+	}
+	return cli.ProjectFromOptions(options)
+}

cli/server/proxy/containers.go

@@ -0,0 +1,200 @@
+/*
+   Copyright 2020 Docker Compose CLI authors
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package proxy
+
+import (
+	"context"
+	"errors"
+
+	"github.com/compose-spec/compose-go/types"
+	"github.com/containerd/containerd/platforms"
+	specs "github.com/opencontainers/image-spec/specs-go/v1"
+
+	"github.com/docker/compose-cli/api/containers"
+	"github.com/docker/compose-cli/cli/server/proxy/streams"
+	"github.com/docker/compose-cli/formatter"
+	containersv1 "github.com/docker/compose-cli/protos/containers/v1"
+)
+
+func portsToGrpc(ports []containers.Port) []*containersv1.Port {
+	var result []*containersv1.Port
+	for _, port := range ports {
+		result = append(result, &containersv1.Port{
+			ContainerPort: port.ContainerPort,
+			HostPort:      port.HostPort,
+			HostIp:        port.HostIP,
+			Protocol:      port.Protocol,
+		})
+	}
+
+	return result
+}
+
+func (p *proxy) List(ctx context.Context, request *containersv1.ListRequest) (*containersv1.ListResponse, error) {
+	containerList, err := Client(ctx).ContainerService().List(ctx, request.GetAll())
+	if err != nil {
+		return &containersv1.ListResponse{}, err
+	}
+
+	response := &containersv1.ListResponse{
+		Containers: []*containersv1.Container{},
+	}
+	for _, container := range containerList {
+		response.Containers = append(response.Containers, toGrpcContainer(container))
+	}
+
+	return response, nil
+}
+
+func (p *proxy) Start(ctx context.Context, request *containersv1.StartRequest) (*containersv1.StartResponse, error) {
+	return &containersv1.StartResponse{}, Client(ctx).ContainerService().Start(ctx, request.Id)
+}
+
+func (p *proxy) Stop(ctx context.Context, request *containersv1.StopRequest) (*containersv1.StopResponse, error) {
+	timeoutValue := request.GetTimeout()
+	return &containersv1.StopResponse{}, Client(ctx).ContainerService().Stop(ctx, request.Id, &timeoutValue)
+}
+
+func (p *proxy) Kill(ctx context.Context, request *containersv1.KillRequest) (*containersv1.KillResponse, error) {
+	signal := request.GetSignal()
+	return &containersv1.KillResponse{}, Client(ctx).ContainerService().Kill(ctx, request.Id, signal)
+}
+
+func (p *proxy) Run(ctx context.Context, request *containersv1.RunRequest) (*containersv1.RunResponse, error) {
+	containerConfig, err := grpcContainerToContainerConfig(request)
+	if err != nil {
+		return nil, err
+	}
+	return &containersv1.RunResponse{}, Client(ctx).ContainerService().Run(ctx, containerConfig)
+}
+
+func (p *proxy) Inspect(ctx context.Context, request *containersv1.InspectRequest) (*containersv1.InspectResponse, error) {
+	c, err := Client(ctx).ContainerService().Inspect(ctx, request.Id)
+	if err != nil {
+		return nil, err
+	}
+	response := &containersv1.InspectResponse{
+		Container: toGrpcContainer(c),
+	}
+	return response, err
+}
+
+func (p *proxy) Delete(ctx context.Context, request *containersv1.DeleteRequest) (*containersv1.DeleteResponse, error) {
+	return &containersv1.DeleteResponse{}, Client(ctx).ContainerService().Delete(ctx, request.Id, containers.DeleteRequest{
+		Force: request.Force,
+	})
+}
+
+func (p *proxy) Exec(ctx context.Context, request *containersv1.ExecRequest) (*containersv1.ExecResponse, error) {
+	p.mu.Lock()
+	stream, ok := p.streams[request.StreamId]
+	p.mu.Unlock()
+	if !ok {
+		return &containersv1.ExecResponse{}, errors.New("unknown stream id")
+	}
+
+	io := &streams.IO{
+		Stream: stream,
+	}
+
+	return &containersv1.ExecResponse{}, Client(ctx).ContainerService().Exec(ctx, request.GetId(), containers.ExecRequest{
+		Stdin:   io,
+		Stdout:  io,
+		Command: request.GetCommand(),
+		Tty:     request.GetTty(),
+	})
+}
+
+func (p *proxy) Logs(request *containersv1.LogsRequest, stream containersv1.Containers_LogsServer) error {
+	return Client(stream.Context()).ContainerService().Logs(stream.Context(), request.GetContainerId(), containers.LogsRequest{
+		Follow: request.Follow,
+		Writer: &streams.Log{
+			Stream: stream,
+		},
+	})
+}
+
+func toGrpcContainer(c containers.Container) *containersv1.Container {
+	return &containersv1.Container{
+		Id:          c.ID,
+		Image:       c.Image,
+		Status:      c.Status,
+		Command:     c.Command,
+		CpuTime:     c.CPUTime,
+		MemoryUsage: c.MemoryUsage,
+		Platform:    c.Platform,
+		PidsCurrent: c.PidsCurrent,
+		PidsLimit:   c.PidsLimit,
+		Labels:      c.Config.Labels,
+		Ports:       portsToGrpc(c.Ports),
+		HostConfig: &containersv1.HostConfig{
+			MemoryReservation: c.HostConfig.MemoryReservation,
+			MemoryLimit:       c.HostConfig.MemoryLimit,
+			CpuReservation:    uint64(c.HostConfig.CPUReservation),
+			CpuLimit:          uint64(c.HostConfig.CPULimit),
+			RestartPolicy:     c.HostConfig.RestartPolicy,
+			AutoRemove:        c.HostConfig.AutoRemove,
+		},
+		Healthcheck: &containersv1.Healthcheck{
+			Disable:  c.Healthcheck.Disable,
+			Test:     c.Healthcheck.Test,
+			Interval: int64(c.Healthcheck.Interval),
+		},
+	}
+}
+
+func grpcContainerToContainerConfig(request *containersv1.RunRequest) (containers.ContainerConfig, error) {
+	var ports []containers.Port
+	for _, p := range request.GetPorts() {
+		ports = append(ports, containers.Port{
+			ContainerPort: p.ContainerPort,
+			HostIP:        p.HostIp,
+			HostPort:      p.HostPort,
+			Protocol:      p.Protocol,
+		})
+	}
+
+	var platform *specs.Platform
+
+	if request.Platform != "" {
+		p, err := platforms.Parse(request.Platform)
+		if err != nil {
+			return containers.ContainerConfig{}, err
+		}
+		platform = &p
+	}
+
+	return containers.ContainerConfig{
+		ID:                     request.GetId(),
+		Image:                  request.GetImage(),
+		Command:                request.GetCommand(),
+		Ports:                  ports,
+		Labels:                 request.GetLabels(),
+		Volumes:                request.GetVolumes(),
+		MemLimit:               formatter.MemBytes(request.GetMemoryLimit()),
+		CPULimit:               float64(request.GetCpuLimit()),
+		RestartPolicyCondition: request.RestartPolicyCondition,
+		Environment:            request.Environment,
+		AutoRemove:             request.AutoRemove,
+		Healthcheck: containers.Healthcheck{
+			Disable:  request.GetHealthcheck().GetDisable(),
+			Test:     request.GetHealthcheck().GetTest(),
+			Interval: types.Duration(request.GetHealthcheck().GetInterval()),
+		},
+		Platform: platform,
+	}, nil
+}

cli/server/proxy/containers_test.go

@@ -0,0 +1,71 @@
+/*
+   Copyright 2020 Docker Compose CLI authors
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package proxy
+
+import (
+	"testing"
+
+	"gotest.tools/v3/assert"
+
+	"github.com/docker/compose-cli/api/containers"
+	"github.com/docker/compose-cli/formatter"
+	containersv1 "github.com/docker/compose-cli/protos/containers/v1"
+)
+
+func TestGrpcContainerToContainerConfig(t *testing.T) {
+	r := &containersv1.RunRequest{
+		Id:    "myId",
+		Image: "myImage",
+		Ports: []*containersv1.Port{
+			{
+				HostPort:      8080,
+				ContainerPort: 80,
+				Protocol:      "tcp",
+				HostIp:        "42.42.42.42",
+			},
+		},
+		Labels: map[string]string{
+			"mykey": "mylabel",
+		},
+		Volumes: []string{
+			"myvolume",
+		},
+		MemoryLimit: 41,
+		CpuLimit:    42,
+		Environment: []string{"PROTOVAR=VALUE"},
+	}
+
+	cc, err := grpcContainerToContainerConfig(r)
+	assert.NilError(t, err)
+	assert.Equal(t, cc.ID, "myId")
+	assert.Equal(t, cc.Image, "myImage")
+	assert.Equal(t, cc.MemLimit, formatter.MemBytes(41))
+	assert.Equal(t, cc.CPULimit, float64(42))
+	assert.DeepEqual(t, cc.Volumes, []string{"myvolume"})
+	assert.DeepEqual(t, cc.Ports, []containers.Port{
+		{
+			HostPort:      uint32(8080),
+			ContainerPort: 80,
+			Protocol:      "tcp",
+			HostIP:        "42.42.42.42",
+		},
+	})
+	assert.DeepEqual(t, cc.Labels, map[string]string{
+		"mykey": "mylabel",
+	})
+	assert.DeepEqual(t, cc.Environment, []string{"PROTOVAR=VALUE"})
+}

cli/server/proxy/contexts.go

@@ -0,0 +1,111 @@
+/*
+   Copyright 2020 Docker Compose CLI authors
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package proxy
+
+import (
+	"context"
+
+	"github.com/docker/compose-cli/config"
+	"github.com/docker/compose-cli/context/store"
+	contextsv1 "github.com/docker/compose-cli/protos/contexts/v1"
+)
+
+type contextsProxy struct {
+	configDir string
+}
+
+func (cp *contextsProxy) SetCurrent(ctx context.Context, request *contextsv1.SetCurrentRequest) (*contextsv1.SetCurrentResponse, error) {
+	if err := config.WriteCurrentContext(cp.configDir, request.GetName()); err != nil {
+		return &contextsv1.SetCurrentResponse{}, err
+	}
+
+	return &contextsv1.SetCurrentResponse{}, nil
+}
+
+func (cp *contextsProxy) List(ctx context.Context, request *contextsv1.ListRequest) (*contextsv1.ListResponse, error) {
+	s := store.ContextStore(ctx)
+	configFile, err := config.LoadFile(cp.configDir)
+	if err != nil {
+		return nil, err
+	}
+	contexts, err := s.List()
+	if err != nil {
+		return &contextsv1.ListResponse{}, err
+	}
+
+	return convertContexts(contexts, configFile.CurrentContext), nil
+}
+
+func convertContexts(contexts []*store.DockerContext, currentContext string) *contextsv1.ListResponse {
+	result := &contextsv1.ListResponse{}
+
+	for _, c := range contexts {
+		endpointName := c.Type()
+		if c.Type() == store.DefaultContextType {
+			endpointName = "docker"
+		}
+		var endpoint interface{} = c.Endpoints[endpointName]
+
+		context := contextsv1.Context{
+			Name:        c.Name,
+			ContextType: c.Type(),
+			Description: c.Metadata.Description,
+			Current:     c.Name == currentContext,
+		}
+		switch c.Type() {
+		case store.DefaultContextType:
+			context.Endpoint = getDockerEndpoint(endpoint)
+		case store.AciContextType:
+			context.Endpoint = getAciEndpoint(endpoint)
+		case store.EcsContextType:
+			context.Endpoint = getEcsEndpoint(endpoint)
+		}
+
+		result.Contexts = append(result.Contexts, &context)
+	}
+	return result
+}
+
+func getDockerEndpoint(endpoint interface{}) *contextsv1.Context_DockerEndpoint {
+	typedEndpoint := endpoint.(*store.Endpoint)
+	return &contextsv1.Context_DockerEndpoint{
+		DockerEndpoint: &contextsv1.DockerEndpoint{
+			Host: typedEndpoint.Host,
+		},
+	}
+}
+
+func getAciEndpoint(endpoint interface{}) *contextsv1.Context_AciEndpoint {
+	typedEndpoint := endpoint.(*store.AciContext)
+	return &contextsv1.Context_AciEndpoint{
+		AciEndpoint: &contextsv1.AciEndpoint{
+			ResourceGroup:  typedEndpoint.ResourceGroup,
+			Region:         typedEndpoint.Location,
+			SubscriptionId: typedEndpoint.SubscriptionID,
+		},
+	}
+}
+
+func getEcsEndpoint(endpoint interface{}) *contextsv1.Context_EcsEndpoint {
+	typedEndpoint := endpoint.(*store.EcsContext)
+	return &contextsv1.Context_EcsEndpoint{
+		EcsEndpoint: &contextsv1.EcsEndpoint{
+			FromEnvironment: typedEndpoint.CredentialsFromEnv,
+			Profile:         typedEndpoint.Profile,
+		},
+	}
+}

cli/server/proxy/contexts_test.go

@@ -0,0 +1,111 @@
+/*
+   Copyright 2020 Docker Compose CLI authors
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package proxy
+
+import (
+	"testing"
+
+	"gotest.tools/v3/assert"
+
+	"github.com/docker/compose-cli/context/store"
+	contextsv1 "github.com/docker/compose-cli/protos/contexts/v1"
+	"github.com/google/go-cmp/cmp/cmpopts"
+)
+
+func TestConvertContext(t *testing.T) {
+	contexts := []*store.DockerContext{
+		{
+			Name: store.DefaultContextName,
+			Metadata: store.ContextMetadata{
+				Description: "description 1",
+				Type:        store.DefaultContextType,
+			},
+			Endpoints: map[string]interface{}{
+				"docker": &store.Endpoint{
+					Host: "unix://var/run/docker.sock",
+				},
+			},
+		},
+		{
+			Name: "acicontext",
+			Metadata: store.ContextMetadata{
+				Description: "group1@eastus",
+				Type:        store.AciContextType,
+			},
+			Endpoints: map[string]interface{}{
+				"aci": &store.AciContext{
+					Location:       "eastus",
+					ResourceGroup:  "group1",
+					SubscriptionID: "Subscription id",
+				},
+			},
+		},
+		{
+			Name: "ecscontext",
+			Metadata: store.ContextMetadata{
+				Description: "ecs description",
+				Type:        store.EcsContextType,
+			},
+			Endpoints: map[string]interface{}{
+				"ecs": &store.EcsContext{
+					CredentialsFromEnv: false,
+					Profile:            "awsprofile",
+				},
+			},
+		},
+	}
+	converted := convertContexts(contexts, "acicontext")
+	expected := []*contextsv1.Context{
+		{
+			Name:        store.DefaultContextName,
+			Current:     false,
+			ContextType: store.DefaultContextType,
+			Description: "description 1",
+			Endpoint: &contextsv1.Context_DockerEndpoint{
+				DockerEndpoint: &contextsv1.DockerEndpoint{
+					Host: "unix://var/run/docker.sock",
+				},
+			},
+		},
+		{
+			Name:        "acicontext",
+			Current:     true,
+			ContextType: store.AciContextType,
+			Description: "group1@eastus",
+			Endpoint: &contextsv1.Context_AciEndpoint{
+				AciEndpoint: &contextsv1.AciEndpoint{
+					Region:         "eastus",
+					ResourceGroup:  "group1",
+					SubscriptionId: "Subscription id",
+				},
+			},
+		},
+		{
+			Name:        "ecscontext",
+			Current:     false,
+			ContextType: store.EcsContextType,
+			Description: "ecs description",
+			Endpoint: &contextsv1.Context_EcsEndpoint{
+				EcsEndpoint: &contextsv1.EcsEndpoint{
+					FromEnvironment: false,
+					Profile:         "awsprofile",
+				},
+			},
+		},
+	}
+	assert.DeepEqual(t, converted.Contexts, expected, cmpopts.IgnoreUnexported(contextsv1.Context{}, contextsv1.DockerEndpoint{}, contextsv1.AciEndpoint{}, contextsv1.EcsEndpoint{}))
+}

cli/server/proxy/proxy.go

@@ -0,0 +1,77 @@
+/*
+   Copyright 2020 Docker Compose CLI authors
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package proxy
+
+import (
+	"context"
+	"sync"
+
+	"github.com/docker/compose-cli/api/client"
+	"github.com/docker/compose-cli/cli/server/proxy/streams"
+	"github.com/docker/compose-cli/config"
+	composev1 "github.com/docker/compose-cli/protos/compose/v1"
+	containersv1 "github.com/docker/compose-cli/protos/containers/v1"
+	contextsv1 "github.com/docker/compose-cli/protos/contexts/v1"
+	streamsv1 "github.com/docker/compose-cli/protos/streams/v1"
+	volumesv1 "github.com/docker/compose-cli/protos/volumes/v1"
+)
+
+type clientKey struct{}
+
+// WithClient adds the client to the context
+func WithClient(ctx context.Context, c *client.Client) context.Context {
+	return context.WithValue(ctx, clientKey{}, c)
+}
+
+// Client returns the client from the context
+func Client(ctx context.Context) *client.Client {
+	c, _ := ctx.Value(clientKey{}).(*client.Client)
+	return c
+}
+
+// Proxy implements the gRPC server and forwards the actions
+// to the right backend
+type Proxy interface {
+	composev1.ComposeServer
+	containersv1.ContainersServer
+	streamsv1.StreamingServer
+	volumesv1.VolumesServer
+	ContextsProxy() contextsv1.ContextsServer
+}
+
+type proxy struct {
+	configDir     string
+	mu            sync.Mutex
+	streams       map[string]*streams.Stream
+	contextsProxy *contextsProxy
+}
+
+// New creates a new proxy server
+func New(ctx context.Context) Proxy {
+	configDir := config.Dir(ctx)
+	return &proxy{
+		configDir: configDir,
+		streams:   map[string]*streams.Stream{},
+		contextsProxy: &contextsProxy{
+			configDir: configDir,
+		},
+	}
+}
+
+func (p *proxy) ContextsProxy() contextsv1.ContextsServer {
+	return p.contextsProxy
+}

cli/server/proxy/streams.go

@@ -0,0 +1,67 @@
+/*
+   Copyright 2020 Docker Compose CLI authors
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package proxy
+
+import (
+	"github.com/hashicorp/go-uuid"
+	"github.com/sirupsen/logrus"
+	"google.golang.org/grpc/metadata"
+
+	"github.com/docker/compose-cli/cli/server/proxy/streams"
+	streamsv1 "github.com/docker/compose-cli/protos/streams/v1"
+)
+
+func (p *proxy) NewStream(stream streamsv1.Streaming_NewStreamServer) error {
+	var (
+		ctx = stream.Context()
+	)
+	id, err := uuid.GenerateUUID()
+	if err != nil {
+		return err
+	}
+	md := metadata.New(map[string]string{
+		"id": id,
+	})
+
+	// return the id of the stream to the client
+	if err := stream.SendHeader(md); err != nil {
+		return err
+	}
+
+	errc := make(chan error)
+
+	p.mu.Lock()
+	p.streams[id] = &streams.Stream{
+		Streaming_NewStreamServer: stream,
+		ErrChan:                   errc,
+	}
+	p.mu.Unlock()
+
+	defer func() {
+		p.mu.Lock()
+		delete(p.streams, id)
+		p.mu.Unlock()
+	}()
+
+	select {
+	case err := <-errc:
+		return err
+	case <-ctx.Done():
+		logrus.Debug("client context canceled")
+		return ctx.Err()
+	}
+}

cli/server/proxy/streams/io.go

@@ -0,0 +1,61 @@
+/*
+   Copyright 2020 Docker Compose CLI authors
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package streams
+
+import (
+	"github.com/golang/protobuf/ptypes"
+
+	streamsv1 "github.com/docker/compose-cli/protos/streams/v1"
+)
+
+// IO implements an io.ReadWriter that forwards everything to the stream
+type IO struct {
+	Stream *Stream
+}
+
+func (io *IO) Read(p []byte) (int, error) {
+	a, err := io.Stream.Recv()
+	if err != nil {
+		return 0, err
+	}
+
+	var m streamsv1.BytesMessage
+	err = ptypes.UnmarshalAny(a, &m)
+	if err != nil {
+		return 0, err
+	}
+
+	return copy(p, m.Value), nil
+}
+
+func (io *IO) Write(p []byte) (n int, err error) {
+	if len(p) == 0 {
+		return 0, nil
+	}
+
+	message := streamsv1.BytesMessage{
+		Type:  streamsv1.IOStream_STDOUT,
+		Value: p,
+	}
+
+	m, err := ptypes.MarshalAny(&message)
+	if err != nil {
+		return 0, err
+	}
+
+	return len(message.Value), io.Stream.SendMsg(m)
+}

cli/server/proxy/streams/logs.go

@@ -0,0 +1,42 @@
+/*
+   Copyright 2020 Docker Compose CLI authors
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package streams
+
+import (
+	"io"
+
+	"google.golang.org/grpc"
+
+	containersv1 "github.com/docker/compose-cli/protos/containers/v1"
+)
+
+// Log implements an io.Writer that proxies logs over a gRPC stream
+type Log struct {
+	Stream grpc.ServerStream
+}
+
+func newStreamWriter(stream grpc.ServerStream) io.Writer {
+	return &Log{
+		Stream: stream,
+	}
+}
+
+func (w *Log) Write(p []byte) (n int, err error) {
+	return len(p), w.Stream.SendMsg(&containersv1.LogsResponse{
+		Value: p,
+	})
+}

cli/server/proxy/streams/logs_test.go

@@ -0,0 +1,77 @@
+/*
+   Copyright 2020 Docker Compose CLI authors
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package streams
+
+import (
+	"context"
+	"testing"
+
+	"google.golang.org/grpc/metadata"
+	"gotest.tools/v3/assert"
+	"gotest.tools/v3/assert/cmp"
+
+	v1 "github.com/docker/compose-cli/protos/containers/v1"
+)
+
+type logServer struct {
+	logs interface{}
+}
+
+func (ls *logServer) Send(response *v1.LogsResponse) error {
+	return nil
+}
+
+func (ls *logServer) SetHeader(metadata.MD) error {
+	return nil
+}
+
+func (ls *logServer) SendHeader(metadata.MD) error {
+	return nil
+}
+
+func (ls *logServer) SetTrailer(metadata.MD) {
+}
+
+func (ls *logServer) Context() context.Context {
+	return nil
+}
+
+func (ls *logServer) SendMsg(m interface{}) error {
+	ls.logs = m
+	return nil
+}
+
+func (ls *logServer) RecvMsg(m interface{}) error {
+	return nil
+}
+
+func TestLogStreamWriter(t *testing.T) {
+	ls := &logServer{}
+	sw := newStreamWriter(ls)
+	in := []byte{104, 101, 108, 108, 111}
+	expected := &v1.LogsResponse{
+		Value: in,
+	}
+
+	l, err := sw.Write(in)
+
+	assert.NilError(t, err)
+	assert.Assert(t, cmp.Len(in, l))
+	logs, ok := (ls.logs).(*v1.LogsResponse)
+	assert.Assert(t, ok)
+	assert.DeepEqual(t, logs.Value, expected.Value)
+}

cli/server/proxy/streams/stream.go

@@ -0,0 +1,47 @@
+/*
+   Copyright 2020 Docker Compose CLI authors
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package streams
+
+import (
+	"sync"
+
+	streamsv1 "github.com/docker/compose-cli/protos/streams/v1"
+)
+
+// Stream is a bidirectional stream for container IO
+type Stream struct {
+	streamsv1.Streaming_NewStreamServer
+
+	errm    sync.Mutex
+	ErrChan chan<- error
+}
+
+// CloseWithError sends the result of an action to the errChan or nil
+// if no erros
+func (s *Stream) CloseWithError(err error) error {
+	s.errm.Lock()
+	defer s.errm.Unlock()
+
+	if s.ErrChan != nil {
+		if err != nil {
+			s.ErrChan <- err
+		}
+		close(s.ErrChan)
+		s.ErrChan = nil
+	}
+	return nil
+}

cli/server/proxy/streams/stream_test.go

@@ -0,0 +1,141 @@
+/*
+   Copyright 2020 Docker Compose CLI authors
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package streams
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"google.golang.org/grpc/metadata"
+	"google.golang.org/protobuf/types/known/anypb"
+
+	"github.com/golang/protobuf/ptypes"
+	"github.com/golang/protobuf/ptypes/any"
+	"gotest.tools/v3/assert"
+	"gotest.tools/v3/assert/cmp"
+
+	streamsv1 "github.com/docker/compose-cli/protos/streams/v1"
+)
+
+type byteStream struct {
+	recvResult *any.Any
+	recvErr    error
+
+	sendResult interface{}
+}
+
+func (bs *byteStream) SetHeader(metadata.MD) error {
+	return nil
+}
+
+func (bs *byteStream) SendHeader(metadata.MD) error {
+	return nil
+}
+
+func (bs *byteStream) SetTrailer(metadata.MD) {
+}
+
+func (bs *byteStream) Context() context.Context {
+	return nil
+}
+
+func (bs *byteStream) SendMsg(m interface{}) error {
+	bs.sendResult = m
+	return nil
+}
+
+func (bs *byteStream) Send(*any.Any) error {
+	return nil
+}
+
+func (bs *byteStream) Recv() (*any.Any, error) {
+	return bs.recvResult, bs.recvErr
+}
+
+func (bs *byteStream) RecvMsg(m interface{}) error {
+	return nil
+}
+
+func getReader(t *testing.T, in []byte, errResult error) IO {
+	message := streamsv1.BytesMessage{
+		Type:  streamsv1.IOStream_STDOUT,
+		Value: in,
+	}
+	m, err := ptypes.MarshalAny(&message)
+	assert.NilError(t, err)
+
+	return IO{
+		Stream: &Stream{
+			Streaming_NewStreamServer: &byteStream{
+				recvResult: m,
+				recvErr:    errResult,
+			},
+		},
+	}
+}
+
+func getAny(t *testing.T, in []byte) *any.Any {
+	value, err := ptypes.MarshalAny(&streamsv1.BytesMessage{
+		Type:  streamsv1.IOStream_STDOUT,
+		Value: in,
+	})
+	assert.NilError(t, err)
+	return value
+}
+
+func TestStreamReader(t *testing.T) {
+	in := []byte{104, 101, 108, 108, 111}
+	r := getReader(t, in, nil)
+	buffer := make([]byte, 5)
+
+	n, err := r.Read(buffer)
+
+	assert.NilError(t, err)
+	assert.Equal(t, n, 5)
+	assert.DeepEqual(t, buffer, in)
+}
+
+func TestStreamReaderError(t *testing.T) {
+	errResult := errors.New("err")
+	r := getReader(t, nil, errResult)
+	var buffer []byte
+
+	n, err := r.Read(buffer)
+
+	assert.Equal(t, n, 0)
+	assert.Error(t, err, errResult.Error())
+}
+
+func TestStreamWriter(t *testing.T) {
+	in := []byte{104, 101, 108, 108, 111}
+	expected := getAny(t, in)
+
+	bs := byteStream{}
+	w := IO{
+		Stream: &Stream{
+			Streaming_NewStreamServer: &bs,
+		},
+	}
+
+	n, err := w.Write(in)
+	assert.NilError(t, err)
+	assert.Assert(t, cmp.Len(in, n))
+	sendResult, ok := (bs.sendResult).(*anypb.Any)
+	assert.Assert(t, ok)
+	assert.DeepEqual(t, sendResult.Value, expected.Value)
+}

cli/server/proxy/volumes.go

@@ -0,0 +1,87 @@
+/*
+   Copyright 2020 Docker Compose CLI authors
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package proxy
+
+import (
+	"context"
+
+	"github.com/docker/compose-cli/aci"
+	"github.com/docker/compose-cli/api/volumes"
+	volumesv1 "github.com/docker/compose-cli/protos/volumes/v1"
+)
+
+// VolumesCreate creates a volume.
+func (p *proxy) VolumesCreate(ctx context.Context, req *volumesv1.VolumesCreateRequest) (*volumesv1.VolumesCreateResponse, error) {
+	storageAccount := ""
+	aciReq := req.GetAciOption()
+	if aciReq != nil {
+		storageAccount = aciReq.StorageAccount
+	}
+	aciOpts := aci.VolumeCreateOptions{
+		Account: storageAccount,
+	}
+
+	v, err := Client(ctx).VolumeService().Create(ctx, req.Name, aciOpts)
+	if err != nil {
+		return &volumesv1.VolumesCreateResponse{}, err
+	}
+
+	return &volumesv1.VolumesCreateResponse{
+		Volume: toGrpcVolume(v),
+	}, nil
+}
+
+// VolumesList lists the volumes.
+func (p *proxy) VolumesList(ctx context.Context, req *volumesv1.VolumesListRequest) (*volumesv1.VolumesListResponse, error) {
+	volumeList, err := Client(ctx).VolumeService().List(ctx)
+	if err != nil {
+		return &volumesv1.VolumesListResponse{}, err
+	}
+
+	return &volumesv1.VolumesListResponse{
+		Volumes: toGrpcVolumeList(volumeList),
+	}, nil
+}
+
+// VolumesDelete deletes a volume.
+func (p *proxy) VolumesDelete(ctx context.Context, req *volumesv1.VolumesDeleteRequest) (*volumesv1.VolumesDeleteResponse, error) {
+	err := Client(ctx).VolumeService().Delete(ctx, req.Id, nil)
+	return &volumesv1.VolumesDeleteResponse{}, err
+}
+
+// VolumesInspect inspects a volume.
+func (p *proxy) VolumesInspect(ctx context.Context, req *volumesv1.VolumesInspectRequest) (*volumesv1.VolumesInspectResponse, error) {
+	v, err := Client(ctx).VolumeService().Inspect(ctx, req.Id)
+	return &volumesv1.VolumesInspectResponse{
+		Volume: toGrpcVolume(v),
+	}, err
+}
+
+func toGrpcVolumeList(volumeList []volumes.Volume) []*volumesv1.Volume {
+	var ret []*volumesv1.Volume
+	for _, v := range volumeList {
+		ret = append(ret, toGrpcVolume(v))
+	}
+	return ret
+}
+
+func toGrpcVolume(v volumes.Volume) *volumesv1.Volume {
+	return &volumesv1.Volume{
+		Id:          v.ID,
+		Description: v.Description,
+	}
+}