Merge pull request #82 from mikebrow/containerd-client-library

Use containerd client library to connect to containerd services

pkg/server/server.go

@@ -21,22 +21,15 @@ import (
 	"net"
 	"os"
 	"syscall"
-	"time"
 
-	"github.com/containerd/containerd/namespaces"
 	"github.com/golang/glog"
-	"golang.org/x/net/context"
 	"google.golang.org/grpc"
 	runtime "k8s.io/kubernetes/pkg/kubelet/apis/cri/v1alpha1"
 	"k8s.io/kubernetes/pkg/util/interrupt"
 )
 
-const (
-	// unixProtocol is the network protocol of unix socket.
-	unixProtocol = "unix"
-	// k8sContainerdNamespace is the namespace we use to connect containerd.
-	k8sContainerdNamespace = "k8s.io"
-)
+// unixProtocol is the network protocol of unix socket.
+const unixProtocol = "unix"
 
 // CRIContainerdServer is the grpc server of cri-containerd.
 type CRIContainerdServer struct {
@@ -79,38 +72,3 @@ func (s *CRIContainerdServer) Run() error {
 	h := interrupt.New(nil, s.server.Stop)
 	return h.Run(func() error { return s.server.Serve(l) })
 }
-
-// ConnectToContainerd returns a grpc client for containerd.
-func ConnectToContainerd(path string, connectionTimeout time.Duration) (*grpc.ClientConn, error) {
-	// get the containerd client
-	dialOpts := []grpc.DialOption{
-		grpc.WithInsecure(),
-		grpc.WithTimeout(connectionTimeout),
-		grpc.WithDialer(func(addr string, timeout time.Duration) (net.Conn, error) {
-			return net.DialTimeout(unixProtocol, path, timeout)
-		}),
-		grpc.WithUnaryInterceptor(grpc.UnaryClientInterceptor(unary)),
-		grpc.WithStreamInterceptor(grpc.StreamClientInterceptor(stream)),
-	}
-	return grpc.Dial(fmt.Sprintf("%s://%s", unixProtocol, path), dialOpts...)
-}
-
-// TODO(random-liu): Get rid of following functions after switching to containerd client.
-// unary is a wrapper to apply kubernetes namespace in each grpc unary call.
-func unary(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
-	_, ok := namespaces.Namespace(ctx)
-	if !ok {
-		ctx = namespaces.WithNamespace(ctx, k8sContainerdNamespace)
-	}
-	return invoker(ctx, method, req, reply, cc, opts...)
-}
-
-// stream is a wrapper to apply kubernetes namespace in each grpc stream call.
-func stream(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) {
-	_, ok := namespaces.Namespace(ctx)
-	if !ok {
-		ctx = namespaces.WithNamespace(ctx, k8sContainerdNamespace)
-	}
-
-	return streamer(ctx, desc, cc, method, opts...)
-}

pkg/server/service.go

@@ -19,23 +19,16 @@ package server
 import (
 	"fmt"
 
+	"github.com/containerd/containerd"
 	"github.com/containerd/containerd/api/services/containers"
-	contentapi "github.com/containerd/containerd/api/services/content"
-	diffapi "github.com/containerd/containerd/api/services/diff"
 	"github.com/containerd/containerd/api/services/execution"
-	imagesapi "github.com/containerd/containerd/api/services/images"
-	snapshotapi "github.com/containerd/containerd/api/services/snapshot"
 	versionapi "github.com/containerd/containerd/api/services/version"
 	"github.com/containerd/containerd/content"
 	"github.com/containerd/containerd/images"
-	contentservice "github.com/containerd/containerd/services/content"
 	diffservice "github.com/containerd/containerd/services/diff"
-	imagesservice "github.com/containerd/containerd/services/images"
-	snapshotservice "github.com/containerd/containerd/services/snapshot"
 	"github.com/containerd/containerd/snapshot"
 	"github.com/docker/docker/pkg/truncindex"
 	"github.com/kubernetes-incubator/cri-o/pkg/ocicni"
-	"google.golang.org/grpc"
 	healthapi "google.golang.org/grpc/health/grpc_health_v1"
 	runtime "k8s.io/kubernetes/pkg/kubelet/apis/cri/v1alpha1"
 
@@ -46,6 +39,9 @@ import (
 	"github.com/kubernetes-incubator/cri-containerd/pkg/server/agents"
 )
 
+// k8sContainerdNamespace is the namespace we use to connect containerd.
+const k8sContainerdNamespace = "k8s.io"
+
 // CRIContainerdService is the interface implement CRI remote service server.
 type CRIContainerdService interface {
 	Start()
@@ -99,11 +95,19 @@ type criContainerdService struct {
 	netPlugin ocicni.CNIPlugin
 	// agentFactory is the factory to create agent used in the cri containerd service.
 	agentFactory agents.AgentFactory
+	// client is an instance of the containerd client
+	client *containerd.Client
 }
 
 // NewCRIContainerdService returns a new instance of CRIContainerdService
-func NewCRIContainerdService(conn *grpc.ClientConn, rootDir, networkPluginBinDir, networkPluginConfDir string) (CRIContainerdService, error) {
+func NewCRIContainerdService(containerdEndpoint, rootDir, networkPluginBinDir, networkPluginConfDir string) (CRIContainerdService, error) {
 	// TODO(random-liu): [P2] Recover from runtime state and metadata store.
+
+	client, err := containerd.New(containerdEndpoint, containerd.WithDefaultNamespace(k8sContainerdNamespace))
+	if err != nil {
+		return nil, fmt.Errorf("failed to initialize containerd client with endpoint %q: %v", containerdEndpoint, err)
+	}
+
 	c := &criContainerdService{
 		os:                 osinterface.RealOS{},
 		rootDir:            rootDir,
@@ -117,15 +121,16 @@ func NewCRIContainerdService(conn *grpc.ClientConn, rootDir, networkPluginBinDir
 		sandboxIDIndex:   truncindex.NewTruncIndex(nil),
 		// TODO(random-liu): Add container id index.
 		containerNameIndex:  registrar.NewRegistrar(),
-		containerService:    containers.NewContainersClient(conn),
-		taskService:         execution.NewTasksClient(conn),
-		imageStoreService:   imagesservice.NewStoreFromClient(imagesapi.NewImagesClient(conn)),
-		contentStoreService: contentservice.NewStoreFromClient(contentapi.NewContentClient(conn)),
-		snapshotService:     snapshotservice.NewSnapshotterFromClient(snapshotapi.NewSnapshotClient(conn)),
-		diffService:         diffservice.NewDiffServiceFromClient(diffapi.NewDiffClient(conn)),
-		versionService:      versionapi.NewVersionClient(conn),
-		healthService:       healthapi.NewHealthClient(conn),
+		containerService:    client.ContainerService(),
+		taskService:         client.TaskService(),
+		imageStoreService:   client.ImageService(),
+		contentStoreService: client.ContentStore(),
+		snapshotService:     client.SnapshotService(),
+		diffService:         client.DiffService(),
+		versionService:      client.VersionService(),
+		healthService:       client.HealthService(),
 		agentFactory:        agents.NewAgentFactory(),
+		client:              client,
 	}
 
 	netPlugin, err := ocicni.InitCNI(networkPluginBinDir, networkPluginConfDir)