Add ttrpc server to containerd

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

Protobuild.toml

@@ -35,6 +35,10 @@ plugins = ["grpc", "fieldpath"]
 prefixes = ["github.com/containerd/containerd/api/events"]
 plugins = ["fieldpath"] # disable grpc for this package
 
+[[overrides]]
+prefixes = ["github.com/containerd/containerd/api/services/ttrpc/events/v1"]
+plugins = ["ttrpc", "fieldpath"]
+
 [[overrides]]
 # enable ttrpc and disable fieldpath and grpc for the shim
 prefixes = ["github.com/containerd/containerd/runtime/v1/shim/v1", "github.com/containerd/containerd/runtime/v2/task"]

api/next.pb.txt

@@ -4202,6 +4202,119 @@ file {
   weak_dependency: 2
   syntax: "proto3"
 }
+file {
+  name: "github.com/containerd/containerd/api/services/ttrpc/events/v1/events.proto"
+  package: "containerd.services.events.ttrpc.v1"
+  dependency: "github.com/containerd/containerd/protobuf/plugin/fieldpath.proto"
+  dependency: "gogoproto/gogo.proto"
+  dependency: "google/protobuf/any.proto"
+  dependency: "google/protobuf/empty.proto"
+  dependency: "google/protobuf/timestamp.proto"
+  message_type {
+    name: "PublishRequest"
+    field {
+      name: "topic"
+      number: 1
+      label: LABEL_OPTIONAL
+      type: TYPE_STRING
+      json_name: "topic"
+    }
+    field {
+      name: "event"
+      number: 2
+      label: LABEL_OPTIONAL
+      type: TYPE_MESSAGE
+      type_name: ".google.protobuf.Any"
+      json_name: "event"
+    }
+  }
+  message_type {
+    name: "ForwardRequest"
+    field {
+      name: "envelope"
+      number: 1
+      label: LABEL_OPTIONAL
+      type: TYPE_MESSAGE
+      type_name: ".containerd.services.events.ttrpc.v1.Envelope"
+      json_name: "envelope"
+    }
+  }
+  message_type {
+    name: "SubscribeRequest"
+    field {
+      name: "filters"
+      number: 1
+      label: LABEL_REPEATED
+      type: TYPE_STRING
+      json_name: "filters"
+    }
+  }
+  message_type {
+    name: "Envelope"
+    field {
+      name: "timestamp"
+      number: 1
+      label: LABEL_OPTIONAL
+      type: TYPE_MESSAGE
+      type_name: ".google.protobuf.Timestamp"
+      options {
+        65001: 0
+        65010: 1
+      }
+      json_name: "timestamp"
+    }
+    field {
+      name: "namespace"
+      number: 2
+      label: LABEL_OPTIONAL
+      type: TYPE_STRING
+      json_name: "namespace"
+    }
+    field {
+      name: "topic"
+      number: 3
+      label: LABEL_OPTIONAL
+      type: TYPE_STRING
+      json_name: "topic"
+    }
+    field {
+      name: "event"
+      number: 4
+      label: LABEL_OPTIONAL
+      type: TYPE_MESSAGE
+      type_name: ".google.protobuf.Any"
+      json_name: "event"
+    }
+    options {
+      64400: 1
+    }
+  }
+  service {
+    name: "Events"
+    method {
+      name: "Publish"
+      input_type: ".containerd.services.events.ttrpc.v1.PublishRequest"
+      output_type: ".google.protobuf.Empty"
+    }
+    method {
+      name: "Forward"
+      input_type: ".containerd.services.events.ttrpc.v1.ForwardRequest"
+      output_type: ".google.protobuf.Empty"
+    }
+    method {
+      name: "Subscribe"
+      input_type: ".containerd.services.events.ttrpc.v1.SubscribeRequest"
+      output_type: ".containerd.services.events.ttrpc.v1.Envelope"
+      server_streaming: true
+    }
+  }
+  options {
+    go_package: "github.com/containerd/containerd/api/services/ttrpc/events/v1;events"
+  }
+  weak_dependency: 0
+  weak_dependency: 1
+  syntax: "proto3"
+}
 file {
   name: "github.com/containerd/containerd/api/services/version/v1/version.proto"
   package: "containerd.services.version.v1"

api/services/ttrpc/events/v1/doc.go

@@ -0,0 +1,18 @@
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+// Package events defines the event pushing and subscription service.
+package events

api/services/ttrpc/events/v1/events.proto

@@ -0,0 +1,56 @@
+syntax = "proto3";
+
+package containerd.services.events.ttrpc.v1;
+
+import weak "github.com/containerd/containerd/protobuf/plugin/fieldpath.proto";
+import weak "gogoproto/gogo.proto";
+import "google/protobuf/any.proto";
+import "google/protobuf/empty.proto";
+import "google/protobuf/timestamp.proto";
+
+option go_package = "github.com/containerd/containerd/api/services/ttrpc/events/v1;events";
+
+service Events {
+	// Publish an event to a topic.
+	//
+	// The event will be packed into a timestamp envelope with the namespace
+	// introspected from the context. The envelope will then be dispatched.
+	rpc Publish(PublishRequest) returns (google.protobuf.Empty);
+
+	// Forward sends an event that has already been packaged into an envelope
+	// with a timestamp and namespace.
+	//
+	// This is useful if earlier timestamping is required or when forwarding on
+	// behalf of another component, namespace or publisher.
+	rpc Forward(ForwardRequest) returns (google.protobuf.Empty);
+
+	// Subscribe to a stream of events, possibly returning only that match any
+	// of the provided filters.
+	//
+	// Unlike many other methods in containerd, subscribers will get messages
+	// from all namespaces unless otherwise specified. If this is not desired,
+	// a filter can be provided in the format 'namespace==<namespace>' to
+	// restrict the received events.
+	rpc Subscribe(SubscribeRequest) returns (stream Envelope);
+}
+
+message PublishRequest {
+	string topic = 1;
+	google.protobuf.Any event = 2;
+}
+
+message ForwardRequest {
+	Envelope envelope = 1;
+}
+
+message SubscribeRequest {
+	repeated string filters = 1;
+}
+
+message Envelope {
+	option (containerd.plugin.fieldpath) = true;
+	google.protobuf.Timestamp timestamp = 1 [(gogoproto.stdtime) = true, (gogoproto.nullable) = false];
+	string namespace = 2;
+	string topic = 3;
+	google.protobuf.Any event = 4;
+}

cmd/containerd/command/main.go

@@ -147,7 +147,10 @@ func App() *cli.App {
 		for _, w := range warnings {
 			log.G(ctx).WithError(w).Warn("cleanup temp mount")
 		}
-		address := config.GRPC.Address
+		var (
+			address      = config.GRPC.Address
+			ttrpcAddress = fmt.Sprintf("%s.ttrpc", config.GRPC.Address)
+		)
 		if address == "" {
 			return errors.New("grpc address cannot be empty")
 		}
@@ -188,7 +191,14 @@ func App() *cli.App {
 			}
 			serve(ctx, l, server.ServeMetrics)
 		}
+		// setup the ttrpc endpoint
+		tl, err := sys.GetLocalListener(ttrpcAddress, config.GRPC.UID, config.GRPC.GID)
+		if err != nil {
+			return errors.Wrapf(err, "failed to get listener for main ttrpc endpoint")
+		}
+		serve(ctx, tl, server.ServeTTRPC)
 
+		// setup the main grpc endpoint
 		l, err := sys.GetLocalListener(address, config.GRPC.UID, config.GRPC.GID)
 		if err != nil {
 			return errors.Wrapf(err, "failed to get listener for main endpoint")

plugin/plugin.go

@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"sync"
 
+	"github.com/containerd/ttrpc"
 	"github.com/pkg/errors"
 	"google.golang.org/grpc"
 )
@@ -123,6 +124,11 @@ type Service interface {
 	Register(*grpc.Server) error
 }
 
+// TTRPCService allows TTRPC services to be registered with the underlying server
+type TTRPCService interface {
+	RegisterTTRPC(*ttrpc.Server) error
+}
+
 var register = struct {
 	sync.RWMutex
 	r []*Registration

services/events/service.go

@@ -20,10 +20,12 @@ import (
 	"context"
 
 	api "github.com/containerd/containerd/api/services/events/v1"
+	apittrpc "github.com/containerd/containerd/api/services/ttrpc/events/v1"
 	"github.com/containerd/containerd/errdefs"
 	"github.com/containerd/containerd/events"
 	"github.com/containerd/containerd/events/exchange"
 	"github.com/containerd/containerd/plugin"
+	"github.com/containerd/ttrpc"
 	ptypes "github.com/gogo/protobuf/types"
 	"github.com/pkg/errors"
 	"google.golang.org/grpc"
@@ -40,12 +42,18 @@ func init() {
 }
 
 type service struct {
-	events *exchange.Exchange
+	ttService *ttrpcService
+	events    *exchange.Exchange
 }
 
 // NewService returns the GRPC events server
 func NewService(events *exchange.Exchange) api.EventsServer {
-	return &service{events: events}
+	return &service{
+		ttService: &ttrpcService{
+			events: events,
+		},
+		events: events,
+	}
 }
 
 func (s *service) Register(server *grpc.Server) error {
@@ -53,6 +61,11 @@ func (s *service) Register(server *grpc.Server) error {
 	return nil
 }
 
+func (s *service) RegisterTTRPC(server *ttrpc.Server) error {
+	apittrpc.RegisterEventsService(server, s.ttService)
+	return nil
+}
+
 func (s *service) Publish(ctx context.Context, r *api.PublishRequest) (*ptypes.Empty, error) {
 	if err := s.events.Publish(ctx, r.Topic, r.Event); err != nil {
 		return nil, errdefs.ToGRPC(err)

services/events/ttrpc.go

@@ -0,0 +1,60 @@
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package events
+
+import (
+	"context"
+
+	api "github.com/containerd/containerd/api/services/ttrpc/events/v1"
+	"github.com/containerd/containerd/errdefs"
+	"github.com/containerd/containerd/events"
+	"github.com/containerd/containerd/events/exchange"
+	ptypes "github.com/gogo/protobuf/types"
+)
+
+type ttrpcService struct {
+	events *exchange.Exchange
+}
+
+func (s *ttrpcService) Publish(ctx context.Context, r *api.PublishRequest) (*ptypes.Empty, error) {
+	if err := s.events.Publish(ctx, r.Topic, r.Event); err != nil {
+		return nil, errdefs.ToGRPC(err)
+	}
+
+	return &ptypes.Empty{}, nil
+}
+
+func (s *ttrpcService) Forward(ctx context.Context, r *api.ForwardRequest) (*ptypes.Empty, error) {
+	if err := s.events.Forward(ctx, fromTTProto(r.Envelope)); err != nil {
+		return nil, errdefs.ToGRPC(err)
+	}
+
+	return &ptypes.Empty{}, nil
+}
+
+func (s *ttrpcService) Subscribe(ctx context.Context, req *api.SubscribeRequest) (*api.Envelope, error) {
+	return nil, errdefs.ToGRPCf(errdefs.ErrNotImplemented, "ttrpc does not support streaming")
+}
+
+func fromTTProto(env *api.Envelope) *events.Envelope {
+	return &events.Envelope{
+		Timestamp: env.Timestamp,
+		Namespace: env.Namespace,
+		Topic:     env.Topic,
+		Event:     env.Event,
+	}
+}

services/server/server.go

@@ -44,6 +44,7 @@ import (
 	"github.com/containerd/containerd/snapshots"
 	ssproxy "github.com/containerd/containerd/snapshots/proxy"
 	"github.com/containerd/containerd/sys"
+	"github.com/containerd/ttrpc"
 	metrics "github.com/docker/go-metrics"
 	grpc_prometheus "github.com/grpc-ecosystem/go-grpc-prometheus"
 	"github.com/pkg/errors"
@@ -91,13 +92,19 @@ func New(ctx context.Context, config *srvconfig.Config) (*Server, error) {
 	if config.GRPC.MaxSendMsgSize > 0 {
 		serverOpts = append(serverOpts, grpc.MaxSendMsgSize(config.GRPC.MaxSendMsgSize))
 	}
-	rpc := grpc.NewServer(serverOpts...)
+	ttrpcServer, err := newTTRPCServer()
+	if err != nil {
+		return nil, err
+	}
+	grpcServer := grpc.NewServer(serverOpts...)
 	var (
-		services []plugin.Service
+		grpcServices  []plugin.Service
-		s        = &Server{
+		ttrpcServices []plugin.TTRPCService
-			rpc:    rpc,
+		s             = &Server{
-			events: exchange.NewExchange(),
+			grpcServer:  grpcServer,
-			config: config,
+			ttrpcServer: ttrpcServer,
+			events:      exchange.NewExchange(),
+			config:      config,
 		}
 		initialized = plugin.NewPluginSet()
 	)
@@ -138,14 +145,22 @@ func New(ctx context.Context, config *srvconfig.Config) (*Server, error) {
 			continue
 		}
 		// check for grpc services that should be registered with the server
-		if service, ok := instance.(plugin.Service); ok {
+		if src, ok := instance.(plugin.Service); ok {
-			services = append(services, service)
+			grpcServices = append(grpcServices, src)
+		}
+		if src, ok := instance.(plugin.TTRPCService); ok {
+			ttrpcServices = append(ttrpcServices, src)
 		}
 		s.plugins = append(s.plugins, result)
 	}
 	// register services after all plugins have been initialized
-	for _, service := range services {
+	for _, service := range grpcServices {
-		if err := service.Register(rpc); err != nil {
+		if err := service.Register(grpcServer); err != nil {
+			return nil, err
+		}
+	}
+	for _, service := range ttrpcServices {
+		if err := service.RegisterTTRPC(ttrpcServer); err != nil {
 			return nil, err
 		}
 	}
@@ -154,10 +169,11 @@ func New(ctx context.Context, config *srvconfig.Config) (*Server, error) {
 
 // Server is the containerd main daemon
 type Server struct {
-	rpc     *grpc.Server
+	grpcServer  *grpc.Server
-	events  *exchange.Exchange
+	ttrpcServer *ttrpc.Server
-	config  *srvconfig.Config
+	events      *exchange.Exchange
-	plugins []*plugin.Plugin
+	config      *srvconfig.Config
+	plugins     []*plugin.Plugin
 }
 
 // ServeGRPC provides the containerd grpc APIs on the provided listener
@@ -169,8 +185,13 @@ func (s *Server) ServeGRPC(l net.Listener) error {
 	// before we start serving the grpc API register the grpc_prometheus metrics
 	// handler.  This needs to be the last service registered so that it can collect
 	// metrics for every other service
-	grpc_prometheus.Register(s.rpc)
+	grpc_prometheus.Register(s.grpcServer)
-	return trapClosedConnErr(s.rpc.Serve(l))
+	return trapClosedConnErr(s.grpcServer.Serve(l))
+}
+
+// ServeTTRPC provides the containerd ttrpc APIs on the provided listener
+func (s *Server) ServeTTRPC(l net.Listener) error {
+	return trapClosedConnErr(s.ttrpcServer.Serve(context.Background(), l))
 }
 
 // ServeMetrics provides a prometheus endpoint for exposing metrics
@@ -196,7 +217,7 @@ func (s *Server) ServeDebug(l net.Listener) error {
 
 // Stop the containerd server canceling any open connections
 func (s *Server) Stop() {
-	s.rpc.Stop()
+	s.grpcServer.Stop()
 	for i := len(s.plugins) - 1; i >= 0; i-- {
 		p := s.plugins[i]
 		instance, err := p.Instance()

services/server/server_linux.go

@@ -24,6 +24,7 @@ import (
 	"github.com/containerd/containerd/log"
 	srvconfig "github.com/containerd/containerd/services/server/config"
 	"github.com/containerd/containerd/sys"
+	"github.com/containerd/ttrpc"
 	specs "github.com/opencontainers/runtime-spec/specs-go"
 )
 
@@ -53,3 +54,7 @@ func apply(ctx context.Context, config *srvconfig.Config) error {
 	}
 	return nil
 }
+
+func newTTRPCServer() (*ttrpc.Server, error) {
+	return ttrpc.NewServer(ttrpc.WithServerHandshaker(ttrpc.UnixSocketRequireSameUser()))
+}

services/server/server_windows.go

@@ -22,8 +22,13 @@ import (
 	"context"
 
 	srvconfig "github.com/containerd/containerd/services/server/config"
+	"github.com/containerd/ttrpc"
 )
 
 func apply(_ context.Context, _ *srvconfig.Config) error {
 	return nil
 }
+
+func newTTRPCServer() (*ttrpc.Server, error) {
+	return ttrpc.NewServer()
+}