Replace events/convert with typeurl

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

cmd/ctr/events.go

@@ -7,7 +7,7 @@ import (
 	"text/tabwriter"
 
 	eventsapi "github.com/containerd/containerd/api/services/events/v1"
-	"github.com/containerd/containerd/events"
+	"github.com/containerd/containerd/typeurl"
 	"github.com/urfave/cli"
 )
 
@@ -57,13 +57,11 @@ var eventsCommand = cli.Command{
 
 func getEventOutput(env *eventsapi.Envelope) (string, error) {
 	out := ""
-
+	v, err := typeurl.UnmarshalAny(env.Event)
-	var de events.DynamicEvent
+	if err != nil {
-	if err := events.UnmarshalEvent(env.Event, &de); err != nil {
 		return "", err
 	}
-
+	switch e := v.(type) {
-	switch e := de.Event.(type) {
 	case *eventsapi.ContainerCreate:
 		out = fmt.Sprintf("id=%s image=%s runtime=%s", e.ContainerID, e.Image, e.Runtime)
 	case *eventsapi.TaskCreate:
@@ -105,6 +103,5 @@ func getEventOutput(env *eventsapi.Envelope) (string, error) {
 	default:
 		out = env.Event.TypeUrl
 	}
-
 	return out, nil
 }

cmd/ctr/shim.go

@@ -236,7 +236,7 @@ var shimExecCommand = cli.Command{
 		if err != nil {
 			return err
 		}
-		url, err := typeurl.TypeUrl(specs.Process{})
+		url, err := typeurl.TypeURL(specs.Process{})
 		if err != nil {
 			return err
 		}

events/convert.go

@@ -1,72 +0,0 @@
-package events
-
-import (
-	"github.com/gogo/protobuf/proto"
-	"github.com/gogo/protobuf/types"
-	"github.com/pkg/errors"
-)
-
-const (
-	typesPrefix = "types.containerd.io/"
-)
-
-// 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 &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/convert_test.go

@@ -1,62 +0,0 @@
-package events
-
-import (
-	"fmt"
-	"reflect"
-	"testing"
-
-	events "github.com/containerd/containerd/api/services/events/v1"
-)
-
-func TestMarshalEvent(t *testing.T) {
-	for _, testcase := range []struct {
-		event Event
-		url   string
-	}{
-		{
-			event: &events.TaskStart{},
-			url:   "types.containerd.io/containerd.services.events.v1.TaskStart",
-		},
-
-		{
-			event: &events.NamespaceUpdate{},
-			url:   "types.containerd.io/containerd.services.events.v1.NamespaceUpdate",
-		},
-	} {
-		t.Run(fmt.Sprintf("%T", testcase.event), func(t *testing.T) {
-			a, err := MarshalEvent(testcase.event)
-			if err != nil {
-				t.Fatal(err)
-			}
-			if a.TypeUrl != testcase.url {
-				t.Fatalf("unexpected url: %v != %v", a.TypeUrl, testcase.url)
-			}
-
-			var de DynamicEvent
-			if err := UnmarshalEvent(a, &de); err != nil {
-				t.Fatal(err)
-			}
-
-			if !reflect.DeepEqual(de.Event, testcase.event) {
-				t.Fatalf("round trip failed %v != %v", de.Event, testcase.event)
-			}
-		})
-	}
-}
-
-func BenchmarkMarshalEvent(b *testing.B) {
-	ev := &events.TaskStart{}
-	expected, err := MarshalEvent(ev)
-	if err != nil {
-		b.Fatal(err)
-	}
-	for i := 0; i < b.N; i++ {
-		a, err := MarshalEvent(ev)
-		if err != nil {
-			b.Fatal(err)
-		}
-		if a.TypeUrl != expected.TypeUrl {
-			b.Fatalf("incorrect type url: %v != %v", a, expected)
-		}
-	}
-}

events/sink.go

@@ -8,6 +8,7 @@ import (
 	"github.com/containerd/containerd/api/services/events/v1"
 	"github.com/containerd/containerd/log"
 	"github.com/containerd/containerd/namespaces"
+	"github.com/containerd/containerd/typeurl"
 	goevents "github.com/docker/go-events"
 	"github.com/pkg/errors"
 )
@@ -35,7 +36,7 @@ func (s *eventSink) Write(evt goevents.Event) error {
 		return nil
 	}
 
-	eventData, err := MarshalEvent(e.event)
+	eventData, err := typeurl.MarshalAny(e.event)
 	if err != nil {
 		return err
 	}

process.go

@@ -6,7 +6,6 @@ import (
 
 	eventsapi "github.com/containerd/containerd/api/services/events/v1"
 	"github.com/containerd/containerd/api/services/tasks/v1"
-	"github.com/containerd/containerd/events"
 	"github.com/containerd/containerd/typeurl"
 	specs "github.com/opencontainers/runtime-spec/specs-go"
 )
@@ -79,14 +78,11 @@ evloop:
 		if err != nil {
 			return UnknownExitStatus, err
 		}
-
+		v, err := typeurl.UnmarshalAny(evt.Event)
-		switch {
+		if err != nil {
-		case events.Is(evt.Event, &eventsapi.RuntimeEvent{}):
+			return UnknownExitStatus, err
-			var e eventsapi.RuntimeEvent
+		}
-			if err := events.UnmarshalEvent(evt.Event, &e); err != nil {
+		if e, ok := v.(*eventsapi.RuntimeEvent); ok {
-				return UnknownExitStatus, err
-			}
-
 			if e.Type != eventsapi.RuntimeEvent_EXIT {
 				continue evloop
 			}

task.go

@@ -14,9 +14,9 @@ import (
 	eventsapi "github.com/containerd/containerd/api/services/events/v1"
 	"github.com/containerd/containerd/api/services/tasks/v1"
 	"github.com/containerd/containerd/content"
-	"github.com/containerd/containerd/events"
 	"github.com/containerd/containerd/rootfs"
 	"github.com/containerd/containerd/runtime"
+	"github.com/containerd/containerd/typeurl"
 	"github.com/opencontainers/image-spec/specs-go/v1"
 	specs "github.com/opencontainers/runtime-spec/specs-go"
 	"google.golang.org/grpc"
@@ -149,35 +149,25 @@ func (t *task) Status(ctx context.Context) (TaskStatus, error) {
 
 // Wait is a blocking call that will wait for the task to exit and return the exit status
 func (t *task) Wait(ctx context.Context) (uint32, error) {
-	// TODO (ehazlett): add filtering for specific event
 	eventstream, err := t.client.EventService().Stream(ctx, &eventsapi.StreamEventsRequest{})
 	if err != nil {
 		return UnknownExitStatus, err
 	}
 	<-t.pidSync
 
-	var e eventsapi.RuntimeEvent
-
 	for {
 		evt, err := eventstream.Recv()
 		if err != nil {
 			return UnknownExitStatus, err
 		}
-
+		v, err := typeurl.UnmarshalAny(evt.Event)
-		if !events.Is(evt.Event, &eventsapi.RuntimeEvent{}) {
+		if e, ok := v.(*eventsapi.RuntimeEvent); ok {
-			continue
+			if e.Type != eventsapi.RuntimeEvent_EXIT {
-		}
+				continue
-
+			}
-		if err := events.UnmarshalEvent(evt.Event, &e); err != nil {
+			if e.ID == t.containerID && e.Pid == t.pid {
-			return UnknownExitStatus, err
+				return e.ExitStatus, nil
-		}
+			}
-
-		if e.Type != eventsapi.RuntimeEvent_EXIT {
-			continue
-		}
-
-		if e.ID == t.containerID && e.Pid == t.pid {
-			return e.ExitStatus, nil
 		}
 	}
 }

typeurl/types.go

@@ -3,15 +3,16 @@ package typeurl
 import (
 	"encoding/json"
 	"errors"
-	"path/filepath"
+	"path"
 	"reflect"
+	"strings"
 	"sync"
 
 	"github.com/gogo/protobuf/proto"
 	"github.com/gogo/protobuf/types"
 )
 
-const Namespace = "types.containerd.io"
+const Prefix = "types.containerd.io"
 
 var (
 	mu            sync.Mutex
@@ -28,16 +29,21 @@ func Register(v interface{}, args ...string) {
 	if _, ok := registry[t]; ok {
 		panic(ErrRegistered)
 	}
-	registry[t] = filepath.Join(append([]string{Namespace}, args...)...)
+	registry[t] = path.Join(append([]string{Prefix}, args...)...)
 }
 
-// TypeUrl returns the type url for a registred type
+// TypeURL returns the type url for a registred type
-func TypeUrl(v interface{}) (string, error) {
+func TypeURL(v interface{}) (string, error) {
 	mu.Lock()
 	defer mu.Unlock()
 	u, ok := registry[tryDereference(v)]
 	if !ok {
-		return "", ErrNotExists
+		// fallback to the proto registry if it is a proto message
+		pb, ok := v.(proto.Message)
+		if !ok {
+			return "", ErrNotExists
+		}
+		return path.Join(Prefix, proto.MessageName(pb)), nil
 	}
 	return u, nil
 }
@@ -47,7 +53,7 @@ func MarshalAny(v interface{}) (*types.Any, error) {
 		err  error
 		data []byte
 	)
-	url, err := TypeUrl(v)
+	url, err := TypeURL(v)
 	if err != nil {
 		return nil, err
 	}
@@ -71,23 +77,38 @@ func UnmarshalAny(any *types.Any) (interface{}, error) {
 	if err != nil {
 		return nil, err
 	}
-	v := reflect.New(t).Interface()
+	v := reflect.New(t.t).Interface()
-	switch dt := v.(type) {
+	if t.isProto {
-	case proto.Message:
+		err = proto.Unmarshal(any.Value, v.(proto.Message))
-		err = proto.Unmarshal(any.Value, dt)
+	} else {
-	default:
 		err = json.Unmarshal(any.Value, v)
 	}
 	return v, err
 }
 
-func getTypeByUrl(url string) (reflect.Type, error) {
+type urlType struct {
+	t       reflect.Type
+	isProto bool
+}
+
+func getTypeByUrl(url string) (urlType, error) {
 	for t, u := range registry {
 		if u == url {
-			return t, nil
+			return urlType{
+				t: t,
+			}, nil
 		}
 	}
-	return nil, ErrNotExists
+	// fallback to proto registry
+	t := proto.MessageType(strings.TrimPrefix(url, Prefix+"/"))
+	if t != nil {
+		return urlType{
+			// get the underlying Elem because proto returns a pointer to the type
+			t:       t.Elem(),
+			isProto: true,
+		}, nil
+	}
+	return urlType{}, ErrNotExists
 }
 
 func tryDereference(v interface{}) reflect.Type {

typeurl/types_test.go

@@ -1,9 +1,13 @@
 package typeurl
 
 import (
+	"fmt"
 	"path/filepath"
 	"reflect"
 	"testing"
+
+	eventsapi "github.com/containerd/containerd/api/services/events/v1"
+	events "github.com/containerd/containerd/events"
 )
 
 type test struct {
@@ -17,10 +21,10 @@ func clear() {
 
 func TestRegsiterValueGetValue(t *testing.T) {
 	clear()
-	expected := filepath.Join(Namespace, "test")
+	expected := filepath.Join(Prefix, "test")
 	Register(test{}, "test")
 
-	url, err := TypeUrl(test{})
+	url, err := TypeURL(test{})
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -31,10 +35,10 @@ func TestRegsiterValueGetValue(t *testing.T) {
 
 func TestRegsiterValueGetPointer(t *testing.T) {
 	clear()
-	expected := filepath.Join(Namespace, "test")
+	expected := filepath.Join(Prefix, "test")
 	Register(test{}, "test")
 
-	url, err := TypeUrl(&test{})
+	url, err := TypeURL(&test{})
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -45,10 +49,10 @@ func TestRegsiterValueGetPointer(t *testing.T) {
 
 func TestRegsiterPointerGetPointer(t *testing.T) {
 	clear()
-	expected := filepath.Join(Namespace, "test")
+	expected := filepath.Join(Prefix, "test")
 	Register(&test{}, "test")
 
-	url, err := TypeUrl(&test{})
+	url, err := TypeURL(&test{})
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -59,10 +63,10 @@ func TestRegsiterPointerGetPointer(t *testing.T) {
 
 func TestRegsiterPointerGetValue(t *testing.T) {
 	clear()
-	expected := filepath.Join(Namespace, "test")
+	expected := filepath.Join(Prefix, "test")
 	Register(&test{}, "test")
 
-	url, err := TypeUrl(test{})
+	url, err := TypeURL(test{})
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -73,7 +77,7 @@ func TestRegsiterPointerGetValue(t *testing.T) {
 
 func TestMarshal(t *testing.T) {
 	clear()
-	expected := filepath.Join(Namespace, "test")
+	expected := filepath.Join(Prefix, "test")
 	Register(test{}, "test")
 
 	v := &test{
@@ -116,3 +120,55 @@ func TestMarshalUnmarshal(t *testing.T) {
 		t.Fatal("invalid age")
 	}
 }
+
+func TestMarshalEvent(t *testing.T) {
+	for _, testcase := range []struct {
+		event events.Event
+		url   string
+	}{
+		{
+			event: &eventsapi.TaskStart{},
+			url:   "types.containerd.io/containerd.services.events.v1.TaskStart",
+		},
+
+		{
+			event: &eventsapi.NamespaceUpdate{},
+			url:   "types.containerd.io/containerd.services.events.v1.NamespaceUpdate",
+		},
+	} {
+		t.Run(fmt.Sprintf("%T", testcase.event), func(t *testing.T) {
+			a, err := MarshalAny(testcase.event)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if a.TypeUrl != testcase.url {
+				t.Fatalf("unexpected url: %v != %v", a.TypeUrl, testcase.url)
+			}
+
+			v, err := UnmarshalAny(a)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if !reflect.DeepEqual(v, testcase.event) {
+				t.Fatalf("round trip failed %v != %v", v, testcase.event)
+			}
+		})
+	}
+}
+
+func BenchmarkMarshalEvent(b *testing.B) {
+	ev := &eventsapi.TaskStart{}
+	expected, err := MarshalAny(ev)
+	if err != nil {
+		b.Fatal(err)
+	}
+	for i := 0; i < b.N; i++ {
+		a, err := MarshalAny(ev)
+		if err != nil {
+			b.Fatal(err)
+		}
+		if a.TypeUrl != expected.TypeUrl {
+			b.Fatalf("incorrect type url: %v != %v", a, expected)
+		}
+	}
+}