diff --git a/cmd/ctr/events.go b/cmd/ctr/events.go index da3cdf547..7367629d5 100644 --- a/cmd/ctr/events.go +++ b/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 := "" - - var de events.DynamicEvent - if err := events.UnmarshalEvent(env.Event, &de); err != nil { + v, err := typeurl.UnmarshalAny(env.Event) + if err != nil { return "", err } - - switch e := de.Event.(type) { + switch e := v.(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 } diff --git a/cmd/ctr/shim.go b/cmd/ctr/shim.go index b291124f3..7932490e3 100644 --- a/cmd/ctr/shim.go +++ b/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 } diff --git a/events/convert.go b/events/convert.go deleted file mode 100644 index d9872298f..000000000 --- a/events/convert.go +++ /dev/null @@ -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) -} diff --git a/events/convert_test.go b/events/convert_test.go deleted file mode 100644 index ead3528c8..000000000 --- a/events/convert_test.go +++ /dev/null @@ -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) - } - } -} diff --git a/events/sink.go b/events/sink.go index c02cf7064..d94319bc6 100644 --- a/events/sink.go +++ b/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 } diff --git a/process.go b/process.go index 336d7eb0f..9d9c5ee22 100644 --- a/process.go +++ b/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 } - - switch { - case events.Is(evt.Event, &eventsapi.RuntimeEvent{}): - var e eventsapi.RuntimeEvent - if err := events.UnmarshalEvent(evt.Event, &e); err != nil { - return UnknownExitStatus, err - } - + v, err := typeurl.UnmarshalAny(evt.Event) + if err != nil { + return UnknownExitStatus, err + } + if e, ok := v.(*eventsapi.RuntimeEvent); ok { if e.Type != eventsapi.RuntimeEvent_EXIT { continue evloop } diff --git a/task.go b/task.go index b1c38a6c5..d64cf6916 100644 --- a/task.go +++ b/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 } - - if !events.Is(evt.Event, &eventsapi.RuntimeEvent{}) { - continue - } - - if err := events.UnmarshalEvent(evt.Event, &e); err != nil { - return UnknownExitStatus, err - } - - if e.Type != eventsapi.RuntimeEvent_EXIT { - continue - } - - if e.ID == t.containerID && e.Pid == t.pid { - return e.ExitStatus, nil + v, err := typeurl.UnmarshalAny(evt.Event) + if e, ok := v.(*eventsapi.RuntimeEvent); ok { + if e.Type != eventsapi.RuntimeEvent_EXIT { + continue + } + if e.ID == t.containerID && e.Pid == t.pid { + return e.ExitStatus, nil + } } } } diff --git a/typeurl/types.go b/typeurl/types.go index 7dc00c019..19fee0208 100644 --- a/typeurl/types.go +++ b/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 -func TypeUrl(v interface{}) (string, error) { +// TypeURL returns the type url for a registred type +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() - switch dt := v.(type) { - case proto.Message: - err = proto.Unmarshal(any.Value, dt) - default: + v := reflect.New(t.t).Interface() + if t.isProto { + err = proto.Unmarshal(any.Value, v.(proto.Message)) + } else { 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 { diff --git a/typeurl/types_test.go b/typeurl/types_test.go index aad840c0b..8056c17ab 100644 --- a/typeurl/types_test.go +++ b/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) + } + } +}