Merge pull request #1207 from mlaventure/update-linux-events-topic

Update linux events topic

linux/shim/service.go

@@ -20,6 +20,7 @@ import (
 	"github.com/containerd/containerd/log"
 	"github.com/containerd/containerd/namespaces"
 	"github.com/containerd/containerd/reaper"
+	"github.com/containerd/containerd/runtime"
 	"github.com/containerd/containerd/typeurl"
 	google_protobuf "github.com/golang/protobuf/ptypes/empty"
 	"github.com/pkg/errors"
@@ -392,7 +393,7 @@ func (s *Service) forward(client poster) {
 		if _, err := client.Post(s.context, &events.PostEventRequest{
 			Envelope: &events.Envelope{
 				Timestamp: time.Now(),
-				Topic:     "/task/" + getTopic(e),
+				Topic:     getTopic(e),
 				Event:     a,
 			},
 		}); err != nil {
@@ -404,23 +405,23 @@ func (s *Service) forward(client poster) {
 func getTopic(e interface{}) string {
 	switch e.(type) {
 	case *events.TaskCreate:
-		return "task-create"
+		return runtime.TaskCreateEventTopic
 	case *events.TaskStart:
-		return "task-start"
+		return runtime.TaskStartEventTopic
 	case *events.TaskOOM:
-		return "task-oom"
+		return runtime.TaskOOMEventTopic
 	case *events.TaskExit:
-		return "task-exit"
+		return runtime.TaskExitEventTopic
 	case *events.TaskDelete:
-		return "task-delete"
+		return runtime.TaskDeleteEventTopic
 	case *events.TaskExecAdded:
-		return "task-exec-added"
+		return runtime.TaskExecAddedEventTopic
 	case *events.TaskPaused:
-		return "task-paused"
+		return runtime.TaskPausedEventTopic
 	case *events.TaskResumed:
-		return "task-resumed"
+		return runtime.TaskResumedEventTopic
 	case *events.TaskCheckpointed:
-		return "task-checkpointed"
+		return runtime.TaskCheckpointedEventTopic
 	}
 	return "?"
 }

runtime/events.go

@@ -0,0 +1,13 @@
+package runtime
+
+const (
+	TaskCreateEventTopic       = "/tasks/create"
+	TaskStartEventTopic        = "/tasks/start"
+	TaskOOMEventTopic          = "/tasks/oom"
+	TaskExitEventTopic         = "/tasks/exit"
+	TaskDeleteEventTopic       = "/tasks/delete"
+	TaskExecAddedEventTopic    = "/tasks/exec-added"
+	TaskPausedEventTopic       = "/tasks/paused"
+	TaskResumedEventTopic      = "/tasks/resumed"
+	TaskCheckpointedEventTopic = "/tasks/checkpointed"
+)

services/tasks/service.go

@@ -9,7 +9,6 @@ import (
 	"path/filepath"
 
 	"github.com/boltdb/bolt"
-	eventsapi "github.com/containerd/containerd/api/services/events/v1"
 	api "github.com/containerd/containerd/api/services/tasks/v1"
 	"github.com/containerd/containerd/api/types"
 	"github.com/containerd/containerd/api/types/task"
@@ -148,12 +147,6 @@ func (s *Service) Create(ctx context.Context, r *api.CreateTaskRequest) (*api.Cr
 		log.G(ctx).Error(err)
 	}
 
-	if err := s.emit(ctx, "/tasks/create", &eventsapi.TaskCreate{
-		ContainerID: r.ContainerID,
-	}); err != nil {
-		return nil, err
-	}
-
 	return &api.CreateTaskResponse{
 		ContainerID: r.ContainerID,
 		Pid:         state.Pid,
@@ -169,12 +162,6 @@ func (s *Service) Start(ctx context.Context, r *api.StartTaskRequest) (*google_p
 		return nil, err
 	}
 
-	if err := s.emit(ctx, "/tasks/start", &eventsapi.TaskStart{
-		ContainerID: r.ContainerID,
-	}); err != nil {
-		return nil, err
-	}
-
 	return empty, nil
 }
 
@@ -191,13 +178,6 @@ func (s *Service) Delete(ctx context.Context, r *api.DeleteTaskRequest) (*api.De
 	if err != nil {
 		return nil, err
 	}
-	if err := s.emit(ctx, "/tasks/delete", &eventsapi.TaskDelete{
-		ContainerID: r.ContainerID,
-		Pid:         exit.Pid,
-		ExitStatus:  exit.Status,
-	}); err != nil {
-		return nil, err
-	}
 	return &api.DeleteResponse{
 		ExitStatus: exit.Status,
 		ExitedAt:   exit.Timestamp,