introduce docker compose events

Signed-off-by: Nicolas De Loof <nicolas.deloof@gmail.com>

aci/compose.go

@@ -229,3 +229,7 @@ func (cs *aciComposeService) Exec(ctx context.Context, project *types.Project, o
 func (cs *aciComposeService) Top(ctx context.Context, projectName string, services []string) ([]compose.ContainerProcSummary, error) {
 	return nil, errdefs.ErrNotImplemented
 }
+
+func (cs *aciComposeService) Events(ctx context.Context, project string, options compose.EventsOptions) error {
+	return errdefs.ErrNotImplemented
+}

api/client/compose.go

@@ -103,3 +103,7 @@ func (c *composeService) UnPause(ctx context.Context, project *types.Project) er
 func (c *composeService) Top(ctx context.Context, projectName string, services []string) ([]compose.ContainerProcSummary, error) {
 	return nil, errdefs.ErrNotImplemented
 }
+
+func (c *composeService) Events(ctx context.Context, project string, options compose.EventsOptions) error {
+	return errdefs.ErrNotImplemented
+}

api/compose/api.go

@@ -18,6 +18,7 @@ package compose
 
 import (
 	"context"
+	"fmt"
 	"io"
 	"strings"
 	"time"
@@ -65,6 +66,8 @@ type Service interface {
 	UnPause(ctx context.Context, project *types.Project) error
 	// Top executes the equivalent to a `compose top`
 	Top(ctx context.Context, projectName string, services []string) ([]ContainerProcSummary, error)
+	// Events executes the equivalent to a `compose events`
+	Events(ctx context.Context, project string, options EventsOptions) error
 }
 
 // BuildOptions group options of the Build API
@@ -156,7 +159,7 @@ type RemoveOptions struct {
 	Force bool
 }
 
-// RunOptions options to execute compose run
+// RunOptions group options of the Run API
 type RunOptions struct {
 	Name              string
 	Service           string
@@ -177,6 +180,31 @@ type RunOptions struct {
 	Index int
 }
 
+// EventsOptions group options of the Events API
+type EventsOptions struct {
+	Services []string
+	Consumer func(event Event) error
+}
+
+// Event is a container runtime event served by Events API
+type Event struct {
+	Timestamp  time.Time
+	Service    string
+	Container  string
+	Status     string
+	Attributes map[string]string
+}
+
+func (e Event) String() string {
+	t := e.Timestamp.Format("2006-01-02 15:04:05.000000")
+	var attr []string
+	for k, v := range e.Attributes {
+		attr = append(attr, fmt.Sprintf("%s=%s", k, v))
+	}
+	return fmt.Sprintf("%s container %s %s (%s)\n", t, e.Status, e.Container, strings.Join(attr, ", "))
+
+}
+
 // EnvironmentMap return RunOptions.Environment as a MappingWithEquals
 func (opts *RunOptions) EnvironmentMap() types.MappingWithEquals {
 	environment := types.MappingWithEquals{}

cli/cmd/compose/compose.go

@@ -136,6 +136,7 @@ func Command(contextType string) *cobra.Command {
 		pauseCommand(&opts),
 		unpauseCommand(&opts),
 		topCommand(&opts),
+		eventsCommand(&opts),
 	)
 
 	if contextType == store.LocalContextType || contextType == store.DefaultContextType {

cli/cmd/compose/events.go

@@ -0,0 +1,86 @@
+/*
+   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 compose
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+
+	"github.com/docker/compose-cli/api/client"
+	"github.com/docker/compose-cli/api/compose"
+
+	"github.com/spf13/cobra"
+)
+
+type eventsOpts struct {
+	*composeOptions
+	json bool
+}
+
+func eventsCommand(p *projectOptions) *cobra.Command {
+	opts := eventsOpts{
+		composeOptions: &composeOptions{
+			projectOptions: p,
+		},
+	}
+	cmd := &cobra.Command{
+		Use:   "events [options] [--] [SERVICE...]",
+		Short: "Receive real time events from containers.",
+		RunE: func(cmd *cobra.Command, args []string) error {
+			return runEvents(cmd.Context(), opts, args)
+		},
+	}
+
+	cmd.Flags().BoolVar(&opts.json, "json", false, "Output events as a stream of json objects")
+	return cmd
+}
+
+func runEvents(ctx context.Context, opts eventsOpts, services []string) error {
+	c, err := client.NewWithDefaultLocalBackend(ctx)
+	if err != nil {
+		return err
+	}
+
+	project, err := opts.toProjectName()
+	if err != nil {
+		return err
+	}
+
+	return c.ComposeService().Events(ctx, project, compose.EventsOptions{
+		Services: services,
+		Consumer: func(event compose.Event) error {
+			if opts.json {
+				marshal, err := json.Marshal(map[string]interface{}{
+					"time":       event.Timestamp,
+					"type":       "container",
+					"service":    event.Service,
+					"id":         event.Container,
+					"action":     event.Status,
+					"attributes": event.Attributes,
+				})
+				if err != nil {
+					return err
+				}
+				fmt.Println(string(marshal))
+			} else {
+				fmt.Println(event)
+			}
+			return nil
+		},
+	})
+}

ecs/local/compose.go

@@ -195,3 +195,7 @@ func (e ecsLocalSimulation) UnPause(ctx context.Context, project *types.Project)
 func (e ecsLocalSimulation) Top(ctx context.Context, projectName string, services []string) ([]compose.ContainerProcSummary, error) {
 	return e.compose.Top(ctx, projectName, services)
 }
+	
+func (e ecsLocalSimulation) Events(ctx context.Context, project string, options compose.EventsOptions) error {
+	return e.compose.Events(ctx, project, options)
+}

ecs/up.go

@@ -63,6 +63,10 @@ func (b *ecsAPIService) UnPause(ctx context.Context, project *types.Project) err
 	return errdefs.ErrNotImplemented
 }
 
+func (b *ecsAPIService) Events(ctx context.Context, project string, options compose.EventsOptions) error {
+	return errdefs.ErrNotImplemented
+}
+
 func (b *ecsAPIService) Up(ctx context.Context, project *types.Project, options compose.UpOptions) error {
 	logrus.Debugf("deploying on AWS with region=%q", b.Region)
 	err := b.aws.CheckRequirements(ctx, b.Region)

kube/compose.go

@@ -258,3 +258,7 @@ func (s *composeService) UnPause(ctx context.Context, project *types.Project) er
 func (s *composeService) Top(ctx context.Context, projectName string, services []string) ([]compose.ContainerProcSummary, error) {
 	return nil, errdefs.ErrNotImplemented
 }
+	
+func (s *composeService) Events(ctx context.Context, project string, options compose.EventsOptions) error {
+	return errdefs.ErrNotImplemented
+}

local/compose/events.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 compose
+
+import (
+	"context"
+	"strings"
+	"time"
+
+	"github.com/docker/compose-cli/api/compose"
+	"github.com/docker/compose-cli/utils"
+
+	moby "github.com/docker/docker/api/types"
+	"github.com/docker/docker/api/types/filters"
+)
+
+func (s *composeService) Events(ctx context.Context, project string, options compose.EventsOptions) error {
+	events, errors := s.apiClient.Events(ctx, moby.EventsOptions{
+		Filters: filters.NewArgs(projectFilter(project)),
+	})
+	for {
+		select {
+		case event := <-events:
+			// TODO: support other event types
+			if event.Type != "container" {
+				continue
+			}
+
+			service := event.Actor.Attributes[serviceLabel]
+			if len(options.Services) > 0 && !utils.StringContains(options.Services, service) {
+				continue
+			}
+
+			attributes := map[string]string{}
+			for k, v := range event.Actor.Attributes {
+				if strings.HasPrefix(k, "com.docker.compose.") {
+					continue
+				}
+				attributes[k] = v
+			}
+
+			err := options.Consumer(compose.Event{
+				Timestamp:  time.Unix(event.Time, event.TimeNano),
+				Service:    service,
+				Container:  event.ID,
+				Status:     event.Status,
+				Attributes: attributes,
+			})
+			if err != nil {
+				return err
+			}
+
+		case err := <-errors:
+			return err
+		}
+	}
+}