events: move types into service package

When using events, it was found to be fairly unwieldy with a number of
extra packages. For the most part, when interacting with the events
service, we want types of the same version of the service. This has been
accomplished by moving all events types into the events package.

In addition, several fixes to the way events are marshaled have been
included. Specifically, we defer to the protobuf type registration
system to assemble events and type urls, with a little bit sheen on top
of add a containerd.io oriented namespace.

This has resulted in much cleaner event consumption and has removed the
reliance on error prone type urls, in favor of concrete types.

Signed-off-by: Stephen J Day <stephen.day@docker.com>

events/convert.go

@@ -1,92 +1,72 @@
 package events
 
 import (
-	"path"
-	"strings"
-
-	"github.com/containerd/containerd/api/types/event"
 	"github.com/gogo/protobuf/proto"
-	protobuf "github.com/gogo/protobuf/types"
+	"github.com/gogo/protobuf/types"
 	"github.com/pkg/errors"
 )
 
-func getUrl(name string) string {
-	base := "types.containerd.io"
-	return path.Join(base, strings.Join([]string{
-		"containerd",
-		EventVersion,
-		"types",
-		"event",
-		name,
-	}, "."))
-}
+const (
+	typesPrefix = "types.containerd.io/"
+)
 
-func convertToAny(evt Event) (*protobuf.Any, error) {
-	url := ""
-	var pb proto.Message
-	switch v := evt.(type) {
-	case event.ContainerCreate:
-		url = getUrl("ContainerCreate")
-		pb = &v
-	case event.ContainerDelete:
-		url = getUrl("ContainerDelete")
-		pb = &v
-	case event.TaskCreate:
-		url = getUrl("TaskCreate")
-		pb = &v
-	case event.TaskStart:
-		url = getUrl("TaskStart")
-		pb = &v
-	case event.TaskDelete:
-		url = getUrl("TaskDelete")
-		pb = &v
-	case event.ContentDelete:
-		url = getUrl("ContentDelete")
-		pb = &v
-	case event.SnapshotPrepare:
-		url = getUrl("SnapshotPrepare")
-		pb = &v
-	case event.SnapshotCommit:
-		url = getUrl("SnapshotCommit")
-		pb = &v
-	case event.SnapshotRemove:
-		url = getUrl("SnapshotRemove")
-		pb = &v
-	case event.ImageUpdate:
-		url = getUrl("ImageUpdate")
-		pb = &v
-	case event.ImageDelete:
-		url = getUrl("ImageDelete")
-		pb = &v
-	case event.NamespaceCreate:
-		url = getUrl("NamespaceCreate")
-		pb = &v
-	case event.NamespaceUpdate:
-		url = getUrl("NamespaceUpdate")
-		pb = &v
-	case event.NamespaceDelete:
-		url = getUrl("NamespaceDelete")
-		pb = &v
-	case event.RuntimeCreate:
-		url = getUrl("RuntimeCreate")
-		pb = &v
-	case event.RuntimeEvent:
-		url = getUrl("RuntimeEvent")
-		pb = &v
-	case event.RuntimeDelete:
-		url = getUrl("RuntimeDelete")
-		pb = &v
-	default:
-		return nil, errors.Errorf("unsupported event type: %T", v)
+// MarshalEvent marshal the event into an any, namespacing the type url to the
+// containerd types.
+func MarshalEvent(event Event) (*types.Any, error) {
+	pb, ok := event.(proto.Message)
+	if !ok {
+		return nil, errors.Errorf("%T not a protobuf", event)
 	}
 
+	url := typesPrefix + proto.MessageName(pb)
 	val, err := proto.Marshal(pb)
 	if err != nil {
 		return nil, err
 	}
 
-	return &protobuf.Any{
+	return &types.Any{
 		TypeUrl: url,
 		Value:   val,
 	}, nil
 }
+
+// DynamEvent acts as a holder type for unmarshaling events where the type is
+// not previously known.
+type DynamicEvent struct {
+	Event
+}
+
+// UnmarshalEvent provides an event object based on the provided any.
+//
+// Use with DynamicEvent (or protobuf/types.DynamicAny) if the type is not
+// known before hand.
+func UnmarshalEvent(any *types.Any, event Event) error {
+	switch v := event.(type) {
+	case proto.Message:
+		if err := types.UnmarshalAny(any, v); err != nil {
+			return errors.Wrapf(err, "failed to unmarshal event %v", any)
+		}
+	case *DynamicEvent:
+		var da types.DynamicAny
+
+		if err := types.UnmarshalAny(any, &da); err != nil {
+			return errors.Wrapf(err, "failed to unmarshal event %v", any)
+		}
+		v.Event = da.Message
+	default:
+		return errors.Errorf("unsupported event type: %T", event)
+	}
+
+	return nil
+
+}
+
+// Is returns true if the event in any will unmarashal into the provided event.
+func Is(any *types.Any, event Event) bool {
+	pb, ok := event.(proto.Message)
+	if !ok {
+		return false
+	}
+
+	return types.Is(any, pb)
+}

events/emitter.go

@@ -4,7 +4,7 @@ import (
 	"context"
 	"sync"
 
-	"github.com/containerd/containerd/api/types/event"
+	events "github.com/containerd/containerd/api/services/events/v1"
 	"github.com/containerd/containerd/namespaces"
 	goevents "github.com/docker/go-events"
 )
@@ -38,12 +38,12 @@ func (e *Emitter) Post(ctx context.Context, evt Event) error {
 	return nil
 }
 
-func (e *Emitter) Events(ctx context.Context, clientID string) chan *event.Envelope {
+func (e *Emitter) Events(ctx context.Context, clientID string) chan *events.Envelope {
 	e.m.Lock()
 	if _, ok := e.sinks[clientID]; !ok {
 		ns, _ := namespaces.Namespace(ctx)
 		s := &eventSink{
-			ch: make(chan *event.Envelope),
+			ch: make(chan *events.Envelope),
 			ns: ns,
 		}
 		e.sinks[clientID] = s

events/sink.go

@@ -5,7 +5,7 @@ import (
 	"time"
 
 	"github.com/Sirupsen/logrus"
-	"github.com/containerd/containerd/api/types/event"
+	"github.com/containerd/containerd/api/services/events/v1"
 	"github.com/containerd/containerd/namespaces"
 	goevents "github.com/docker/go-events"
 	"github.com/pkg/errors"
@@ -18,7 +18,7 @@ type sinkEvent struct {
 
 type eventSink struct {
 	ns string
-	ch chan *event.Envelope
+	ch chan *events.Envelope
 }
 
 func (s *eventSink) Write(evt goevents.Event) error {
@@ -34,7 +34,7 @@ func (s *eventSink) Write(evt goevents.Event) error {
 		return nil
 	}
 
-	eventData, err := convertToAny(e.event)
+	eventData, err := MarshalEvent(e.event)
 	if err != nil {
 		return err
 	}
@@ -45,7 +45,7 @@ func (s *eventSink) Write(evt goevents.Event) error {
 		"ns":    ns,
 	}).Debug("event")
 
-	s.ch <- &event.Envelope{
+	s.ch <- &events.Envelope{
 		Timestamp: time.Now(),
 		Topic:     topic,
 		Event:     eventData,