
This change further plumbs the components required for implementing event filters. Specifically, we now have the ability to filter on the `topic` and `namespace`. In the course of implementing this functionality, it was found that there were mismatches in the events API that created extra serialization round trips. A modification to `typeurl.MarshalAny` and a clear separation between publishing and forwarding allow us to avoid these serialization issues. Unfortunately, this has required a few tweaks to the GRPC API, so this is a breaking change. `Publish` and `Forward` have been clearly separated in the GRPC API. `Publish` honors the contextual namespace and performs timestamping while `Forward` simply validates and forwards. The behavior of `Subscribe` is to propagate events for all namespaces unless specifically filtered (and hence the relation to this particular change. The following is an example of using filters to monitor the task events generated while running the [bucketbench tool](https://github.com/estesp/bucketbench): ``` $ ctr events 'topic~=/tasks/.+,namespace==bb' ... 2017-07-28 22:19:51.78944874 +0000 UTC bb /tasks/start {"container_id":"bb-ctr-6-8","pid":25889} 2017-07-28 22:19:51.791893688 +0000 UTC bb /tasks/start {"container_id":"bb-ctr-4-8","pid":25882} 2017-07-28 22:19:51.792608389 +0000 UTC bb /tasks/start {"container_id":"bb-ctr-2-9","pid":25860} 2017-07-28 22:19:51.793035217 +0000 UTC bb /tasks/start {"container_id":"bb-ctr-5-6","pid":25869} 2017-07-28 22:19:51.802659622 +0000 UTC bb /tasks/start {"container_id":"bb-ctr-0-7","pid":25877} 2017-07-28 22:19:51.805192898 +0000 UTC bb /tasks/start {"container_id":"bb-ctr-3-6","pid":25856} 2017-07-28 22:19:51.832374931 +0000 UTC bb /tasks/exit {"container_id":"bb-ctr-8-6","id":"bb-ctr-8-6","pid":25864,"exited_at":"2017-07-28T22:19:51.832013043Z"} 2017-07-28 22:19:51.84001249 +0000 UTC bb /tasks/exit {"container_id":"bb-ctr-2-9","id":"bb-ctr-2-9","pid":25860,"exited_at":"2017-07-28T22:19:51.839717714Z"} 2017-07-28 22:19:51.840272635 +0000 UTC bb /tasks/exit {"container_id":"bb-ctr-7-6","id":"bb-ctr-7-6","pid":25855,"exited_at":"2017-07-28T22:19:51.839796335Z"} ... ``` In addition to the events changes, we now display the namespace origin of the event in the cli tool. This will be followed by a PR to add individual field filtering for the events API for each event type. Signed-off-by: Stephen J Day <stephen.day@docker.com>
145 lines
3.1 KiB
Go
145 lines
3.1 KiB
Go
package typeurl
|
|
|
|
import (
|
|
"encoding/json"
|
|
"path"
|
|
"reflect"
|
|
"strings"
|
|
"sync"
|
|
|
|
"github.com/containerd/containerd/errdefs"
|
|
"github.com/gogo/protobuf/proto"
|
|
"github.com/gogo/protobuf/types"
|
|
"github.com/pkg/errors"
|
|
)
|
|
|
|
const Prefix = "types.containerd.io"
|
|
|
|
var (
|
|
mu sync.Mutex
|
|
registry = make(map[reflect.Type]string)
|
|
)
|
|
|
|
// Register a type with the base url of the type
|
|
func Register(v interface{}, args ...string) {
|
|
var (
|
|
t = tryDereference(v)
|
|
p = path.Join(append([]string{Prefix}, args...)...)
|
|
)
|
|
mu.Lock()
|
|
defer mu.Unlock()
|
|
if et, ok := registry[t]; ok {
|
|
if et != p {
|
|
panic(errors.Errorf("type registred with alternate path %q != %q", et, p))
|
|
}
|
|
return
|
|
}
|
|
registry[t] = p
|
|
}
|
|
|
|
// TypeURL returns the type url for a registred type
|
|
func TypeURL(v interface{}) (string, error) {
|
|
mu.Lock()
|
|
u, ok := registry[tryDereference(v)]
|
|
mu.Unlock()
|
|
if !ok {
|
|
// fallback to the proto registry if it is a proto message
|
|
pb, ok := v.(proto.Message)
|
|
if !ok {
|
|
return "", errors.Wrapf(errdefs.ErrNotFound, "type %s", reflect.TypeOf(v))
|
|
}
|
|
return path.Join(Prefix, proto.MessageName(pb)), nil
|
|
}
|
|
return u, nil
|
|
}
|
|
|
|
// Is returns true if the type of the Any is the same as v
|
|
func Is(any *types.Any, v interface{}) bool {
|
|
// call to check that v is a pointer
|
|
tryDereference(v)
|
|
url, err := TypeURL(v)
|
|
if err != nil {
|
|
return false
|
|
}
|
|
return any.TypeUrl == url
|
|
}
|
|
|
|
// MarshalAny marshals the value v into an any with the correct TypeUrl
|
|
func MarshalAny(v interface{}) (*types.Any, error) {
|
|
var marshal func(v interface{}) ([]byte, error)
|
|
switch t := v.(type) {
|
|
case *types.Any:
|
|
// avoid reserializing the type if we have an any.
|
|
return t, nil
|
|
case proto.Message:
|
|
marshal = func(v interface{}) ([]byte, error) {
|
|
return proto.Marshal(t)
|
|
}
|
|
default:
|
|
marshal = json.Marshal
|
|
}
|
|
|
|
url, err := TypeURL(v)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
data, err := marshal(v)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return &types.Any{
|
|
TypeUrl: url,
|
|
Value: data,
|
|
}, nil
|
|
}
|
|
|
|
// UnmarshalAny unmarshals the any type into a concrete type
|
|
func UnmarshalAny(any *types.Any) (interface{}, error) {
|
|
t, err := getTypeByUrl(any.TypeUrl)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
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
|
|
}
|
|
|
|
type urlType struct {
|
|
t reflect.Type
|
|
isProto bool
|
|
}
|
|
|
|
func getTypeByUrl(url string) (urlType, error) {
|
|
for t, u := range registry {
|
|
if u == url {
|
|
return urlType{
|
|
t: t,
|
|
}, nil
|
|
}
|
|
}
|
|
// 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{}, errors.Wrapf(errdefs.ErrNotFound, "type with url %s", url)
|
|
}
|
|
|
|
func tryDereference(v interface{}) reflect.Type {
|
|
t := reflect.TypeOf(v)
|
|
if t.Kind() == reflect.Ptr {
|
|
// require check of pointer but dereference to register
|
|
return t.Elem()
|
|
}
|
|
panic("v is not a pointer to a type")
|
|
}
|