Merge pull request #1383 from HusterWan/0.5.x

cherry-pick: cherry-pick #1359 from master to 0.5.x

Author: HusterWan

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,114 @@
+package cri
+
+import (
+	"fmt"
+
+	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":
+		err = runv1alpha1(daemonconfig, containerMgr, imageMgr)
+	case "v1alpha2":
+		err = runv1alpha2(daemonconfig, containerMgr, imageMgr)
+	default:
+		err = fmt.Errorf("invalid CRI version,failed to start CRI service")
+	}
+	return
+}
+
+// Start CRI service with CRI version: v1alpha1
+func runv1alpha1(daemonconfig *config.Config, containerMgr mgr.ContainerMgr, imageMgr mgr.ImageMgr) error {
+	logrus.Infof("Start CRI service with CRI version: v1alpha1")
+	criMgr, err := criv1alpha1.NewCriManager(daemonconfig, containerMgr, imageMgr)
+	if err != nil {
+		return fmt.Errorf("failed to get CriManager with error: %v", err)
+	}
+
+	service, err := servicev1alpha1.NewService(daemonconfig, criMgr)
+	if err != nil {
+		return fmt.Errorf("failed to start CRI service with error: %v", err)
+	}
+
+	// TODO: Stop the whole CRI service if any of the critical service exits
+	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)
+	}()
+
+	// TODO: refactor it with select
+	<-streamServerCloseCh
+	logrus.Infof("CRI Stream server stopped")
+	<-grpcServerCloseCh
+	logrus.Infof("CRI GRPC server stopped")
+
+	logrus.Infof("CRI service stopped")
+	return nil
+}
+
+// Start CRI service with CRI version: v1alpha2
+func runv1alpha2(daemonconfig *config.Config, containerMgr mgr.ContainerMgr, imageMgr mgr.ImageMgr) error {
+	logrus.Infof("Start CRI service with CRI version: v1alpha2")
+	criMgr, err := criv1alpha2.NewCriManager(daemonconfig, containerMgr, imageMgr)
+	if err != nil {
+		return fmt.Errorf("failed to get CriManager with error: %v", err)
+	}
+
+	service, err := servicev1alpha2.NewService(daemonconfig, criMgr)
+	if err != nil {
+		return fmt.Errorf("failed to start CRI service with error: %v", err)
+	}
+	// TODO: Stop the whole CRI service if any of the critical service exits
+	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)
+	}()
+	// TODO: refactor it with select
+	<-streamServerCloseCh
+	logrus.Infof("CRI Stream server stopped")
+	<-grpcServerCloseCh
+	logrus.Infof("CRI GRPC server stopped")
+
+	logrus.Infof("CRI service stopped")
+	return nil
+}

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 renamed to 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 renamed to cri/v1alpha1/cri_network.go

@@ -1,4 +1,4 @@
-package src
+package v1alpha1
 
 import (
 	"fmt"
@@ -78,9 +78,6 @@ func (c *CniManager) Name() string {
 // are launched.
 func (c *CniManager) SetUpPodNetwork(podNetwork *ocicni.PodNetwork) error {
 	_, err := c.plugin.SetUpPod(*podNetwork)
-	if err != nil {
-		return fmt.Errorf("failed to setup network for sandbox %q: %v", podNetwork.ID, err)
-	}
 
 	defer func() {
 		if err != nil {
@@ -92,6 +89,10 @@ func (c *CniManager) SetUpPodNetwork(podNetwork *ocicni.PodNetwork) error {
 		}
 	}()
 
+	if err != nil {
+		return fmt.Errorf("failed to setup network for sandbox %q: %v", podNetwork.ID, err)
+	}
+
 	return nil
 }
 

cri/src/cri_stream.go renamed to 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}
 }
 
@@ -61,6 +60,7 @@ func (s *streamRuntime) Exec(containerID string, cmd []string, streamOpts *remot
 		return 0, fmt.Errorf("failed to start exec for container %q: %v", containerID, err)
 	}
 
+	// TODO Find a better way instead of the dead loop
 	var ei *apitypes.ContainerExecInspect
 	for {
 		ei, err = s.containerMgr.InspectExec(ctx, execid)

cri/src/cri_types.go renamed to 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 renamed to cri/v1alpha1/cri_utils.go

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

cri/src/cri_utils_test.go renamed to cri/v1alpha1/cri_utils_test.go

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

cri/src/cri_wrapper.go renamed to cri/v1alpha1/cri_wrapper.go

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