github/containerd
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 2 Wiki Activity

events: refactor event distribution

Browse Source 
In the course of setting out to add filters and address some cleanup, it
was found that we had a few problems in the events subsystem that needed
addressing before moving forward.

The biggest change was to move to the more standard terminology of
publish and subscribe. We make this terminology change across the Go
interface and the GRPC API, making the behavior more familier. The
previous system was very context-oriented, which is no longer required.

With this, we've removed a large amount of dead and unneeded code. Event
transactions, context storage and the concept of `Poster` is gone. This
has been replaced in most places with a `Publisher`, which matches the
actual usage throughout the codebase, removing the need for helpers.

There are still some questions around the way events are handled in the
shim. Right now, we've preserved some of the existing bugs which may
require more extensive changes to resolve correctly.

Signed-off-by: Stephen J Day <stephen.day@docker.com>
This commit is contained in:
Stephen J Day 2017-07-24 22:19:02 -07:00
parent a9ab45fb77
commit a615a6fe5d
No known key found for this signature in database
GPG Key ID: 67B3DED84EDC823F
30 changed files with 669 additions and 644 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
api/services/events/v1/container.pb.go
View File

@ -19,8 +19,8 @@
ContainerUpdate ContainerUpdate
ContainerDelete ContainerDelete
ContentDelete ContentDelete
StreamEventsRequest SubscribeRequest
PostEventRequest PublishRequest
Envelope Envelope
ImageCreate ImageCreate
ImageUpdate ImageUpdate

304
api/services/events/v1/events.pb.go
View File

@ -32,25 +32,27 @@ var _ = fmt.Errorf
var _ = math.Inf var _ = math.Inf
var _ = time.Kitchen var _ = time.Kitchen
type StreamEventsRequest struct { type SubscribeRequest struct {
Filters []string `protobuf:"bytes,1,rep,name=filters" json:"filters,omitempty"`
} }
func (m *StreamEventsRequest) Reset() { *m = StreamEventsRequest{} } func (m *SubscribeRequest) Reset() { *m = SubscribeRequest{} }
func (*StreamEventsRequest) ProtoMessage() {} func (*SubscribeRequest) ProtoMessage() {}
func (*StreamEventsRequest) Descriptor() ([]byte, []int) { return fileDescriptorEvents, []int{0} } func (*SubscribeRequest) Descriptor() ([]byte, []int) { return fileDescriptorEvents, []int{0} }
type PostEventRequest struct { type PublishRequest struct {
Envelope *Envelope `protobuf:"bytes,1,opt,name=envelope" json:"envelope,omitempty"` Envelope *Envelope `protobuf:"bytes,1,opt,name=envelope" json:"envelope,omitempty"`
} }
func (m *PostEventRequest) Reset() { *m = PostEventRequest{} } func (m *PublishRequest) Reset() { *m = PublishRequest{} }
func (*PostEventRequest) ProtoMessage() {} func (*PublishRequest) ProtoMessage() {}
func (*PostEventRequest) Descriptor() ([]byte, []int) { return fileDescriptorEvents, []int{1} } func (*PublishRequest) Descriptor() ([]byte, []int) { return fileDescriptorEvents, []int{1} }
type Envelope struct { type Envelope struct {
Timestamp time.Time `protobuf:"bytes,1,opt,name=timestamp,stdtime" json:"timestamp"` Timestamp time.Time `protobuf:"bytes,1,opt,name=timestamp,stdtime" json:"timestamp"`
Topic string `protobuf:"bytes,2,opt,name=topic,proto3" json:"topic,omitempty"` Namespace string `protobuf:"bytes,2,opt,name=namespace,proto3" json:"namespace,omitempty"`
Event *google_protobuf1.Any `protobuf:"bytes,3,opt,name=event" json:"event,omitempty"` Topic string `protobuf:"bytes,3,opt,name=topic,proto3" json:"topic,omitempty"`
Event *google_protobuf1.Any `protobuf:"bytes,4,opt,name=event" json:"event,omitempty"`
} }
func (m *Envelope) Reset() { *m = Envelope{} } func (m *Envelope) Reset() { *m = Envelope{} }
@ -58,8 +60,8 @@ func (*Envelope) ProtoMessage() {}
func (*Envelope) Descriptor() ([]byte, []int) { return fileDescriptorEvents, []int{2} } func (*Envelope) Descriptor() ([]byte, []int) { return fileDescriptorEvents, []int{2} }
func init() { func init() {
proto.RegisterType((*StreamEventsRequest)(nil), "containerd.services.events.v1.StreamEventsRequest") proto.RegisterType((*SubscribeRequest)(nil), "containerd.services.events.v1.SubscribeRequest")
proto.RegisterType((*PostEventRequest)(nil), "containerd.services.events.v1.PostEventRequest") proto.RegisterType((*PublishRequest)(nil), "containerd.services.events.v1.PublishRequest")
proto.RegisterType((*Envelope)(nil), "containerd.services.events.v1.Envelope") proto.RegisterType((*Envelope)(nil), "containerd.services.events.v1.Envelope")
} }
@ -74,8 +76,8 @@ const _ = grpc.SupportPackageIsVersion4
// Client API for Events service // Client API for Events service
type EventsClient interface { type EventsClient interface {
Stream(ctx context.Context, in *StreamEventsRequest, opts ...grpc.CallOption) (Events_StreamClient, error) Publish(ctx context.Context, in *PublishRequest, opts ...grpc.CallOption) (*google_protobuf2.Empty, error)
Post(ctx context.Context, in *PostEventRequest, opts ...grpc.CallOption) (*google_protobuf2.Empty, error) Subscribe(ctx context.Context, in *SubscribeRequest, opts ...grpc.CallOption) (Events_SubscribeClient, error)
} }
type eventsClient struct { type eventsClient struct {
@ -86,12 +88,21 @@ func NewEventsClient(cc *grpc.ClientConn) EventsClient {
return &eventsClient{cc} return &eventsClient{cc}
} }
func (c *eventsClient) Stream(ctx context.Context, in *StreamEventsRequest, opts ...grpc.CallOption) (Events_StreamClient, error) { func (c *eventsClient) Publish(ctx context.Context, in *PublishRequest, opts ...grpc.CallOption) (*google_protobuf2.Empty, error) {
stream, err := grpc.NewClientStream(ctx, &_Events_serviceDesc.Streams[0], c.cc, "/containerd.services.events.v1.Events/Stream", opts...) out := new(google_protobuf2.Empty)
err := grpc.Invoke(ctx, "/containerd.services.events.v1.Events/Publish", in, out, c.cc, opts...)
if err != nil { if err != nil {
return nil, err return nil, err
} }
x := &eventsStreamClient{stream} return out, nil
}
func (c *eventsClient) Subscribe(ctx context.Context, in *SubscribeRequest, opts ...grpc.CallOption) (Events_SubscribeClient, error) {
stream, err := grpc.NewClientStream(ctx, &_Events_serviceDesc.Streams[0], c.cc, "/containerd.services.events.v1.Events/Subscribe", opts...)
if err != nil {
return nil, err
}
x := &eventsSubscribeClient{stream}
if err := x.ClientStream.SendMsg(in); err != nil { if err := x.ClientStream.SendMsg(in); err != nil {
return nil, err return nil, err
} }
@ -101,16 +112,16 @@ func (c *eventsClient) Stream(ctx context.Context, in *StreamEventsRequest, opts
return x, nil return x, nil
} }
type Events_StreamClient interface { type Events_SubscribeClient interface {
Recv() (*Envelope, error) Recv() (*Envelope, error)
grpc.ClientStream grpc.ClientStream
} }
type eventsStreamClient struct { type eventsSubscribeClient struct {
grpc.ClientStream grpc.ClientStream
} }
func (x *eventsStreamClient) Recv() (*Envelope, error) { func (x *eventsSubscribeClient) Recv() (*Envelope, error) {
m := new(Envelope) m := new(Envelope)
if err := x.ClientStream.RecvMsg(m); err != nil { if err := x.ClientStream.RecvMsg(m); err != nil {
return nil, err return nil, err
@ -118,85 +129,76 @@ func (x *eventsStreamClient) Recv() (*Envelope, error) {
return m, nil return m, nil
} }
func (c *eventsClient) Post(ctx context.Context, in *PostEventRequest, opts ...grpc.CallOption) (*google_protobuf2.Empty, error) {
out := new(google_protobuf2.Empty)
err := grpc.Invoke(ctx, "/containerd.services.events.v1.Events/Post", in, out, c.cc, opts...)
if err != nil {
return nil, err
}
return out, nil
}
// Server API for Events service // Server API for Events service
type EventsServer interface { type EventsServer interface {
Stream(*StreamEventsRequest, Events_StreamServer) error Publish(context.Context, *PublishRequest) (*google_protobuf2.Empty, error)
Post(context.Context, *PostEventRequest) (*google_protobuf2.Empty, error) Subscribe(*SubscribeRequest, Events_SubscribeServer) error
} }
func RegisterEventsServer(s *grpc.Server, srv EventsServer) { func RegisterEventsServer(s *grpc.Server, srv EventsServer) {
s.RegisterService(&_Events_serviceDesc, srv) s.RegisterService(&_Events_serviceDesc, srv)
} }
func _Events_Stream_Handler(srv interface{}, stream grpc.ServerStream) error { func _Events_Publish_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
m := new(StreamEventsRequest) in := new(PublishRequest)
if err := stream.RecvMsg(m); err != nil {
return err
}
return srv.(EventsServer).Stream(m, &eventsStreamServer{stream})
}
type Events_StreamServer interface {
Send(*Envelope) error
grpc.ServerStream
}
type eventsStreamServer struct {
grpc.ServerStream
}
func (x *eventsStreamServer) Send(m *Envelope) error {
return x.ServerStream.SendMsg(m)
}
func _Events_Post_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(PostEventRequest)
if err := dec(in); err != nil { if err := dec(in); err != nil {
return nil, err return nil, err
} }
if interceptor == nil { if interceptor == nil {
return srv.(EventsServer).Post(ctx, in) return srv.(EventsServer).Publish(ctx, in)
} }
info := &grpc.UnaryServerInfo{ info := &grpc.UnaryServerInfo{
Server: srv, Server: srv,
FullMethod: "/containerd.services.events.v1.Events/Post", FullMethod: "/containerd.services.events.v1.Events/Publish",
} }
handler := func(ctx context.Context, req interface{}) (interface{}, error) { handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(EventsServer).Post(ctx, req.(*PostEventRequest)) return srv.(EventsServer).Publish(ctx, req.(*PublishRequest))
} }
return interceptor(ctx, in, info, handler) return interceptor(ctx, in, info, handler)
} }
func _Events_Subscribe_Handler(srv interface{}, stream grpc.ServerStream) error {
m := new(SubscribeRequest)
if err := stream.RecvMsg(m); err != nil {
return err
}
return srv.(EventsServer).Subscribe(m, &eventsSubscribeServer{stream})
}
type Events_SubscribeServer interface {
Send(*Envelope) error
grpc.ServerStream
}
type eventsSubscribeServer struct {
grpc.ServerStream
}
func (x *eventsSubscribeServer) Send(m *Envelope) error {
return x.ServerStream.SendMsg(m)
}
var _Events_serviceDesc = grpc.ServiceDesc{ var _Events_serviceDesc = grpc.ServiceDesc{
ServiceName: "containerd.services.events.v1.Events", ServiceName: "containerd.services.events.v1.Events",
HandlerType: (*EventsServer)(nil), HandlerType: (*EventsServer)(nil),
Methods: []grpc.MethodDesc{ Methods: []grpc.MethodDesc{
{ {
MethodName: "Post", MethodName: "Publish",
Handler: _Events_Post_Handler, Handler: _Events_Publish_Handler,
}, },
}, },
Streams: []grpc.StreamDesc{ Streams: []grpc.StreamDesc{
{ {
StreamName: "Stream", StreamName: "Subscribe",
Handler: _Events_Stream_Handler, Handler: _Events_Subscribe_Handler,
ServerStreams: true, ServerStreams: true,
}, },
}, },
Metadata: "github.com/containerd/containerd/api/services/events/v1/events.proto", Metadata: "github.com/containerd/containerd/api/services/events/v1/events.proto",
} }
func (m *StreamEventsRequest) Marshal() (dAtA []byte, err error) { func (m *SubscribeRequest) Marshal() (dAtA []byte, err error) {
size := m.Size() size := m.Size()
dAtA = make([]byte, size) dAtA = make([]byte, size)
n, err := m.MarshalTo(dAtA) n, err := m.MarshalTo(dAtA)
@ -206,15 +208,30 @@ func (m *StreamEventsRequest) Marshal() (dAtA []byte, err error) {
return dAtA[:n], nil return dAtA[:n], nil
} }
func (m *StreamEventsRequest) MarshalTo(dAtA []byte) (int, error) { func (m *SubscribeRequest) MarshalTo(dAtA []byte) (int, error) {
var i int var i int
_ = i _ = i
var l int var l int
_ = l _ = l
if len(m.Filters) > 0 {
for _, s := range m.Filters {
dAtA[i] = 0xa
i++
l = len(s)
for l >= 1<<7 {
dAtA[i] = uint8(uint64(l)&0x7f | 0x80)
l >>= 7
i++
}
dAtA[i] = uint8(l)
i++
i += copy(dAtA[i:], s)
}
}
return i, nil return i, nil
} }
func (m *PostEventRequest) Marshal() (dAtA []byte, err error) { func (m *PublishRequest) Marshal() (dAtA []byte, err error) {
size := m.Size() size := m.Size()
dAtA = make([]byte, size) dAtA = make([]byte, size)
n, err := m.MarshalTo(dAtA) n, err := m.MarshalTo(dAtA)
@ -224,7 +241,7 @@ func (m *PostEventRequest) Marshal() (dAtA []byte, err error) {
return dAtA[:n], nil return dAtA[:n], nil
} }
func (m *PostEventRequest) MarshalTo(dAtA []byte) (int, error) { func (m *PublishRequest) MarshalTo(dAtA []byte) (int, error) {
var i int var i int
_ = i _ = i
var l int var l int
@ -265,14 +282,20 @@ func (m *Envelope) MarshalTo(dAtA []byte) (int, error) {
return 0, err return 0, err
} }
i += n2 i += n2
if len(m.Topic) > 0 { if len(m.Namespace) > 0 {
dAtA[i] = 0x12 dAtA[i] = 0x12
i++ i++
i = encodeVarintEvents(dAtA, i, uint64(len(m.Namespace)))
i += copy(dAtA[i:], m.Namespace)
}
if len(m.Topic) > 0 {
dAtA[i] = 0x1a
i++
i = encodeVarintEvents(dAtA, i, uint64(len(m.Topic))) i = encodeVarintEvents(dAtA, i, uint64(len(m.Topic)))
i += copy(dAtA[i:], m.Topic) i += copy(dAtA[i:], m.Topic)
} }
if m.Event != nil { if m.Event != nil {
dAtA[i] = 0x1a dAtA[i] = 0x22
i++ i++
i = encodeVarintEvents(dAtA, i, uint64(m.Event.Size())) i = encodeVarintEvents(dAtA, i, uint64(m.Event.Size()))
n3, err := m.Event.MarshalTo(dAtA[i:]) n3, err := m.Event.MarshalTo(dAtA[i:])
@ -311,13 +334,19 @@ func encodeVarintEvents(dAtA []byte, offset int, v uint64) int {
dAtA[offset] = uint8(v) dAtA[offset] = uint8(v)
return offset + 1 return offset + 1
} }
func (m *StreamEventsRequest) Size() (n int) { func (m *SubscribeRequest) Size() (n int) {
var l int var l int
_ = l _ = l
if len(m.Filters) > 0 {
for _, s := range m.Filters {
l = len(s)
n += 1 + l + sovEvents(uint64(l))
}
}
return n return n
} }
func (m *PostEventRequest) Size() (n int) { func (m *PublishRequest) Size() (n int) {
var l int var l int
_ = l _ = l
if m.Envelope != nil { if m.Envelope != nil {
@ -332,6 +361,10 @@ func (m *Envelope) Size() (n int) {
_ = l _ = l
l = github_com_gogo_protobuf_types.SizeOfStdTime(m.Timestamp) l = github_com_gogo_protobuf_types.SizeOfStdTime(m.Timestamp)
n += 1 + l + sovEvents(uint64(l)) n += 1 + l + sovEvents(uint64(l))
l = len(m.Namespace)
if l > 0 {
n += 1 + l + sovEvents(uint64(l))
}
l = len(m.Topic) l = len(m.Topic)
if l > 0 { if l > 0 {
n += 1 + l + sovEvents(uint64(l)) n += 1 + l + sovEvents(uint64(l))
@ -356,20 +389,21 @@ func sovEvents(x uint64) (n int) {
func sozEvents(x uint64) (n int) { func sozEvents(x uint64) (n int) {
return sovEvents(uint64((x << 1) ^ uint64((int64(x) >> 63)))) return sovEvents(uint64((x << 1) ^ uint64((int64(x) >> 63))))
} }
func (this *StreamEventsRequest) String() string { func (this *SubscribeRequest) String() string {
if this == nil { if this == nil {
return "nil" return "nil"
} }
s := strings.Join([]string{`&StreamEventsRequest{`, s := strings.Join([]string{`&SubscribeRequest{`,
`Filters:` + fmt.Sprintf("%v", this.Filters) + `,`,
`}`, `}`,
}, "") }, "")
return s return s
} }
func (this *PostEventRequest) String() string { func (this *PublishRequest) String() string {
if this == nil { if this == nil {
return "nil" return "nil"
} }
s := strings.Join([]string{`&PostEventRequest{`, s := strings.Join([]string{`&PublishRequest{`,
`Envelope:` + strings.Replace(fmt.Sprintf("%v", this.Envelope), "Envelope", "Envelope", 1) + `,`, `Envelope:` + strings.Replace(fmt.Sprintf("%v", this.Envelope), "Envelope", "Envelope", 1) + `,`,
`}`, `}`,
}, "") }, "")
@ -381,6 +415,7 @@ func (this *Envelope) String() string {
} }
s := strings.Join([]string{`&Envelope{`, s := strings.Join([]string{`&Envelope{`,
`Timestamp:` + strings.Replace(strings.Replace(this.Timestamp.String(), "Timestamp", "google_protobuf3.Timestamp", 1), `&`, ``, 1) + `,`, `Timestamp:` + strings.Replace(strings.Replace(this.Timestamp.String(), "Timestamp", "google_protobuf3.Timestamp", 1), `&`, ``, 1) + `,`,
`Namespace:` + fmt.Sprintf("%v", this.Namespace) + `,`,
`Topic:` + fmt.Sprintf("%v", this.Topic) + `,`, `Topic:` + fmt.Sprintf("%v", this.Topic) + `,`,
`Event:` + strings.Replace(fmt.Sprintf("%v", this.Event), "Any", "google_protobuf1.Any", 1) + `,`, `Event:` + strings.Replace(fmt.Sprintf("%v", this.Event), "Any", "google_protobuf1.Any", 1) + `,`,
`}`, `}`,
@ -395,7 +430,7 @@ func valueToStringEvents(v interface{}) string {
pv := reflect.Indirect(rv).Interface() pv := reflect.Indirect(rv).Interface()
return fmt.Sprintf("*%v", pv) return fmt.Sprintf("*%v", pv)
} }
func (m *StreamEventsRequest) Unmarshal(dAtA []byte) error { func (m *SubscribeRequest) Unmarshal(dAtA []byte) error {
l := len(dAtA) l := len(dAtA)
iNdEx := 0 iNdEx := 0
for iNdEx < l { for iNdEx < l {
@ -418,12 +453,41 @@ func (m *StreamEventsRequest) Unmarshal(dAtA []byte) error {
fieldNum := int32(wire >> 3) fieldNum := int32(wire >> 3)
wireType := int(wire & 0x7) wireType := int(wire & 0x7)
if wireType == 4 { if wireType == 4 {
return fmt.Errorf("proto: StreamEventsRequest: wiretype end group for non-group") return fmt.Errorf("proto: SubscribeRequest: wiretype end group for non-group")
} }
if fieldNum <= 0 { if fieldNum <= 0 {
return fmt.Errorf("proto: StreamEventsRequest: illegal tag %d (wire type %d)", fieldNum, wire) return fmt.Errorf("proto: SubscribeRequest: illegal tag %d (wire type %d)", fieldNum, wire)
} }
switch fieldNum { switch fieldNum {
case 1:
if wireType != 2 {
return fmt.Errorf("proto: wrong wireType = %d for field Filters", wireType)
}
var stringLen uint64
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowEvents
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
stringLen |= (uint64(b) & 0x7F) << shift
if b < 0x80 {
break
}
}
intStringLen := int(stringLen)
if intStringLen < 0 {
return ErrInvalidLengthEvents
}
postIndex := iNdEx + intStringLen
if postIndex > l {
return io.ErrUnexpectedEOF
}
m.Filters = append(m.Filters, string(dAtA[iNdEx:postIndex]))
iNdEx = postIndex
default: default:
iNdEx = preIndex iNdEx = preIndex
skippy, err := skipEvents(dAtA[iNdEx:]) skippy, err := skipEvents(dAtA[iNdEx:])
@ -445,7 +509,7 @@ func (m *StreamEventsRequest) Unmarshal(dAtA []byte) error {
} }
return nil return nil
} }
func (m *PostEventRequest) Unmarshal(dAtA []byte) error { func (m *PublishRequest) Unmarshal(dAtA []byte) error {
l := len(dAtA) l := len(dAtA)
iNdEx := 0 iNdEx := 0
for iNdEx < l { for iNdEx < l {
@ -468,10 +532,10 @@ func (m *PostEventRequest) Unmarshal(dAtA []byte) error {
fieldNum := int32(wire >> 3) fieldNum := int32(wire >> 3)
wireType := int(wire & 0x7) wireType := int(wire & 0x7)
if wireType == 4 { if wireType == 4 {
return fmt.Errorf("proto: PostEventRequest: wiretype end group for non-group") return fmt.Errorf("proto: PublishRequest: wiretype end group for non-group")
} }
if fieldNum <= 0 { if fieldNum <= 0 {
return fmt.Errorf("proto: PostEventRequest: illegal tag %d (wire type %d)", fieldNum, wire) return fmt.Errorf("proto: PublishRequest: illegal tag %d (wire type %d)", fieldNum, wire)
} }
switch fieldNum { switch fieldNum {
case 1: case 1:
@ -588,6 +652,35 @@ func (m *Envelope) Unmarshal(dAtA []byte) error {
} }
iNdEx = postIndex iNdEx = postIndex
case 2: case 2:
if wireType != 2 {
return fmt.Errorf("proto: wrong wireType = %d for field Namespace", wireType)
}
var stringLen uint64
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowEvents
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
stringLen |= (uint64(b) & 0x7F) << shift
if b < 0x80 {
break
}
}
intStringLen := int(stringLen)
if intStringLen < 0 {
return ErrInvalidLengthEvents
}
postIndex := iNdEx + intStringLen
if postIndex > l {
return io.ErrUnexpectedEOF
}
m.Namespace = string(dAtA[iNdEx:postIndex])
iNdEx = postIndex
case 3:
if wireType != 2 { if wireType != 2 {
return fmt.Errorf("proto: wrong wireType = %d for field Topic", wireType) return fmt.Errorf("proto: wrong wireType = %d for field Topic", wireType)
} }
@ -616,7 +709,7 @@ func (m *Envelope) Unmarshal(dAtA []byte) error {
} }
m.Topic = string(dAtA[iNdEx:postIndex]) m.Topic = string(dAtA[iNdEx:postIndex])
iNdEx = postIndex iNdEx = postIndex
case 3: case 4:
if wireType != 2 { if wireType != 2 {
return fmt.Errorf("proto: wrong wireType = %d for field Event", wireType) return fmt.Errorf("proto: wrong wireType = %d for field Event", wireType)
} }
@ -780,28 +873,31 @@ func init() {
} }
var fileDescriptorEvents = []byte{ var fileDescriptorEvents = []byte{
// 367 bytes of a gzipped FileDescriptorProto // 407 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x9c, 0x92, 0xc1, 0x4e, 0xc2, 0x40, 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x9c, 0x92, 0xcd, 0x6e, 0xd3, 0x40,
0x10, 0x86, 0x59, 0x15, 0x02, 0xeb, 0xc5, 0xac, 0x68, 0xb0, 0xc6, 0x42, 0xb8, 0x48, 0x3c, 0xec, 0x10, 0xc7, 0xb3, 0x84, 0x7c, 0x78, 0x91, 0x10, 0x5a, 0x45, 0xc8, 0x18, 0x70, 0xa2, 0x5c, 0x88,
0x0a, 0x1e, 0x4d, 0x4c, 0x44, 0x39, 0x6b, 0xaa, 0x89, 0xc6, 0x5b, 0xa9, 0x63, 0x6d, 0x42, 0xbb, 0x10, 0xec, 0x92, 0x70, 0x44, 0x42, 0x22, 0x90, 0x7b, 0x64, 0x40, 0x42, 0xdc, 0x6c, 0x77, 0xe2,
0xb5, 0x5d, 0x9a, 0x70, 0xf3, 0x11, 0x78, 0x26, 0x4f, 0x1c, 0x3d, 0x7a, 0x52, 0xe9, 0x93, 0x18, 0xac, 0x64, 0x7b, 0x5d, 0xef, 0xda, 0x52, 0x6e, 0x7d, 0x84, 0x3e, 0x49, 0x5f, 0xa2, 0x97, 0x1c,
0x76, 0xb7, 0x60, 0xc0, 0x88, 0xf1, 0x36, 0x3b, 0xf3, 0xcd, 0xf4, 0x9f, 0x7f, 0x8a, 0xcf, 0x5d, 0x7b, 0xec, 0xa9, 0x6d, 0xfc, 0x24, 0x55, 0xfc, 0x91, 0xb4, 0x89, 0xd4, 0x54, 0xbd, 0xcd, 0xec,
0x4f, 0x3c, 0xf6, 0xbb, 0xd4, 0xe1, 0x3e, 0x73, 0x78, 0x20, 0x6c, 0x2f, 0x80, 0xe8, 0xfe, 0x7b, 0xff, 0x37, 0x3b, 0xfb, 0x9f, 0x59, 0xfc, 0xcb, 0xe3, 0x6a, 0x9e, 0x38, 0xd4, 0x15, 0x01, 0x73,
0x68, 0x87, 0x1e, 0x8b, 0x21, 0x4a, 0x3c, 0x07, 0x62, 0x06, 0x09, 0x04, 0x22, 0x66, 0x49, 0x53, 0x45, 0xa8, 0x6c, 0x1e, 0x42, 0x7c, 0x74, 0x37, 0xb4, 0x23, 0xce, 0x24, 0xc4, 0x29, 0x77, 0x41,
0x47, 0x34, 0x8c, 0xb8, 0xe0, 0x64, 0x6f, 0xc6, 0xd3, 0x8c, 0xa5, 0x9a, 0x48, 0x9a, 0x46, 0xd9, 0x32, 0x48, 0x21, 0x54, 0x92, 0xa5, 0xc3, 0x32, 0xa2, 0x51, 0x2c, 0x94, 0x20, 0xef, 0xb7, 0x3c,
0xe5, 0x2e, 0x97, 0x24, 0x9b, 0x44, 0xaa, 0xc9, 0xd8, 0x71, 0x39, 0x77, 0x7b, 0xc0, 0xe4, 0xab, 0xad, 0x58, 0x5a, 0x12, 0xe9, 0xd0, 0xe8, 0x78, 0xc2, 0x13, 0x39, 0xc9, 0xd6, 0x51, 0x51, 0x64,
0xdb, 0x7f, 0x60, 0x76, 0x30, 0xd0, 0xa5, 0xdd, 0xf9, 0x12, 0xf8, 0xa1, 0xc8, 0x8a, 0xd5, 0xf9, 0xbc, 0xf1, 0x84, 0xf0, 0x7c, 0x60, 0x79, 0xe6, 0x24, 0x33, 0x66, 0x87, 0x8b, 0x52, 0x7a, 0xbb,
0xa2, 0xf0, 0x7c, 0x88, 0x85, 0xed, 0x87, 0x0a, 0xa8, 0x6f, 0xe1, 0xcd, 0x2b, 0x11, 0x81, 0xed, 0x2b, 0x41, 0x10, 0xa9, 0x4a, 0xec, 0xee, 0x8a, 0x8a, 0x07, 0x20, 0x95, 0x1d, 0x44, 0x05, 0xd0,
0x77, 0xa4, 0x02, 0x0b, 0x9e, 0xfa, 0x10, 0x8b, 0xfa, 0x0d, 0xde, 0xb8, 0xe4, 0xb1, 0x90, 0x49, 0xff, 0x84, 0x5f, 0xfd, 0x4e, 0x1c, 0xe9, 0xc6, 0xdc, 0x01, 0x0b, 0x8e, 0x13, 0x90, 0x8a, 0xe8,
0x9d, 0x23, 0x67, 0xb8, 0x08, 0x41, 0x02, 0x3d, 0x1e, 0x42, 0x05, 0xd5, 0x50, 0x63, 0xbd, 0xb5, 0xb8, 0x35, 0xe3, 0xbe, 0x82, 0x58, 0xea, 0xa8, 0x57, 0x1f, 0x68, 0x56, 0x95, 0xf6, 0xff, 0xe2,
0x4f, 0x7f, 0xdd, 0x85, 0x76, 0x34, 0x6e, 0x4d, 0x1b, 0xeb, 0x43, 0x84, 0x8b, 0x59, 0x9a, 0xb4, 0x97, 0xd3, 0xc4, 0xf1, 0xb9, 0x9c, 0x57, 0xec, 0x4f, 0xdc, 0x86, 0x30, 0x05, 0x5f, 0x44, 0xa0,
0x71, 0x69, 0xaa, 0x47, 0x8f, 0x34, 0xa8, 0x52, 0x4c, 0x33, 0xc5, 0xf4, 0x3a, 0x23, 0xda, 0xc5, 0xa3, 0x1e, 0x1a, 0xbc, 0x18, 0x7d, 0xa0, 0x0f, 0x1a, 0xa4, 0x93, 0x12, 0xb7, 0x36, 0x85, 0xfd,
0xd1, 0x7b, 0x35, 0x37, 0xfc, 0xa8, 0x22, 0x6b, 0xd6, 0x46, 0xca, 0x38, 0x2f, 0x78, 0xe8, 0x39, 0x33, 0x84, 0xdb, 0xd5, 0x31, 0x19, 0x63, 0x6d, 0xf3, 0xc8, 0xf2, 0x4a, 0x83, 0x16, 0x36, 0x68,
0x95, 0x95, 0x1a, 0x6a, 0x94, 0x2c, 0xf5, 0x20, 0x07, 0x38, 0x2f, 0x65, 0x54, 0x56, 0xe5, 0xd4, 0x65, 0x83, 0xfe, 0xa9, 0x88, 0x71, 0x7b, 0x79, 0xd5, 0xad, 0x9d, 0x5e, 0x77, 0x91, 0xb5, 0x2d,
0xf2, 0xc2, 0xd4, 0xd3, 0x60, 0x60, 0x29, 0xa4, 0xf5, 0x82, 0x70, 0x41, 0x6d, 0x4f, 0x5c, 0x5c, 0x23, 0xef, 0xb0, 0x16, 0xda, 0x01, 0xc8, 0xc8, 0x76, 0x41, 0x7f, 0xd6, 0x43, 0x03, 0xcd, 0xda,
0x50, 0x6e, 0x90, 0xd6, 0x92, 0xd5, 0x7e, 0x30, 0xcd, 0xf8, 0xab, 0x1d, 0x87, 0x88, 0x5c, 0xe0, 0x1e, 0x90, 0x0e, 0x6e, 0x28, 0x11, 0x71, 0x57, 0xaf, 0xe7, 0x4a, 0x91, 0x90, 0x8f, 0xb8, 0x91,
0xb5, 0x89, 0xbf, 0x84, 0x2d, 0x69, 0x99, 0x3f, 0x82, 0xb1, 0xbd, 0xb0, 0x49, 0x67, 0x72, 0xee, 0x3f, 0x52, 0x7f, 0x9e, 0xf7, 0xec, 0xec, 0xf5, 0xfc, 0x11, 0x2e, 0xac, 0x02, 0x19, 0x9d, 0x23,
0xf6, 0xed, 0x68, 0x6c, 0xe6, 0xde, 0xc6, 0x66, 0xee, 0x39, 0x35, 0xd1, 0x28, 0x35, 0xd1, 0x6b, 0xdc, 0x9c, 0xe4, 0x8e, 0xc8, 0x14, 0xb7, 0xca, 0x91, 0x90, 0xcf, 0x07, 0x9c, 0xdf, 0x1f, 0x9d,
0x6a, 0xa2, 0xcf, 0xd4, 0x44, 0x77, 0x27, 0xff, 0xfc, 0x6b, 0x8f, 0x55, 0xd4, 0x2d, 0xc8, 0x2f, 0xf1, 0x7a, 0xaf, 0xc3, 0x64, 0xbd, 0x39, 0xe2, 0x61, 0x6d, 0xb3, 0x12, 0xc2, 0x0e, 0xdc, 0xb9,
0x1d, 0x7d, 0x05, 0x00, 0x00, 0xff, 0xff, 0xdd, 0x37, 0xcb, 0x0e, 0xfe, 0x02, 0x00, 0x00, 0xbb, 0x3c, 0xe3, 0xb1, 0xe3, 0xff, 0x82, 0xc6, 0xff, 0x96, 0x2b, 0xb3, 0x76, 0xb9, 0x32, 0x6b,
0x27, 0x99, 0x89, 0x96, 0x99, 0x89, 0x2e, 0x32, 0x13, 0xdd, 0x64, 0x26, 0xfa, 0xff, 0xfd, 0x89,
0x3f, 0xfd, 0x5b, 0x11, 0x39, 0xcd, 0xdc, 0xd2, 0xd7, 0xdb, 0x00, 0x00, 0x00, 0xff, 0xff, 0x13,
0x35, 0xd0, 0x60, 0x32, 0x03, 0x00, 0x00,
} }

15
api/services/events/v1/events.proto
View File

@ -10,18 +10,21 @@ import "google/protobuf/timestamp.proto";
option go_package = "github.com/containerd/containerd/api/services/events/v1;events"; option go_package = "github.com/containerd/containerd/api/services/events/v1;events";
service Events { service Events {
rpc Stream(StreamEventsRequest) returns (stream Envelope); rpc Publish(PublishRequest) returns (google.protobuf.Empty);
rpc Post(PostEventRequest) returns (google.protobuf.Empty); rpc Subscribe(SubscribeRequest) returns (stream Envelope);
} }
message StreamEventsRequest {} message SubscribeRequest {
repeated string filters = 1;
}
message PostEventRequest { message PublishRequest {
Envelope envelope = 1; Envelope envelope = 1;
} }
message Envelope { message Envelope {
google.protobuf.Timestamp timestamp = 1 [(gogoproto.stdtime) = true, (gogoproto.nullable) = false]; google.protobuf.Timestamp timestamp = 1 [(gogoproto.stdtime) = true, (gogoproto.nullable) = false];
string topic = 2; string namespace = 2;
google.protobuf.Any event = 3; string topic = 3;
google.protobuf.Any event = 4;
} }

2
cmd/ctr/events.go
View File

@ -22,7 +22,7 @@ var eventsCommand = cli.Command{
ctx, cancel := appContext(context) ctx, cancel := appContext(context)
defer cancel() defer cancel()
events, err := eventsClient.Stream(ctx, &eventsapi.StreamEventsRequest{}) events, err := eventsClient.Subscribe(ctx, &eventsapi.SubscribeRequest{})
if err != nil { if err != nil {
return err return err
} }

67
events/emitter.go
View File

@ -1,67 +0,0 @@
package events
import (
"context"
"sync"
events "github.com/containerd/containerd/api/services/events/v1"
"github.com/containerd/containerd/namespaces"
goevents "github.com/docker/go-events"
)
const (
EventVersion = "v1"
)
type Emitter struct {
sinks map[string]*eventSink
broadcaster *goevents.Broadcaster
m sync.Mutex
}
func NewEmitter() *Emitter {
return &Emitter{
sinks: make(map[string]*eventSink),
broadcaster: goevents.NewBroadcaster(),
m: sync.Mutex{},
}
}
func (e *Emitter) Post(ctx context.Context, evt Event) error {
if err := e.broadcaster.Write(&sinkEvent{
ctx: ctx,
event: evt,
}); err != nil {
return err
}
return nil
}
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 *events.Envelope),
ns: ns,
}
e.sinks[clientID] = s
e.m.Unlock()
e.broadcaster.Add(s)
return s.ch
}
ch := e.sinks[clientID].ch
e.m.Unlock()
return ch
}
func (e *Emitter) Remove(clientID string) {
e.m.Lock()
if v, ok := e.sinks[clientID]; ok {
e.broadcaster.Remove(v)
delete(e.sinks, clientID)
}
e.m.Unlock()
}

3
events/event.go
View File

@ -1,3 +0,0 @@
package events
type Event interface{}

24
events/events.go Normal file
View File

@ -0,0 +1,24 @@
package events
import (
"context"
events "github.com/containerd/containerd/api/services/events/v1"
)
type Event interface{}
// Publisher posts the event.
type Publisher interface {
Publish(ctx context.Context, topic string, event Event) error
}
type Forwarder interface {
Forward(ctx context.Context, envelope *events.Envelope) error
}
type publisherFunc func(ctx context.Context, topic string, event Event) error
func (fn publisherFunc) Publish(ctx context.Context, topic string, event Event) error {
return fn(ctx, topic, event)
}

33
events/events_test.go
View File

@ -1,33 +0,0 @@
package events
import (
"context"
"fmt"
"testing"
)
func TestBasicEvent(t *testing.T) {
ctx := context.Background()
// simulate a layer pull with events
ctx, commit, _ := WithTx(ctx)
G(ctx).Post(ctx, "pull ubuntu")
for layer := 0; layer < 4; layer++ {
// make a subtransaction for each layer
ctx, commit, _ := WithTx(ctx)
G(ctx).Post(ctx, fmt.Sprintf("fetch layer %v", layer))
ctx = WithTopic(ctx, "content")
// simulate sub-operations with a separate topic, on the content store
G(ctx).Post(ctx, fmt.Sprint("received sha:256"))
G(ctx).Post(ctx, fmt.Sprintf("unpack layer %v", layer))
commit()
}
commit()
}

162
events/exchange.go Normal file
View File

@ -0,0 +1,162 @@
package events
import (
"context"
"strings"
"time"
events "github.com/containerd/containerd/api/services/events/v1"
"github.com/containerd/containerd/errdefs"
"github.com/containerd/containerd/filters"
"github.com/containerd/containerd/identifiers"
"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"
"github.com/sirupsen/logrus"
)
type Exchange struct {
broadcaster *goevents.Broadcaster
}
func NewExchange() *Exchange {
return &Exchange{
broadcaster: goevents.NewBroadcaster(),
}
}
// Forward accepts an envelope to be direcly distributed on the exchange.
func (e *Exchange) Forward(ctx context.Context, envelope *events.Envelope) error {
log.G(ctx).WithFields(logrus.Fields{
"topic": envelope.Topic,
"ns": envelope.Namespace,
"type": envelope.Event.TypeUrl,
}).Debug("forward event")
if err := namespaces.Validate(envelope.Namespace); err != nil {
return errors.Wrapf(err, "event envelope has invalid namespace")
}
if err := validateTopic(envelope.Topic); err != nil {
return errors.Wrapf(err, "envelope topic %q", envelope.Topic)
}
return e.broadcaster.Write(envelope)
}
// Publish packages and sends an event. The caller will be considered the
// initial publisher of the event. This means the timestamp will be calculated
// at this point and this method may read from the calling context.
func (e *Exchange) Publish(ctx context.Context, topic string, event Event) error {
namespace, err := namespaces.NamespaceRequired(ctx)
if err != nil {
return errors.Wrapf(err, "failed publishing event")
}
if err := validateTopic(topic); err != nil {
return errors.Wrapf(err, "envelope topic %q", topic)
}
evany, err := typeurl.MarshalAny(event)
if err != nil {
return err
}
env := events.Envelope{
Timestamp: time.Now().UTC(),
Topic: topic,
Event: evany,
}
if err := e.broadcaster.Write(&env); err != nil {
return err
}
log.G(ctx).WithFields(logrus.Fields{
"topic": topic,
"type": evany.TypeUrl,
"ns": namespace,
}).Debug("published event")
return nil
}
// Subscribe to events on the exchange. Events are sent through the returned
// channel ch. If an error is encountered, it will be sent on channel errs and
// errs will be closed. To end the subscription, cancel the provided context.
func (e *Exchange) Subscribe(ctx context.Context, filters ...filters.Filter) (ch <-chan *events.Envelope, errs <-chan error) {
var (
evch = make(chan *events.Envelope)
errq = make(chan error, 1)
channel = goevents.NewChannel(0)
queue = goevents.NewQueue(channel)
)
// TODO(stevvooe): Insert the filter!
e.broadcaster.Add(queue)
go func() {
defer close(errq)
defer e.broadcaster.Remove(queue)
defer queue.Close()
var err error
loop:
for {
select {
case ev := <-channel.C:
env, ok := ev.(*events.Envelope)
if !ok {
// TODO(stevvooe): For the most part, we are well protected
// from this condition. Both Forward and Publish protect
// from this.
err = errors.Errorf("invalid envelope encountered %#v; please file a bug", ev)
break
}
select {
case evch <- env:
case <-ctx.Done():
break loop
}
case <-ctx.Done():
break loop
}
}
if err == nil {
if cerr := ctx.Err(); cerr != context.Canceled {
err = cerr
}
}
errq <- err
}()
ch = evch
errs = errq
return
}
func validateTopic(topic string) error {
if topic == "" {
return errors.Wrap(errdefs.ErrInvalidArgument, "must not be empty")
}
if topic[0] != '/' {
return errors.Wrapf(errdefs.ErrInvalidArgument, "must start with '/'", topic)
}
if len(topic) == 1 {
return errors.Wrapf(errdefs.ErrInvalidArgument, "must have at least one component", topic)
}
components := strings.Split(topic[1:], "/")
for _, component := range components {
if err := identifiers.Validate(component); err != nil {
return errors.Wrapf(err, "failed validation on component %q", component)
}
}
return nil
}

149
events/exchange_test.go Normal file
View File

@ -0,0 +1,149 @@
package events
import (
"context"
"fmt"
"reflect"
"sync"
"testing"
events "github.com/containerd/containerd/api/services/events/v1"
"github.com/containerd/containerd/errdefs"
"github.com/containerd/containerd/namespaces"
"github.com/containerd/containerd/typeurl"
"github.com/pkg/errors"
)
func TestExchangeBasic(t *testing.T) {
ctx := namespaces.WithNamespace(context.Background(), t.Name())
testevents := []Event{
&events.ContainerCreate{ID: "asdf"},
&events.ContainerCreate{ID: "qwer"},
&events.ContainerCreate{ID: "zxcv"},
}
exchange := NewExchange()
t.Log("subscribe")
var cancel1, cancel2 func()
// Create two subscribers for same set of events and make sure they
// traverse the exchange.
ctx1, cancel1 := context.WithCancel(ctx)
eventq1, errq1 := exchange.Subscribe(ctx1)
ctx2, cancel2 := context.WithCancel(ctx)
eventq2, errq2 := exchange.Subscribe(ctx2)
t.Log("publish")
var wg sync.WaitGroup
wg.Add(1)
go func() {
defer wg.Done()
for _, event := range testevents {
fmt.Println("publish", event)
if err := exchange.Publish(ctx, "/test", event); err != nil {
fmt.Println("publish error", err)
t.Fatal(err)
}
}
t.Log("finished publishing")
}()
t.Log("waiting")
wg.Wait()
for _, subscriber := range []struct {
eventq <-chan *events.Envelope
errq <-chan error
cancel func()
}{
{
eventq: eventq1,
errq: errq1,
cancel: cancel1,
},
{
eventq: eventq2,
errq: errq2,
cancel: cancel2,
},
} {
var received []Event
subscribercheck:
for {
select {
case env := <-subscriber.eventq:
ev, err := typeurl.UnmarshalAny(env.Event)
if err != nil {
t.Fatal(err)
}
received = append(received, ev.(*events.ContainerCreate))
case err := <-subscriber.errq:
if err != nil {
t.Fatal(err)
}
break subscribercheck
}
if reflect.DeepEqual(received, testevents) {
// when we do this, we expect the errs channel to be closed and
// this will return.
subscriber.cancel()
}
}
}
}
func TestExchangeValidateTopic(t *testing.T) {
namespace := t.Name()
ctx := namespaces.WithNamespace(context.Background(), namespace)
exchange := NewExchange()
for _, testcase := range []struct {
input string
err error
}{
{
input: "/test",
},
{
input: "/test/test",
},
{
input: "test",
err: errdefs.ErrInvalidArgument,
},
} {
t.Run(testcase.input, func(t *testing.T) {
event := &events.ContainerCreate{ID: t.Name()}
if err := exchange.Publish(ctx, testcase.input, event); errors.Cause(err) != testcase.err {
if err == nil {
t.Fatalf("expected error %v, received nil", testcase.err)
} else {
t.Fatalf("expected error %v, received %v", testcase.err, err)
}
}
evany, err := typeurl.MarshalAny(event)
if err != nil {
t.Fatal(err)
}
envelope := events.Envelope{
Namespace: namespace,
Topic: testcase.input,
Event: evany,
}
// make sure we get same errors with forward.
if err := exchange.Forward(ctx, &envelope); errors.Cause(err) != testcase.err {
if err == nil {
t.Fatalf("expected error %v, received nil", testcase.err)
} else {
t.Fatalf("expected error %v, received %v", testcase.err, err)
}
}
})
}
}

65
events/poster.go
View File

@ -1,65 +0,0 @@
package events
import (
"context"
"github.com/containerd/containerd/log"
"github.com/containerd/containerd/namespaces"
"github.com/sirupsen/logrus"
)
var (
G = GetPoster
)
// Poster posts the event.
type Poster interface {
Post(ctx context.Context, evt Event) error
}
type posterKey struct{}
func WithPoster(ctx context.Context, poster Poster) context.Context {
return context.WithValue(ctx, posterKey{}, poster)
}
func GetPoster(ctx context.Context) Poster {
poster := ctx.Value(posterKey{})
if poster == nil {
tx, _ := getTx(ctx)
topic := getTopic(ctx)
// likely means we don't have a configured event system. Just return
// the default poster, which merely logs events.
return posterFunc(func(ctx context.Context, evt Event) error {
fields := logrus.Fields{"event": evt}
if topic != "" {
fields["topic"] = topic
}
ns, _ := namespaces.Namespace(ctx)
fields["ns"] = ns
if tx != nil {
fields["tx.id"] = tx.id
if tx.parent != nil {
fields["tx.parent.id"] = tx.parent.id
}
}
log.G(ctx).WithFields(fields).Debug("event fired")
return nil
})
}
return poster.(Poster)
}
type posterFunc func(ctx context.Context, evt Event) error
func (fn posterFunc) Post(ctx context.Context, evt Event) error {
fn(ctx, evt)
return nil
}

65
events/sink.go
View File

@ -1,65 +0,0 @@
package events
import (
"context"
"time"
"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"
"github.com/sirupsen/logrus"
)
type sinkEvent struct {
ctx context.Context
event Event
}
type eventSink struct {
ns string
ch chan *events.Envelope
}
func (s *eventSink) Write(evt goevents.Event) error {
e, ok := evt.(*sinkEvent)
if !ok {
return errors.New("event is not a sink event")
}
ns, _ := namespaces.Namespace(e.ctx)
if ns != "" && ns != s.ns {
// ignore events not intended for this ns
return nil
}
if ev, ok := e.event.(*events.Envelope); ok {
s.ch <- ev
return nil
}
topic := getTopic(e.ctx)
eventData, err := typeurl.MarshalAny(e.event)
if err != nil {
return err
}
log.G(e.ctx).WithFields(logrus.Fields{
"topic": topic,
"type": eventData.TypeUrl,
"ns": ns,
}).Debug("event")
s.ch <- &events.Envelope{
Timestamp: time.Now(),
Topic: topic,
Event: eventData,
}
return nil
}
func (s *eventSink) Close() error {
return nil
}

35
events/topic.go
View File

@ -1,35 +0,0 @@
package events
import "context"
type topicKey struct{}
// WithTopic returns a context with a new topic set, such that events emitted
// from the resulting context will be marked with the topic.
//
// A topic groups events by the target module they operate on. This is
// primarily designed to support multi-module log compaction of events. In
// typical journaling systems, the entries operate on a single data structure.
// When compacting the journal, we can replace all former log entries with a
// summary data structure that will result in the same state.
//
// By providing a compaction mechanism by topic, we can prune down to a data
// structure oriented towards a single topic, leaving unrelated messages alone.
func WithTopic(ctx context.Context, topic string) context.Context {
return context.WithValue(ctx, topicKey{}, topic)
}
func getTopic(ctx context.Context) string {
topic := ctx.Value(topicKey{})
if topic == nil {
return ""
}
return topic.(string)
}
// RegisterCompactor sets the compacter for the given topic.
func RegisterCompactor(topic string, compactor interface{}) {
panic("not implemented")
}

96
events/transaction.go
View File

@ -1,96 +0,0 @@
package events
import (
"context"
"fmt"
"sync"
"sync/atomic"
"time"
)
var txCounter int64 // replace this with something that won't break
// nextTXID provides the next transaction identifier.
func nexttxID() int64 {
// TODO(stevvooe): Need to coordinate this with existing transaction logs.
// For now, this is a toy, but not a racy one.
return atomic.AddInt64(&txCounter, 1)
}
type transaction struct {
ctx context.Context
id int64
parent *transaction // if nil, no parent transaction
finish sync.Once
start, end time.Time // informational
}
// begin creates a sub-transaction.
func (tx *transaction) begin(ctx context.Context, poster Poster) *transaction {
id := nexttxID()
child := &transaction{
ctx: ctx,
id: id,
parent: tx,
start: time.Now(),
}
// post the transaction started event
poster.Post(ctx, child.makeTransactionEvent("begin")) // transactions are really just events
return child
}
// commit writes out the transaction.
func (tx *transaction) commit(poster Poster) {
tx.finish.Do(func() {
tx.end = time.Now()
poster.Post(tx.ctx, tx.makeTransactionEvent("commit"))
})
}
func (tx *transaction) rollback(poster Poster, cause error) {
tx.finish.Do(func() {
tx.end = time.Now()
event := tx.makeTransactionEvent("rollback")
event = fmt.Sprintf("%s error=%q", event, cause.Error())
poster.Post(tx.ctx, event)
})
}
func (tx *transaction) makeTransactionEvent(status string) Event {
// TODO(stevvooe): obviously, we need more structure than this.
event := fmt.Sprintf("%v %v", status, tx.id)
if tx.parent != nil {
event += " parent=" + fmt.Sprint(tx.parent.id)
}
return event
}
type txKey struct{}
func getTx(ctx context.Context) (*transaction, bool) {
tx := ctx.Value(txKey{})
if tx == nil {
return nil, false
}
return tx.(*transaction), true
}
// WithTx returns a new context with an event transaction, such that events
// posted to the underlying context will be committed to the event log as a
// group, organized by a transaction id, when commit is called.
func WithTx(pctx context.Context) (ctx context.Context, commit func(), rollback func(err error)) {
poster := G(pctx)
parent, _ := getTx(pctx)
tx := parent.begin(pctx, poster)
return context.WithValue(pctx, txKey{}, tx), func() {
tx.commit(poster)
}, func(err error) {
tx.rollback(poster, err)
}
}

11
linux/shim/local.go
View File

@ -89,17 +89,16 @@ func (c *local) Update(ctx context.Context, in *shimapi.UpdateTaskRequest, opts
return c.s.Update(ctx, in) return c.s.Update(ctx, in)
} }
type poster interface { type publisher interface {
Post(ctx context.Context, in *events.PostEventRequest, opts ...grpc.CallOption) (*google_protobuf.Empty, error) Publish(ctx context.Context, in *events.PublishRequest, opts ...grpc.CallOption) (*google_protobuf.Empty, error)
} }
type localEventsClient struct { type localEventsClient struct {
emitter evt.Poster forwarder evt.Forwarder
} }
func (l *localEventsClient) Post(ctx context.Context, r *events.PostEventRequest, opts ...grpc.CallOption) (*google_protobuf.Empty, error) { func (l *localEventsClient) Publish(ctx context.Context, r *events.PublishRequest, opts ...grpc.CallOption) (*google_protobuf.Empty, error) {
ctx = evt.WithTopic(ctx, r.Envelope.Topic) if err := l.forwarder.Forward(ctx, r.Envelope); err != nil {
if err := l.emitter.Post(ctx, r.Envelope); err != nil {
return nil, err return nil, err
} }
return empty, nil return empty, nil

12
linux/shim/service.go
View File

@ -37,7 +37,7 @@ func NewService(path, namespace, address string) (*Service, error) {
return nil, fmt.Errorf("shim namespace cannot be empty") return nil, fmt.Errorf("shim namespace cannot be empty")
} }
context := namespaces.WithNamespace(context.Background(), namespace) context := namespaces.WithNamespace(context.Background(), namespace)
var client poster var client publisher
if address != "" { if address != "" {
conn, err := connect(address, dialer) conn, err := connect(address, dialer)
if err != nil { if err != nil {
@ -46,7 +46,7 @@ func NewService(path, namespace, address string) (*Service, error) {
client = events.NewEventsClient(conn) client = events.NewEventsClient(conn)
} else { } else {
client = &localEventsClient{ client = &localEventsClient{
emitter: evt.GetPoster(context), forwarder: evt.NewExchange(),
} }
} }
s := &Service{ s := &Service{
@ -379,16 +379,18 @@ func (s *Service) getContainerPids(ctx context.Context, id string) ([]uint32, er
return pids, nil return pids, nil
} }
func (s *Service) forward(client poster) { func (s *Service) forward(client publisher) {
for e := range s.events { for e := range s.events {
a, err := typeurl.MarshalAny(e) a, err := typeurl.MarshalAny(e)
if err != nil { if err != nil {
log.G(s.context).WithError(err).Error("marshal event") log.G(s.context).WithError(err).Error("marshal event")
continue continue
} }
if _, err := client.Post(s.context, &events.PostEventRequest{
if _, err := client.Publish(s.context, &events.PublishRequest{
Envelope: &events.Envelope{ Envelope: &events.Envelope{
Timestamp: time.Now(), Namespace: s.namespace,
Timestamp: time.Now().UTC(),
Topic: getTopic(e), Topic: getTopic(e),
Event: a, Event: a,
}, },

10
metrics/cgroups/cgroups.go
View File

@ -4,8 +4,8 @@ package cgroups
import ( import (
"github.com/containerd/cgroups" "github.com/containerd/cgroups"
events "github.com/containerd/containerd/api/services/events/v1" eventsapi "github.com/containerd/containerd/api/services/events/v1"
evt "github.com/containerd/containerd/events" "github.com/containerd/containerd/events"
"github.com/containerd/containerd/log" "github.com/containerd/containerd/log"
"github.com/containerd/containerd/plugin" "github.com/containerd/containerd/plugin"
"github.com/containerd/containerd/runtime" "github.com/containerd/containerd/runtime"
@ -35,7 +35,7 @@ func New(ic *plugin.InitContext) (interface{}, error) {
collector: collector, collector: collector,
oom: oom, oom: oom,
context: ic.Context, context: ic.Context,
emitter: ic.Emitter, publisher: ic.Events,
}, nil }, nil
} }
@ -43,7 +43,7 @@ type cgroupsMonitor struct {
collector *Collector collector *Collector
oom *OOMCollector oom *OOMCollector
context context.Context context context.Context
emitter *evt.Emitter publisher events.Publisher
} }
func (m *cgroupsMonitor) Monitor(c runtime.Task) error { func (m *cgroupsMonitor) Monitor(c runtime.Task) error {
@ -69,7 +69,7 @@ func (m *cgroupsMonitor) Stop(c runtime.Task) error {
} }
func (m *cgroupsMonitor) trigger(id string, cg cgroups.Cgroup) { func (m *cgroupsMonitor) trigger(id string, cg cgroups.Cgroup) {
if err := m.emitter.Post(m.context, &events.TaskOOM{ if err := m.publisher.Publish(m.context, runtime.TaskOOMEventTopic, &eventsapi.TaskOOM{
ContainerID: id, ContainerID: id,
}); err != nil { }); err != nil {
log.G(m.context).WithError(err).Error("post OOM event") log.G(m.context).WithError(err).Error("post OOM event")

2
plugin/context.go
View File

@ -22,7 +22,7 @@ type InitContext struct {
Address string Address string
Context context.Context Context context.Context
Config interface{} Config interface{}
Emitter *events.Emitter Events *events.Exchange
plugins map[PluginType]map[string]interface{} plugins map[PluginType]map[string]interface{}
} }

2
process.go
View File

@ -64,7 +64,7 @@ func (p *process) Kill(ctx context.Context, s syscall.Signal) error {
} }
func (p *process) Wait(ctx context.Context) (uint32, error) { func (p *process) Wait(ctx context.Context) (uint32, error) {
eventstream, err := p.task.client.EventService().Stream(ctx, &eventsapi.StreamEventsRequest{}) eventstream, err := p.task.client.EventService().Subscribe(ctx, &eventsapi.SubscribeRequest{})
if err != nil { if err != nil {
return UnknownExitStatus, err return UnknownExitStatus, err
} }

8
server/server.go
View File

@ -54,7 +54,7 @@ func New(ctx context.Context, config *Config) (*Server, error) {
services []plugin.Service services []plugin.Service
s = &Server{ s = &Server{
rpc: rpc, rpc: rpc,
emitter: events.NewEmitter(), events: events.NewExchange(),
} }
initialized = make(map[plugin.PluginType]map[string]interface{}) initialized = make(map[plugin.PluginType]map[string]interface{})
) )
@ -63,12 +63,12 @@ func New(ctx context.Context, config *Config) (*Server, error) {
log.G(ctx).WithField("type", p.Type).Infof("loading plugin %q...", id) log.G(ctx).WithField("type", p.Type).Infof("loading plugin %q...", id)
initContext := plugin.NewContext( initContext := plugin.NewContext(
events.WithPoster(ctx, s.emitter), ctx,
initialized, initialized,
config.Root, config.Root,
id, id,
) )
initContext.Emitter = s.emitter initContext.Events = s.events
initContext.Address = config.GRPC.Address initContext.Address = config.GRPC.Address
// load the plugin specific configuration if it is provided // load the plugin specific configuration if it is provided
@ -113,7 +113,7 @@ func New(ctx context.Context, config *Config) (*Server, error) {
// Server is the containerd main daemon // Server is the containerd main daemon
type Server struct { type Server struct {
rpc *grpc.Server rpc *grpc.Server
emitter *events.Emitter events *events.Exchange
} }
// ServeGRPC provides the containerd grpc APIs on the provided listener // ServeGRPC provides the containerd grpc APIs on the provided listener

24
services/containers/service.go
View File

@ -24,23 +24,22 @@ func init() {
plugin.MetadataPlugin, plugin.MetadataPlugin,
}, },
Init: func(ic *plugin.InitContext) (interface{}, error) { Init: func(ic *plugin.InitContext) (interface{}, error) {
e := events.GetPoster(ic.Context)
m, err := ic.Get(plugin.MetadataPlugin) m, err := ic.Get(plugin.MetadataPlugin)
if err != nil { if err != nil {
return nil, err return nil, err
} }
return NewService(m.(*bolt.DB), e), nil return NewService(m.(*bolt.DB), ic.Events), nil
}, },
}) })
} }
type Service struct { type Service struct {
db *bolt.DB db *bolt.DB
emitter events.Poster publisher events.Publisher
} }
func NewService(db *bolt.DB, evts events.Poster) api.ContainersServer { func NewService(db *bolt.DB, publisher events.Publisher) api.ContainersServer {
return &Service{db: db, emitter: evts} return &Service{db: db, publisher: publisher}
} }
func (s *Service) Register(server *grpc.Server) error { func (s *Service) Register(server *grpc.Server) error {
@ -94,7 +93,7 @@ func (s *Service) Create(ctx context.Context, req *api.CreateContainerRequest) (
}); err != nil { }); err != nil {
return &resp, errdefs.ToGRPC(err) return &resp, errdefs.ToGRPC(err)
} }
if err := s.emit(ctx, "/containers/create", &eventsapi.ContainerCreate{ if err := s.publisher.Publish(ctx, "/containers/create", &eventsapi.ContainerCreate{
ID: resp.Container.ID, ID: resp.Container.ID,
Image: resp.Container.Image, Image: resp.Container.Image,
Runtime: &eventsapi.ContainerCreate_Runtime{ Runtime: &eventsapi.ContainerCreate_Runtime{
@ -136,7 +135,7 @@ func (s *Service) Update(ctx context.Context, req *api.UpdateContainerRequest) (
return &resp, errdefs.ToGRPC(err) return &resp, errdefs.ToGRPC(err)
} }
if err := s.emit(ctx, "/containers/update", &eventsapi.ContainerUpdate{ if err := s.publisher.Publish(ctx, "/containers/update", &eventsapi.ContainerUpdate{
ID: resp.Container.ID, ID: resp.Container.ID,
Image: resp.Container.Image, Image: resp.Container.Image,
Labels: resp.Container.Labels, Labels: resp.Container.Labels,
@ -155,7 +154,7 @@ func (s *Service) Delete(ctx context.Context, req *api.DeleteContainerRequest) (
return &empty.Empty{}, errdefs.ToGRPC(err) return &empty.Empty{}, errdefs.ToGRPC(err)
} }
if err := s.emit(ctx, "/containers/delete", &eventsapi.ContainerDelete{ if err := s.publisher.Publish(ctx, "/containers/delete", &eventsapi.ContainerDelete{
ID: req.ID, ID: req.ID,
}); err != nil { }); err != nil {
return &empty.Empty{}, err return &empty.Empty{}, err
@ -175,12 +174,3 @@ func (s *Service) withStoreView(ctx context.Context, fn func(ctx context.Context
func (s *Service) withStoreUpdate(ctx context.Context, fn func(ctx context.Context, store containers.Store) error) error { func (s *Service) withStoreUpdate(ctx context.Context, fn func(ctx context.Context, store containers.Store) error) error {
return s.db.Update(s.withStore(ctx, fn)) return s.db.Update(s.withStore(ctx, fn))
} }
func (s *Service) emit(ctx context.Context, topic string, evt interface{}) error {
emitterCtx := events.WithTopic(ctx, topic)
if err := s.emitter.Post(emitterCtx, evt); err != nil {
return err
}
return nil
}

15
services/content/service.go
View File

@ -24,7 +24,7 @@ import (
type Service struct { type Service struct {
store content.Store store content.Store
emitter events.Poster publisher events.Publisher
} }
var bufPool = sync.Pool{ var bufPool = sync.Pool{
@ -59,7 +59,7 @@ func NewService(ic *plugin.InitContext) (interface{}, error) {
cs := metadata.NewContentStore(m.(*bolt.DB), c.(content.Store)) cs := metadata.NewContentStore(m.(*bolt.DB), c.(content.Store))
return &Service{ return &Service{
store: cs, store: cs,
emitter: events.GetPoster(ic.Context), publisher: ic.Events,
}, nil }, nil
} }
@ -149,7 +149,7 @@ func (s *Service) Delete(ctx context.Context, req *api.DeleteContentRequest) (*e
return nil, errdefs.ToGRPC(err) return nil, errdefs.ToGRPC(err)
} }
if err := s.emit(ctx, "/content/delete", &eventsapi.ContentDelete{ if err := s.publisher.Publish(ctx, "/content/delete", &eventsapi.ContentDelete{
Digest: req.Digest, Digest: req.Digest,
}); err != nil { }); err != nil {
return nil, err return nil, err
@ -459,12 +459,3 @@ func (s *Service) Abort(ctx context.Context, req *api.AbortRequest) (*empty.Empt
return &empty.Empty{}, nil return &empty.Empty{}, nil
} }
func (s *Service) emit(ctx context.Context, topic string, evt interface{}) error {
emitterCtx := events.WithTopic(ctx, topic)
if err := s.emitter.Post(emitterCtx, evt); err != nil {
return err
}
return nil
}

58
services/events/service.go
View File

@ -1,14 +1,13 @@
package events package events
import ( import (
"fmt"
"time"
api "github.com/containerd/containerd/api/services/events/v1" api "github.com/containerd/containerd/api/services/events/v1"
"github.com/containerd/containerd/errdefs"
"github.com/containerd/containerd/events" "github.com/containerd/containerd/events"
"github.com/containerd/containerd/filters"
"github.com/containerd/containerd/plugin" "github.com/containerd/containerd/plugin"
"github.com/golang/protobuf/ptypes/empty" "github.com/golang/protobuf/ptypes/empty"
"github.com/sirupsen/logrus" "github.com/pkg/errors"
"golang.org/x/net/context" "golang.org/x/net/context"
"google.golang.org/grpc" "google.golang.org/grpc"
) )
@ -18,18 +17,17 @@ func init() {
Type: plugin.GRPCPlugin, Type: plugin.GRPCPlugin,
ID: "events", ID: "events",
Init: func(ic *plugin.InitContext) (interface{}, error) { Init: func(ic *plugin.InitContext) (interface{}, error) {
return NewService(ic.Emitter), nil return NewService(ic.Events), nil
}, },
}) })
} }
type Service struct { type Service struct {
emitter *events.Emitter events *events.Exchange
timeouts map[string]*time.Timer
} }
func NewService(e *events.Emitter) api.EventsServer { func NewService(events *events.Exchange) api.EventsServer {
return &Service{emitter: e} return &Service{events: events}
} }
func (s *Service) Register(server *grpc.Server) error { func (s *Service) Register(server *grpc.Server) error {
@ -37,28 +35,36 @@ func (s *Service) Register(server *grpc.Server) error {
return nil return nil
} }
func (s *Service) Stream(req *api.StreamEventsRequest, srv api.Events_StreamServer) error { func (s *Service) Subscribe(req *api.SubscribeRequest, srv api.Events_SubscribeServer) error {
clientID := fmt.Sprintf("%d", time.Now().UnixNano()) ctx, cancel := context.WithCancel(srv.Context())
for { defer cancel()
e := <-s.emitter.Events(srv.Context(), clientID)
// upon the client event timeout this will be nil; ignore filter, err := filters.ParseAll(req.Filters...)
if e == nil { if err != nil {
return nil return errdefs.ToGRPC(err)
} }
if err := srv.Send(e); err != nil {
logrus.WithFields(logrus.Fields{ eventq, errq := s.events.Subscribe(ctx, filter)
"client": clientID, for {
}).Debug("error sending event; unsubscribing client") select {
s.emitter.Remove(clientID) case ev := <-eventq:
return err if err := srv.Send(ev); err != nil {
return errors.Wrapf(err, "failed sending event to subscriber")
}
case err := <-errq:
if err != nil {
return errors.Wrapf(err, "subscription error")
}
return nil
} }
} }
} }
func (s *Service) Post(ctx context.Context, r *api.PostEventRequest) (*empty.Empty, error) { func (s *Service) Publish(ctx context.Context, r *api.PublishRequest) (*empty.Empty, error) {
ctx = events.WithTopic(ctx, r.Envelope.Topic) if err := s.events.Forward(ctx, r.Envelope); err != nil {
if err := s.emitter.Post(ctx, r.Envelope); err != nil { return nil, errdefs.ToGRPC(err)
return nil, err
} }
return &empty.Empty{}, nil return &empty.Empty{}, nil
} }

24
services/images/service.go
View File

@ -24,25 +24,24 @@ func init() {
plugin.MetadataPlugin, plugin.MetadataPlugin,
}, },
Init: func(ic *plugin.InitContext) (interface{}, error) { Init: func(ic *plugin.InitContext) (interface{}, error) {
e := events.GetPoster(ic.Context)
m, err := ic.Get(plugin.MetadataPlugin) m, err := ic.Get(plugin.MetadataPlugin)
if err != nil { if err != nil {
return nil, err return nil, err
} }
return NewService(m.(*bolt.DB), e), nil return NewService(m.(*bolt.DB), ic.Events), nil
}, },
}) })
} }
type Service struct { type Service struct {
db *bolt.DB db *bolt.DB
emitter events.Poster publisher events.Publisher
} }
func NewService(db *bolt.DB, evts events.Poster) imagesapi.ImagesServer { func NewService(db *bolt.DB, publisher events.Publisher) imagesapi.ImagesServer {
return &Service{ return &Service{
db: db, db: db,
emitter: evts, publisher: publisher,
} }
} }
@ -100,7 +99,7 @@ func (s *Service) Create(ctx context.Context, req *imagesapi.CreateImageRequest)
return nil, errdefs.ToGRPC(err) return nil, errdefs.ToGRPC(err)
} }
if err := s.emit(ctx, "/images/create", &eventsapi.ImageCreate{ if err := s.publisher.Publish(ctx, "/images/create", &eventsapi.ImageCreate{
Name: resp.Image.Name, Name: resp.Image.Name,
Labels: resp.Image.Labels, Labels: resp.Image.Labels,
}); err != nil { }); err != nil {
@ -139,7 +138,7 @@ func (s *Service) Update(ctx context.Context, req *imagesapi.UpdateImageRequest)
return nil, errdefs.ToGRPC(err) return nil, errdefs.ToGRPC(err)
} }
if err := s.emit(ctx, "/images/update", &eventsapi.ImageUpdate{ if err := s.publisher.Publish(ctx, "/images/update", &eventsapi.ImageUpdate{
Name: resp.Image.Name, Name: resp.Image.Name,
Labels: resp.Image.Labels, Labels: resp.Image.Labels,
}); err != nil { }); err != nil {
@ -156,7 +155,7 @@ func (s *Service) Delete(ctx context.Context, req *imagesapi.DeleteImageRequest)
return nil, err return nil, err
} }
if err := s.emit(ctx, "/images/delete", &eventsapi.ImageDelete{ if err := s.publisher.Publish(ctx, "/images/delete", &eventsapi.ImageDelete{
Name: req.Name, Name: req.Name,
}); err != nil { }); err != nil {
return nil, err return nil, err
@ -176,12 +175,3 @@ func (s *Service) withStoreView(ctx context.Context, fn func(ctx context.Context
func (s *Service) withStoreUpdate(ctx context.Context, fn func(ctx context.Context, store images.Store) error) error { func (s *Service) withStoreUpdate(ctx context.Context, fn func(ctx context.Context, store images.Store) error) error {
return s.db.Update(s.withStore(ctx, fn)) return s.db.Update(s.withStore(ctx, fn))
} }
func (s *Service) emit(ctx context.Context, topic string, evt interface{}) error {
emitterCtx := events.WithTopic(ctx, topic)
if err := s.emitter.Post(emitterCtx, evt); err != nil {
return err
}
return nil
}

24
services/namespaces/service.go
View File

@ -25,27 +25,26 @@ func init() {
plugin.MetadataPlugin, plugin.MetadataPlugin,
}, },
Init: func(ic *plugin.InitContext) (interface{}, error) { Init: func(ic *plugin.InitContext) (interface{}, error) {
e := events.GetPoster(ic.Context)
m, err := ic.Get(plugin.MetadataPlugin) m, err := ic.Get(plugin.MetadataPlugin)
if err != nil { if err != nil {
return nil, err return nil, err
} }
return NewService(m.(*bolt.DB), e), nil return NewService(m.(*bolt.DB), ic.Events), nil
}, },
}) })
} }
type Service struct { type Service struct {
db *bolt.DB db *bolt.DB
emitter events.Poster publisher events.Publisher
} }
var _ api.NamespacesServer = &Service{} var _ api.NamespacesServer = &Service{}
func NewService(db *bolt.DB, evts events.Poster) api.NamespacesServer { func NewService(db *bolt.DB, publisher events.Publisher) api.NamespacesServer {
return &Service{ return &Service{
db: db, db: db,
emitter: evts, publisher: publisher,
} }
} }
@ -119,7 +118,7 @@ func (s *Service) Create(ctx context.Context, req *api.CreateNamespaceRequest) (
return &resp, err return &resp, err
} }
if err := s.emit(ctx, "/namespaces/create", &eventsapi.NamespaceCreate{ if err := s.publisher.Publish(ctx, "/namespaces/create", &eventsapi.NamespaceCreate{
Name: req.Namespace.Name, Name: req.Namespace.Name,
Labels: req.Namespace.Labels, Labels: req.Namespace.Labels,
}); err != nil { }); err != nil {
@ -172,7 +171,7 @@ func (s *Service) Update(ctx context.Context, req *api.UpdateNamespaceRequest) (
return &resp, err return &resp, err
} }
if err := s.emit(ctx, "/namespaces/update", &eventsapi.NamespaceUpdate{ if err := s.publisher.Publish(ctx, "/namespaces/update", &eventsapi.NamespaceUpdate{
Name: req.Namespace.Name, Name: req.Namespace.Name,
Labels: req.Namespace.Labels, Labels: req.Namespace.Labels,
}); err != nil { }); err != nil {
@ -189,7 +188,7 @@ func (s *Service) Delete(ctx context.Context, req *api.DeleteNamespaceRequest) (
return &empty.Empty{}, err return &empty.Empty{}, err
} }
if err := s.emit(ctx, "/namespaces/delete", &eventsapi.NamespaceDelete{ if err := s.publisher.Publish(ctx, "/namespaces/delete", &eventsapi.NamespaceDelete{
Name: req.Name, Name: req.Name,
}); err != nil { }); err != nil {
return &empty.Empty{}, err return &empty.Empty{}, err
@ -209,12 +208,3 @@ func (s *Service) withStoreView(ctx context.Context, fn func(ctx context.Context
func (s *Service) withStoreUpdate(ctx context.Context, fn func(ctx context.Context, store namespaces.Store) error) error { func (s *Service) withStoreUpdate(ctx context.Context, fn func(ctx context.Context, store namespaces.Store) error) error {
return s.db.Update(s.withStore(ctx, fn)) return s.db.Update(s.withStore(ctx, fn))
} }
func (s *Service) emit(ctx context.Context, topic string, evt interface{}) error {
emitterCtx := events.WithTopic(ctx, topic)
if err := s.emitter.Post(emitterCtx, evt); err != nil {
return err
}
return nil
}

20
services/snapshot/service.go
View File

@ -45,11 +45,10 @@ var empty = &protoempty.Empty{}
type service struct { type service struct {
snapshotters map[string]snapshot.Snapshotter snapshotters map[string]snapshot.Snapshotter
defaultSnapshotterName string defaultSnapshotterName string
emitter events.Poster publisher events.Publisher
} }
func newService(ic *plugin.InitContext) (interface{}, error) { func newService(ic *plugin.InitContext) (interface{}, error) {
evts := events.GetPoster(ic.Context)
rawSnapshotters, err := ic.GetAll(plugin.SnapshotPlugin) rawSnapshotters, err := ic.GetAll(plugin.SnapshotPlugin)
if err != nil { if err != nil {
return nil, err return nil, err
@ -72,7 +71,7 @@ func newService(ic *plugin.InitContext) (interface{}, error) {
return &service{ return &service{
snapshotters: snapshotters, snapshotters: snapshotters,
defaultSnapshotterName: cfg.Default, defaultSnapshotterName: cfg.Default,
emitter: evts, publisher: ic.Events,
}, nil }, nil
} }
@ -105,7 +104,7 @@ func (s *service) Prepare(ctx context.Context, pr *snapshotapi.PrepareSnapshotRe
return nil, errdefs.ToGRPC(err) return nil, errdefs.ToGRPC(err)
} }
if err := s.emit(ctx, "/snapshot/prepare", &eventsapi.SnapshotPrepare{ if err := s.publisher.Publish(ctx, "/snapshot/prepare", &eventsapi.SnapshotPrepare{
Key: pr.Key, Key: pr.Key,
Parent: pr.Parent, Parent: pr.Parent,
}); err != nil { }); err != nil {
@ -162,7 +161,7 @@ func (s *service) Commit(ctx context.Context, cr *snapshotapi.CommitSnapshotRequ
return nil, errdefs.ToGRPC(err) return nil, errdefs.ToGRPC(err)
} }
if err := s.emit(ctx, "/snapshot/commit", &eventsapi.SnapshotCommit{ if err := s.publisher.Publish(ctx, "/snapshot/commit", &eventsapi.SnapshotCommit{
Key: cr.Key, Key: cr.Key,
Name: cr.Name, Name: cr.Name,
}); err != nil { }); err != nil {
@ -183,7 +182,7 @@ func (s *service) Remove(ctx context.Context, rr *snapshotapi.RemoveSnapshotRequ
return nil, errdefs.ToGRPC(err) return nil, errdefs.ToGRPC(err)
} }
if err := s.emit(ctx, "/snapshot/remove", &eventsapi.SnapshotRemove{ if err := s.publisher.Publish(ctx, "/snapshot/remove", &eventsapi.SnapshotRemove{
Key: rr.Key, Key: rr.Key,
}); err != nil { }); err != nil {
return nil, err return nil, err
@ -294,12 +293,3 @@ func fromMounts(mounts []mount.Mount) []*types.Mount {
} }
return out return out
} }
func (s *service) emit(ctx context.Context, topic string, evt interface{}) error {
emitterCtx := events.WithTopic(ctx, topic)
if err := s.emitter.Post(emitterCtx, evt); err != nil {
return err
}
return nil
}

14
services/tasks/service.go
View File

@ -67,12 +67,11 @@ func New(ic *plugin.InitContext) (interface{}, error) {
r := rr.(runtime.Runtime) r := rr.(runtime.Runtime)
runtimes[r.ID()] = r runtimes[r.ID()] = r
} }
e := events.GetPoster(ic.Context)
return &Service{ return &Service{
runtimes: runtimes, runtimes: runtimes,
db: m.(*bolt.DB), db: m.(*bolt.DB),
store: cs, store: cs,
emitter: e, publisher: ic.Events,
}, nil }, nil
} }
@ -80,7 +79,7 @@ type Service struct {
runtimes map[string]runtime.Runtime runtimes map[string]runtime.Runtime
db *bolt.DB db *bolt.DB
store content.Store store content.Store
emitter events.Poster publisher events.Publisher
} }
func (s *Service) Register(server *grpc.Server) error { func (s *Service) Register(server *grpc.Server) error {
@ -502,12 +501,3 @@ func (s *Service) getRuntime(name string) (runtime.Runtime, error) {
} }
return runtime, nil return runtime, nil
} }
func (s *Service) emit(ctx context.Context, topic string, evt interface{}) error {
emitterCtx := events.WithTopic(ctx, topic)
if err := s.emitter.Post(emitterCtx, evt); err != nil {
return err
}
return nil
}

2
task.go
View File

@ -161,7 +161,7 @@ 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 // 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) { func (t *task) Wait(ctx context.Context) (uint32, error) {
eventstream, err := t.client.EventService().Stream(ctx, &eventsapi.StreamEventsRequest{}) eventstream, err := t.client.EventService().Subscribe(ctx, &eventsapi.SubscribeRequest{})
if err != nil { if err != nil {
return UnknownExitStatus, errdefs.FromGRPC(err) return UnknownExitStatus, errdefs.FromGRPC(err)
} }

12
windows/runtime.go
View File

@ -64,7 +64,7 @@ func New(ic *plugin.InitContext) (interface{}, error) {
pidPool: newPidPool(), pidPool: newPidPool(),
events: make(chan interface{}, 4096), events: make(chan interface{}, 4096),
emitter: ic.Emitter, publisher: ic.Events,
// TODO(mlaventure): windows needs a stat monitor // TODO(mlaventure): windows needs a stat monitor
monitor: nil, monitor: nil,
tasks: runtime.NewTaskList(), tasks: runtime.NewTaskList(),
@ -84,7 +84,7 @@ type windowsRuntime struct {
root string root string
pidPool *pidPool pidPool *pidPool
emitter *events.Emitter publisher events.Publisher
events chan interface{} events chan interface{}
monitor runtime.TaskMonitor monitor runtime.TaskMonitor
@ -194,7 +194,8 @@ func (r *windowsRuntime) Delete(ctx context.Context, t runtime.Task) (*runtime.E
wt.cleanup() wt.cleanup()
r.tasks.Delete(ctx, t) r.tasks.Delete(ctx, t)
r.emitter.Post(events.WithTopic(ctx, runtime.TaskDeleteEventTopic), r.publisher.Publish(ctx,
runtime.TaskDeleteEventTopic,
&eventsapi.TaskDelete{ &eventsapi.TaskDelete{
ContainerID: wt.id, ContainerID: wt.id,
Pid: wt.pid, Pid: wt.pid,
@ -296,7 +297,7 @@ func (r *windowsRuntime) newTask(ctx context.Context, namespace, id string, spec
spec: spec, spec: spec,
processes: make(map[string]*process), processes: make(map[string]*process),
hyperV: spec.Windows.HyperV != nil, hyperV: spec.Windows.HyperV != nil,
emitter: r.emitter, publisher: r.publisher,
rwLayer: conf.LayerFolderPath, rwLayer: conf.LayerFolderPath,
pidPool: r.pidPool, pidPool: r.pidPool,
hcsContainer: ctr, hcsContainer: ctr,
@ -312,7 +313,8 @@ func (r *windowsRuntime) newTask(ctx context.Context, namespace, id string, spec
}) })
} }
r.emitter.Post(events.WithTopic(ctx, runtime.TaskCreateEventTopic), r.publisher.Publish(ctx,
runtime.TaskCreateEventTopic,
&eventsapi.TaskCreate{ &eventsapi.TaskCreate{
ContainerID: id, ContainerID: id,
IO: &eventsapi.TaskIO{ IO: &eventsapi.TaskIO{

17
windows/task.go
View File

@ -34,7 +34,7 @@ type task struct {
processes map[string]*process processes map[string]*process
hyperV bool hyperV bool
emitter *events.Emitter publisher events.Publisher
rwLayer string rwLayer string
pidPool *pidPool pidPool *pidPool
@ -112,7 +112,8 @@ func (t *task) Start(ctx context.Context) error {
return err return err
} }
t.emitter.Post(events.WithTopic(ctx, runtime.TaskStartEventTopic), t.publisher.Publish(ctx,
runtime.TaskStartEventTopic,
&eventsapi.TaskStart{ &eventsapi.TaskStart{
ContainerID: t.id, ContainerID: t.id,
Pid: t.pid, Pid: t.pid,
@ -130,7 +131,8 @@ func (t *task) Pause(ctx context.Context) error {
t.Unlock() t.Unlock()
} }
if err == nil { if err == nil {
t.emitter.Post(events.WithTopic(ctx, runtime.TaskPausedEventTopic), t.publisher.Publish(ctx,
runtime.TaskPausedEventTopic,
&eventsapi.TaskPaused{ &eventsapi.TaskPaused{
ContainerID: t.id, ContainerID: t.id,
}) })
@ -150,7 +152,8 @@ func (t *task) Resume(ctx context.Context) error {
t.Unlock() t.Unlock()
} }
if err == nil { if err == nil {
t.emitter.Post(events.WithTopic(ctx, runtime.TaskResumedEventTopic), t.publisher.Publish(ctx,
runtime.TaskResumedEventTopic,
&eventsapi.TaskResumed{ &eventsapi.TaskResumed{
ContainerID: t.id, ContainerID: t.id,
}) })
@ -195,7 +198,8 @@ func (t *task) Exec(ctx context.Context, id string, opts runtime.ExecOpts) (runt
return nil, err return nil, err
} }
t.emitter.Post(events.WithTopic(ctx, runtime.TaskExecAddedEventTopic), t.publisher.Publish(ctx,
runtime.TaskExecAddedEventTopic,
&eventsapi.TaskExecAdded{ &eventsapi.TaskExecAdded{
ContainerID: t.id, ContainerID: t.id,
ExecID: id, ExecID: id,
@ -358,7 +362,8 @@ func (t *task) newProcess(ctx context.Context, id string, conf *hcsshim.ProcessC
} }
wp.exitCode = uint32(ec) wp.exitCode = uint32(ec)
t.emitter.Post(events.WithTopic(ctx, runtime.TaskExitEventTopic), t.publisher.Publish(ctx,
runtime.TaskExitEventTopic,
&eventsapi.TaskExit{ &eventsapi.TaskExit{
ContainerID: t.id, ContainerID: t.id,
ID: id, ID: id,