Add events support in client

Signed-off-by: Michael Crosby <crosbymichael@gmail.com>

api/grpc/server/server.go

@@ -217,12 +217,14 @@ func (s *apiServer) UpdateContainer(ctx context.Context, r *types.UpdateContaine
 
 func (s *apiServer) Events(r *types.EventsRequest, stream types.API_EventsServer) error {
 	events := s.sv.Events()
+	defer s.sv.Unsubscribe(events)
 	for evt := range events {
 		switch evt.Type {
 		case containerd.ExitEventType:
 			ev := &types.Event{
 				Type:   "exit",
 				Id:     evt.ID,
+				Pid:    uint32(evt.Pid),
 				Status: uint32(evt.Status),
 			}
 			if err := stream.Send(ev); err != nil {

ctr/events.go

@@ -0,0 +1,34 @@
+package main
+
+import (
+	"fmt"
+	"os"
+	"text/tabwriter"
+
+	"github.com/codegangsta/cli"
+	"github.com/docker/containerd/api/grpc/types"
+	netcontext "golang.org/x/net/context"
+)
+
+var EventsCommand = cli.Command{
+	Name:  "events",
+	Usage: "receive events from the containerd daemon",
+	Action: func(context *cli.Context) {
+		c := getClient()
+		events, err := c.Events(netcontext.Background(), &types.EventsRequest{})
+		if err != nil {
+			fatal(err.Error(), 1)
+		}
+		w := tabwriter.NewWriter(os.Stdout, 20, 1, 3, ' ', 0)
+		fmt.Fprint(w, "TYPE\tID\tPID\tSTATUS\n")
+		w.Flush()
+		for {
+			e, err := events.Recv()
+			if err != nil {
+				fatal(err.Error(), 1)
+			}
+			fmt.Fprintf(w, "%s\t%s\t%d\t%d\n", e.Type, e.Id, e.Pid, e.Status)
+			w.Flush()
+		}
+	},
+}

ctr/main.go

@@ -36,6 +36,7 @@ func main() {
 	app.Commands = []cli.Command{
 		ContainersCommand,
 		CheckpointCommand,
+		EventsCommand,
 	}
 	app.Before = func(context *cli.Context) error {
 		if context.GlobalBool("debug") {

delete.go

@@ -14,6 +14,12 @@ func (h *DeleteEvent) Handle(e *Event) error {
 		if err := h.deleteContainer(container); err != nil {
 			logrus.WithField("error", err).Error("containerd: deleting container")
 		}
+		h.s.NotifySubscribers(&Event{
+			Type:   ExitEventType,
+			ID:     e.ID,
+			Status: e.Status,
+			Pid:    e.Pid,
+		})
 		ContainersCounter.Dec(1)
 		h.s.containerGroup.Done()
 	}

exit.go

@@ -6,10 +6,6 @@ type ExitEvent struct {
 	s *Supervisor
 }
 
-type ExecExitEvent struct {
-	s *Supervisor
-}
-
 func (h *ExitEvent) Handle(e *Event) error {
 	logrus.WithFields(logrus.Fields{"pid": e.Pid, "status": e.Status}).
 		Debug("containerd: process exited")
@@ -34,10 +30,16 @@ func (h *ExitEvent) Handle(e *Event) error {
 	container.SetExited(e.Status)
 	ne := NewEvent(DeleteEventType)
 	ne.ID = container.ID()
+	ne.Pid = e.Pid
+	ne.Status = e.Status
 	h.s.SendEvent(ne)
 	return nil
 }
 
+type ExecExitEvent struct {
+	s *Supervisor
+}
+
 func (h *ExecExitEvent) Handle(e *Event) error {
 	// exec process: we remove this process without notifying the main event loop
 	container := h.s.processes[e.Pid]

stats.go

@@ -3,9 +3,10 @@ package containerd
 import "github.com/rcrowley/go-metrics"
 
 var (
-	ContainerStartTimer = metrics.NewTimer()
+	ContainerStartTimer    = metrics.NewTimer()
-	ContainersCounter   = metrics.NewCounter()
+	ContainersCounter      = metrics.NewCounter()
-	EventsCounter       = metrics.NewCounter()
+	EventsCounter          = metrics.NewCounter()
+	EventSubscriberCounter = metrics.NewCounter()
 )
 
 func Metrics() map[string]interface{} {
@@ -13,5 +14,6 @@ func Metrics() map[string]interface{} {
 		"container-start-time": ContainerStartTimer,
 		"containers":           ContainersCounter,
 		"events":               EventsCounter,
+		"events-subscribers":   EventSubscriberCounter,
 	}
 }

supervisor.go

@@ -28,13 +28,14 @@ func NewSupervisor(id, stateDir string, tasks chan *StartTask) (*Supervisor, err
 		return nil, err
 	}
 	s := &Supervisor{
-		stateDir:   stateDir,
+		stateDir:    stateDir,
-		containers: make(map[string]runtime.Container),
+		containers:  make(map[string]runtime.Container),
-		processes:  make(map[int]runtime.Container),
+		processes:   make(map[int]runtime.Container),
-		runtime:    r,
+		runtime:     r,
-		tasks:      tasks,
+		tasks:       tasks,
-		events:     make(chan *Event, 2048),
+		events:      make(chan *Event, 2048),
-		machine:    machine,
+		machine:     machine,
+		subscribers: make(map[chan *Event]struct{}),
 	}
 	// register default event handlers
 	s.handlers = map[EventType]Handler{
@@ -62,13 +63,11 @@ type Supervisor struct {
 	runtime        runtime.Runtime
 	events         chan *Event
 	tasks          chan *StartTask
-	subscribers    map[subscriber]bool
+	subscribers    map[chan *Event]struct{}
 	machine        Machine
 	containerGroup sync.WaitGroup
 }
 
-type subscriber chan *Event
-
 func (s *Supervisor) Stop(sig chan os.Signal) {
 	// Close the tasks channel so that no new containers get started
 	close(s.tasks)
@@ -109,12 +108,16 @@ func (s *Supervisor) Close() error {
 	return nil
 }
 
-func (s *Supervisor) Events() subscriber {
+func (s *Supervisor) Events() chan *Event {
-	return subscriber(make(chan *Event))
+	c := make(chan *Event, 2048)
+	EventSubscriberCounter.Inc(1)
+	s.subscribers[c] = struct{}{}
+	return c
 }
 
-func (s *Supervisor) Unsubscribe(sub subscriber) {
+func (s *Supervisor) Unsubscribe(sub chan *Event) {
 	delete(s.subscribers, sub)
+	EventSubscriberCounter.Dec(1)
 }
 
 func (s *Supervisor) NotifySubscribers(e *Event) {