Merge pull request #2274 from HusterWan/zr/fix-events-connect

bugfix:  event api connection may be recycled if keeps idle too long

apis/server/router.go

@@ -28,7 +28,7 @@ func initRoute(s *Server) http.Handler {
 	s.addRoute(r, http.MethodGet, "/info", s.info)
 	s.addRoute(r, http.MethodGet, "/version", s.version)
 	s.addRoute(r, http.MethodPost, "/auth", s.auth)
-	s.addRoute(r, http.MethodGet, "/events", s.events)
+	s.addRoute(r, http.MethodGet, "/events", withCancelHandler(s.events))
 
 	// daemon, we still list this API into system manager.
 	s.addRoute(r, http.MethodPost, "/daemon/update", s.updateDaemon)

apis/server/system_bridge.go

@@ -14,6 +14,7 @@ import (
 
 	"github.com/docker/docker/pkg/ioutils"
 	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
 )
 
 func (s *Server) ping(context context.Context, rw http.ResponseWriter, req *http.Request) (err error) {
@@ -69,9 +70,6 @@ func (s *Server) auth(ctx context.Context, rw http.ResponseWriter, req *http.Req
 }
 
 func (s *Server) events(ctx context.Context, rw http.ResponseWriter, req *http.Request) (err error) {
-	ctx, cancel := context.WithCancel(ctx)
-	defer cancel()
-
 	rw.Header().Set("Content-Type", "application/json")
 	output := ioutils.NewWriteFlusher(rw)
 	defer output.Close()
@@ -128,6 +126,7 @@ func (s *Server) events(ctx context.Context, rw http.ResponseWriter, req *http.R
 		select {
 		case ev := <-eventq:
 			if err := enc.Encode(ev); err != nil {
+				logrus.Errorf("encode events got an error: %v", err)
 				return err
 			}
 		case err := <-errq:
@@ -137,6 +136,9 @@ func (s *Server) events(ctx context.Context, rw http.ResponseWriter, req *http.R
 			return nil
 		case <-timeout:
 			return nil
+		case <-ctx.Done():
+			logrus.Debug("client context is cancelled, stop sending events")
+			return nil
 		}
 	}
 }

cli/events.go

@@ -94,15 +94,24 @@ type eventProcessor func(event types.EventsMessage, err error) error
 // Each output includes the event type, actor id, name and action.
 // Actor attributes are printed at the end if the actor has any.
 func printOutput(event types.EventsMessage, output io.Writer) {
+	// skip empty event message
+	if event == (types.EventsMessage{}) {
+		return
+	}
+
 	if event.TimeNano != 0 {
 		fmt.Fprintf(output, "%s ", time.Unix(0, event.TimeNano).Format(utils.RFC3339NanoFixed))
 	} else if event.Time != 0 {
 		fmt.Fprintf(output, "%s ", time.Unix(event.Time, 0).Format(utils.RFC3339NanoFixed))
 	}
 
-	fmt.Fprintf(output, "%s %s %s", event.Type, event.Action, event.Actor.ID)
+	id := ""
+	if event.Actor != nil {
+		id = event.Actor.ID
+	}
+	fmt.Fprintf(output, "%s %s %s", event.Type, event.Action, id)
 
-	if len(event.Actor.Attributes) > 0 {
+	if event.Actor != nil && len(event.Actor.Attributes) > 0 {
 		var attrs []string
 		var keys []string
 		for k := range event.Actor.Attributes {

ctrd/client.go

@@ -349,7 +349,7 @@ func (c *Client) Cleanup() error {
 // collectContainerdEvents collects events generated by containerd.
 func (c *Client) collectContainerdEvents() {
 	ctx := context.Background()
-	topicsToHandle := []string{TaskOOMEventTopic, ContainersDeleteEventTopic}
+	topicsToHandle := []string{TaskOOMEventTopic, TaskExitEventTopic}
 
 	// set filters for subscribe containerd events,
 	// now we only care about task and container events.
@@ -384,15 +384,15 @@ func (c *Client) collectContainerdEvents() {
 		}
 
 		switch e.Topic {
-		case ContainersDeleteEventTopic:
-			cDelEvent, ok := out.(*eventstypes.ContainerDelete)
+		case TaskExitEventTopic:
+			exitEvent, ok := out.(*eventstypes.TaskExit)
 			if !ok {
-				logrus.Warnf("failed to parse %s event: %#v", ContainersDeleteEventTopic, out)
+				logrus.Warnf("failed to parse %s event: %#v", TaskExitEventTopic, out)
 				continue
 			}
-
 			action = "die"
-			containerID = cDelEvent.ID
+			containerID = exitEvent.ContainerID
+			attributes["exitcode"] = strconv.Itoa(int(exitEvent.ExitStatus))
 		case TaskOOMEventTopic:
 			oomEvent, ok := out.(*eventstypes.TaskOOM)
 			if !ok {