cri-up

cri/config/config.go

@@ -10,4 +10,6 @@ type Config struct {
 	NetworkPluginConfDir string
 	// SandboxImage is the image used by sandbox container.
 	SandboxImage string
+	// CriVersion is the cri version
+	CriVersion string
 }

cri/criservice.go

@@ -0,0 +1,104 @@
+package cri
+
+import (
+	criv1alpha1 "github.com/alibaba/pouch/cri/v1alpha1"
+	servicev1alpha1 "github.com/alibaba/pouch/cri/v1alpha1/service"
+	criv1alpha2 "github.com/alibaba/pouch/cri/v1alpha2"
+	servicev1alpha2 "github.com/alibaba/pouch/cri/v1alpha2/service"
+	"github.com/alibaba/pouch/daemon/config"
+	"github.com/alibaba/pouch/daemon/mgr"
+
+	"github.com/sirupsen/logrus"
+)
+
+// RunCriService start cri service if pouchd is specified with --enable-cri.
+func RunCriService(daemonconfig *config.Config, containerMgr mgr.ContainerMgr, imageMgr mgr.ImageMgr, stopCh chan error) {
+	var err error
+
+	defer func() {
+		stopCh <- err
+		close(stopCh)
+	}()
+	if !daemonconfig.IsCriEnabled {
+		return
+	}
+	switch daemonconfig.CriConfig.CriVersion {
+	case "v1alpha1":
+		runv1alpha1(daemonconfig, containerMgr, imageMgr)
+	case "v1alpha2":
+		runv1alpha2(daemonconfig, containerMgr, imageMgr)
+	}
+	return
+}
+
+func runv1alpha1(daemonconfig *config.Config, containerMgr mgr.ContainerMgr, imageMgr mgr.ImageMgr) {
+	criMgr, err := criv1alpha1.NewCriManager(daemonconfig, containerMgr, imageMgr)
+	if err != nil {
+		return
+	}
+
+	service, err := servicev1alpha1.NewService(daemonconfig, criMgr)
+	if err != nil {
+		return
+	}
+
+	grpcServerCloseCh := make(chan struct{})
+	go func() {
+		if err := service.Serve(); err != nil {
+			logrus.Errorf("failed to start grpc server: %v", err)
+		}
+		close(grpcServerCloseCh)
+	}()
+
+	streamServerCloseCh := make(chan struct{})
+	go func() {
+		if err := criMgr.StreamServerStart(); err != nil {
+			logrus.Errorf("failed to start stream server: %v", err)
+		}
+		close(streamServerCloseCh)
+	}()
+
+	<-streamServerCloseCh
+	logrus.Infof("CRI Stream server stopped")
+	<-grpcServerCloseCh
+	logrus.Infof("CRI GRPC server stopped")
+
+	logrus.Infof("CRI service stopped")
+	return
+}
+
+func runv1alpha2(daemonconfig *config.Config, containerMgr mgr.ContainerMgr, imageMgr mgr.ImageMgr) {
+	criMgr, err := criv1alpha2.NewCriManager(daemonconfig, containerMgr, imageMgr)
+	if err != nil {
+		return
+	}
+
+	service, err := servicev1alpha2.NewService(daemonconfig, criMgr)
+	if err != nil {
+		return
+	}
+
+	grpcServerCloseCh := make(chan struct{})
+	go func() {
+		if err := service.Serve(); err != nil {
+			logrus.Errorf("failed to start grpc server: %v", err)
+		}
+		close(grpcServerCloseCh)
+	}()
+
+	streamServerCloseCh := make(chan struct{})
+	go func() {
+		if err := criMgr.StreamServerStart(); err != nil {
+			logrus.Errorf("failed to start stream server: %v", err)
+		}
+		close(streamServerCloseCh)
+	}()
+
+	<-streamServerCloseCh
+	logrus.Infof("CRI Stream server stopped")
+	<-grpcServerCloseCh
+	logrus.Infof("CRI GRPC server stopped")
+
+	logrus.Infof("CRI service stopped")
+	return
+}

cri/service/cri.go

@@ -5,7 +5,7 @@ import (
 	"os"
 	"syscall"
 
-	cri "github.com/alibaba/pouch/cri/src"
+	cri "github.com/alibaba/pouch/cri/v1alpha1"
 	"github.com/alibaba/pouch/daemon/config"
 
 	"google.golang.org/grpc"

cri/stream/request_cache.go

@@ -19,10 +19,10 @@ var (
 	TokenLen = 8
 )
 
-// requestCache caches streaming (exec/attach/port-forward) requests and generates a single-use
+// RequestCache caches streaming (exec/attach/port-forward) requests and generates a single-use
 // random token for their retrieval. The requestCache is used for building streaming URLs without
 // the need to encode every request parameter in the URL.
-type requestCache struct {
+type RequestCache struct {
 	// tokens maps the generate token to the request for fast retrieval.
 	tokens map[string]*list.Element
 	// ll maintains an age-ordered request list for faster garbage collection of expired requests.
@@ -31,24 +31,25 @@ type requestCache struct {
 	lock sync.Mutex
 }
 
-// Type representing an *ExecRequest, *AttachRequest, or *PortForwardRequest.
-type request interface{}
+// Request representing an *ExecRequest, *AttachRequest, or *PortForwardRequest Type.
+type Request interface{}
 
 type cacheEntry struct {
 	token      string
-	req        request
+	req        Request
 	expireTime time.Time
 }
 
-func newRequestCache() *requestCache {
-	return &requestCache{
+// NewRequestCache return a RequestCache
+func NewRequestCache() *RequestCache {
+	return &RequestCache{
 		ll:     list.New(),
 		tokens: make(map[string]*list.Element),
 	}
 }
 
 // Insert the given request into the cache and returns the token used for fetching it out.
-func (c *requestCache) Insert(req request) (token string, err error) {
+func (c *RequestCache) Insert(req Request) (token string, err error) {
 	c.lock.Lock()
 	defer c.lock.Unlock()
 
@@ -69,7 +70,7 @@ func (c *requestCache) Insert(req request) (token string, err error) {
 }
 
 // Consume the token (remove it from the cache) and return the cached request, if found.
-func (c *requestCache) Consume(token string) (req request, found bool) {
+func (c *RequestCache) Consume(token string) (req Request, found bool) {
 	c.lock.Lock()
 	defer c.lock.Unlock()
 	ele, ok := c.tokens[token]
@@ -88,7 +89,7 @@ func (c *requestCache) Consume(token string) (req request, found bool) {
 }
 
 // uniqueToken generates a random URL-safe token and ensures uniqueness.
-func (c *requestCache) uniqueToken() (string, error) {
+func (c *RequestCache) uniqueToken() (string, error) {
 	const maxTries = 10
 	// Number of bytes to be TokenLen when base64 encoded.
 	tokenSize := math.Ceil(float64(TokenLen) * 6 / 8)
@@ -108,7 +109,7 @@ func (c *requestCache) uniqueToken() (string, error) {
 }
 
 // Must be write-locked prior to calling.
-func (c *requestCache) gc() {
+func (c *RequestCache) gc() {
 	now := time.Now()
 	for c.ll.Len() > 0 {
 		oldest := c.ll.Back()

cri/src/cri.go → cri/v1alpha1/cri.go

@@ -1,4 +1,4 @@
-package src
+package v1alpha1
 
 import (
 	"bytes"
@@ -12,7 +12,6 @@ import (
 	"time"
 
 	apitypes "github.com/alibaba/pouch/apis/types"
-	"github.com/alibaba/pouch/cri/stream"
 	"github.com/alibaba/pouch/daemon/config"
 	"github.com/alibaba/pouch/daemon/mgr"
 	"github.com/alibaba/pouch/pkg/errtypes"
@@ -87,7 +86,7 @@ type CriManager struct {
 	CniMgr       CniMgr
 
 	// StreamServer is the stream server of CRI serves container streaming request.
-	StreamServer stream.Server
+	StreamServer Server
 
 	// SandboxBaseDir is the directory used to store sandbox files like /etc/hosts, /etc/resolv.conf, etc.
 	SandboxBaseDir string

cri/src/cri_network.go → cri/v1alpha1/cri_network.go

@@ -1,4 +1,4 @@
-package src
+package v1alpha1
 
 import (
 	"fmt"

cri/src/cri_stream.go → cri/v1alpha1/cri_stream.go

@@ -1,4 +1,4 @@
-package src
+package v1alpha1
 
 import (
 	"bytes"
@@ -11,25 +11,24 @@ import (
 	"time"
 
 	apitypes "github.com/alibaba/pouch/apis/types"
-	"github.com/alibaba/pouch/cri/stream"
 	"github.com/alibaba/pouch/cri/stream/remotecommand"
 	"github.com/alibaba/pouch/daemon/mgr"
 
 	"github.com/sirupsen/logrus"
 )
 
-func newStreamServer(ctrMgr mgr.ContainerMgr, address string, port string) (stream.Server, error) {
-	config := stream.DefaultConfig
+func newStreamServer(ctrMgr mgr.ContainerMgr, address string, port string) (Server, error) {
+	config := DefaultConfig
 	config.Address = net.JoinHostPort(address, port)
 	runtime := newStreamRuntime(ctrMgr)
-	return stream.NewServer(config, runtime)
+	return NewServer(config, runtime)
 }
 
 type streamRuntime struct {
 	containerMgr mgr.ContainerMgr
 }
 
-func newStreamRuntime(ctrMgr mgr.ContainerMgr) stream.Runtime {
+func newStreamRuntime(ctrMgr mgr.ContainerMgr) Runtime {
 	return &streamRuntime{containerMgr: ctrMgr}
 }
 

cri/src/cri_types.go → cri/v1alpha1/cri_types.go

@@ -1,4 +1,4 @@
-package src
+package v1alpha1
 
 import (
 	"k8s.io/kubernetes/pkg/kubelet/apis/cri/v1alpha1/runtime"

cri/src/cri_utils.go → cri/v1alpha1/cri_utils.go

@@ -1,4 +1,4 @@
-package src
+package v1alpha1
 
 import (
 	"bytes"

cri/src/cri_utils_test.go → cri/v1alpha1/cri_utils_test.go

@@ -1,4 +1,4 @@
-package src
+package v1alpha1
 
 import (
 	"fmt"

cri/src/cri_wrapper.go → cri/v1alpha1/cri_wrapper.go

@@ -1,4 +1,4 @@
-package src
+package v1alpha1
 
 import (
 	"github.com/sirupsen/logrus"

cri/stream/server.go → cri/v1alpha1/server.go

@@ -1,4 +1,4 @@
-package stream
+package v1alpha1
 
 import (
 	"io"
@@ -7,6 +7,7 @@ import (
 	"path"
 	"time"
 
+	"github.com/alibaba/pouch/cri/stream"
 	"github.com/alibaba/pouch/cri/stream/constant"
 	"github.com/alibaba/pouch/cri/stream/portforward"
 	"github.com/alibaba/pouch/cri/stream/remotecommand"
@@ -92,7 +93,7 @@ var DefaultConfig = Config{
 type server struct {
 	config  Config
 	runtime Runtime
-	cache   *requestCache
+	cache   *stream.RequestCache
 	server  *http.Server
 }
 
@@ -101,7 +102,7 @@ func NewServer(config Config, runtime Runtime) (Server, error) {
 	s := &server{
 		config:  config,
 		runtime: runtime,
-		cache:   newRequestCache(),
+		cache:   stream.NewRequestCache(),
 	}
 
 	if s.config.BaseURL == nil {

cri/v1alpha1/service/cri.go

@@ -0,0 +1,51 @@
+package service
+
+import (
+	"net"
+	"os"
+	"syscall"
+
+	cri "github.com/alibaba/pouch/cri/v1alpha1"
+	"github.com/alibaba/pouch/daemon/config"
+
+	"google.golang.org/grpc"
+	"k8s.io/kubernetes/pkg/kubelet/apis/cri/v1alpha1/runtime"
+)
+
+// Service serves the kubelet runtime grpc api which will be consumed by kubelet.
+type Service struct {
+	config *config.Config
+	server *grpc.Server
+	criMgr cri.CriMgr
+}
+
+// NewService creates a brand new cri service.
+func NewService(cfg *config.Config, criMgr cri.CriMgr) (*Service, error) {
+	s := &Service{
+		config: cfg,
+		server: grpc.NewServer(),
+		criMgr: criMgr,
+	}
+
+	// TODO: Prepare streaming server.
+
+	runtime.RegisterRuntimeServiceServer(s.server, s.criMgr)
+	runtime.RegisterImageServiceServer(s.server, s.criMgr)
+
+	return s, nil
+}
+
+// Serve starts grpc server.
+func (s *Service) Serve() error {
+	// Unlink to cleanup the previous socket file.
+	if err := syscall.Unlink(s.config.CriConfig.Listen); err != nil && !os.IsNotExist(err) {
+		return err
+	}
+
+	l, err := net.Listen("unix", s.config.CriConfig.Listen)
+	if err != nil {
+		return err
+	}
+
+	return s.server.Serve(l)
+}