Split snapshot controller using beta APIs

Dockerfile_common

@@ -0,0 +1,6 @@
+FROM gcr.io/distroless/static:latest
+LABEL maintainers="Kubernetes Authors"
+LABEL description="CSI External Snapshotter Common"
+
+COPY ./bin/csi-snapshotter-common csi-snapshotter-common
+ENTRYPOINT ["/csi-snapshotter-common"]

Dockerfile_sidecar

@@ -0,0 +1,6 @@
+FROM gcr.io/distroless/static:latest
+LABEL maintainers="Kubernetes Authors"
+LABEL description="CSI External Snapshotter Sidecar"
+
+COPY ./bin/csi-snapshotter-sidecar csi-snapshotter-sidecar
+ENTRYPOINT ["/csi-snapshotter-sidecar"]

Makefile

@@ -12,8 +12,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-.PHONY: all csi-snapshotter clean test
+.PHONY: all csi-snapshotter-common csi-snapshotter-sidecar clean test
 
-CMDS=csi-snapshotter
+CMDS=csi-snapshotter-common csi-snapshotter-sidecar
 all: build
 include release-tools/build.make

Makefile_old

@@ -0,0 +1,19 @@
+# Copyright 2018 The Kubernetes 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.
+
+.PHONY: all csi-snapshotter clean test
+
+CMDS=csi-snapshotter
+all: build
+include release-tools/build.make

cmd/csi-snapshotter-common/main.go

@@ -0,0 +1,161 @@
+/*
+Copyright 2018 The Kubernetes 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 main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"os"
+	"os/signal"
+	"strings"
+	"time"
+
+	"google.golang.org/grpc"
+
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/kubernetes/scheme"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/clientcmd"
+	"k8s.io/klog"
+
+	"github.com/container-storage-interface/spec/lib/go/csi"
+	"github.com/kubernetes-csi/csi-lib-utils/leaderelection"
+	csirpc "github.com/kubernetes-csi/csi-lib-utils/rpc"
+	controller "github.com/kubernetes-csi/external-snapshotter/pkg/common_controller"
+
+	clientset "github.com/kubernetes-csi/external-snapshotter/pkg/client/clientset/versioned"
+	snapshotscheme "github.com/kubernetes-csi/external-snapshotter/pkg/client/clientset/versioned/scheme"
+	informers "github.com/kubernetes-csi/external-snapshotter/pkg/client/informers/externalversions"
+	coreinformers "k8s.io/client-go/informers"
+)
+
+const (
+	// Number of worker threads
+	threads = 10
+)
+
+// Command line flags
+var (
+	kubeconfig                      = flag.String("kubeconfig", "", "Absolute path to the kubeconfig file. Required only when running out of cluster.")
+	createSnapshotContentRetryCount = flag.Int("create-snapshotcontent-retrycount", 5, "Number of retries when we create a snapshot content object for a snapshot.")
+	createSnapshotContentInterval   = flag.Duration("create-snapshotcontent-interval", 10*time.Second, "Interval between retries when we create a snapshot content object for a snapshot.")
+	resyncPeriod                    = flag.Duration("resync-period", 60*time.Second, "Resync interval of the controller.")
+	showVersion                     = flag.Bool("version", false, "Show version.")
+
+	leaderElection          = flag.Bool("leader-election", false, "Enables leader election.")
+	leaderElectionNamespace = flag.String("leader-election-namespace", "", "The namespace where the leader election resource exists. Defaults to the pod namespace if not set.")
+)
+
+var (
+	version = "unknown"
+	prefix  = "external-snapshotter-leader"
+)
+
+func main() {
+	klog.InitFlags(nil)
+	flag.Set("logtostderr", "true")
+	flag.Parse()
+
+	if *showVersion {
+		fmt.Println(os.Args[0], version)
+		os.Exit(0)
+	}
+	klog.Infof("Version: %s", version)
+
+	// Create the client config. Use kubeconfig if given, otherwise assume in-cluster.
+	config, err := buildConfig(*kubeconfig)
+	if err != nil {
+		klog.Error(err.Error())
+		os.Exit(1)
+	}
+
+	kubeClient, err := kubernetes.NewForConfig(config)
+	if err != nil {
+		klog.Error(err.Error())
+		os.Exit(1)
+	}
+
+	snapClient, err := clientset.NewForConfig(config)
+	if err != nil {
+		klog.Errorf("Error building snapshot clientset: %s", err.Error())
+		os.Exit(1)
+	}
+
+	factory := informers.NewSharedInformerFactory(snapClient, *resyncPeriod)
+	coreFactory := coreinformers.NewSharedInformerFactory(kubeClient, *resyncPeriod)
+
+	// Add Snapshot types to the defualt Kubernetes so events can be logged for them
+	snapshotscheme.AddToScheme(scheme.Scheme)
+
+	klog.V(2).Infof("Start NewCSISnapshotController with kubeconfig [%s] createSnapshotContentRetryCount [%d] createSnapshotContentInterval [%d] resyncPeriod [%+v]", *kubeconfig, *createSnapshotContentRetryCount, *createSnapshotContentInterval, *resyncPeriod)
+
+	ctrl := controller.NewCSISnapshotCommonController(
+		snapClient,
+		kubeClient,
+		factory.Snapshot().V1beta1().VolumeSnapshots(),
+		factory.Snapshot().V1beta1().VolumeSnapshotContents(),
+		factory.Snapshot().V1beta1().VolumeSnapshotClasses(),
+		coreFactory.Core().V1().PersistentVolumeClaims(),
+		*createSnapshotContentRetryCount,
+		*createSnapshotContentInterval,
+		*resyncPeriod,
+	)
+
+	run := func(context.Context) {
+		// run...
+		stopCh := make(chan struct{})
+		factory.Start(stopCh)
+		coreFactory.Start(stopCh)
+		go ctrl.Run(threads, stopCh)
+
+		// ...until SIGINT
+		c := make(chan os.Signal, 1)
+		signal.Notify(c, os.Interrupt)
+		<-c
+		close(stopCh)
+	}
+
+	if !*leaderElection {
+		run(context.TODO())
+	} else {
+		lockName := fmt.Sprintf("%s-%s", prefix, strings.Replace("common-snapshotter", "/", "-", -1))
+		le := leaderelection.NewLeaderElection(kubeClient, lockName, run)
+		if *leaderElectionNamespace != "" {
+			le.WithNamespace(*leaderElectionNamespace)
+		}
+		if err := le.Run(); err != nil {
+			klog.Fatalf("failed to initialize leader election: %v", err)
+		}
+	}
+}
+
+func buildConfig(kubeconfig string) (*rest.Config, error) {
+	if kubeconfig != "" {
+		return clientcmd.BuildConfigFromFlags("", kubeconfig)
+	}
+	return rest.InClusterConfig()
+}
+
+func supportsControllerCreateSnapshot(ctx context.Context, conn *grpc.ClientConn) (bool, error) {
+	capabilities, err := csirpc.GetControllerCapabilities(ctx, conn)
+	if err != nil {
+		return false, err
+	}
+
+	return capabilities[csi.ControllerServiceCapability_RPC_CREATE_DELETE_SNAPSHOT], nil
+}

cmd/csi-snapshotter-common/main_test.go

@@ -0,0 +1,161 @@
+/*
+Copyright 2019 The Kubernetes 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 main
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/container-storage-interface/spec/lib/go/csi"
+	"github.com/golang/mock/gomock"
+	"github.com/kubernetes-csi/csi-lib-utils/connection"
+	"github.com/kubernetes-csi/csi-test/driver"
+
+	"google.golang.org/grpc"
+)
+
+func Test_supportsControllerCreateSnapshot(t *testing.T) {
+	tests := []struct {
+		name         string
+		output       *csi.ControllerGetCapabilitiesResponse
+		injectError  bool
+		expectError  bool
+		expectResult bool
+	}{
+		{
+			name: "success",
+			output: &csi.ControllerGetCapabilitiesResponse{
+				Capabilities: []*csi.ControllerServiceCapability{
+					{
+						Type: &csi.ControllerServiceCapability_Rpc{
+							Rpc: &csi.ControllerServiceCapability_RPC{
+								Type: csi.ControllerServiceCapability_RPC_CREATE_DELETE_VOLUME,
+							},
+						},
+					},
+					{
+						Type: &csi.ControllerServiceCapability_Rpc{
+							Rpc: &csi.ControllerServiceCapability_RPC{
+								Type: csi.ControllerServiceCapability_RPC_CREATE_DELETE_SNAPSHOT,
+							},
+						},
+					},
+				},
+			},
+			expectError:  false,
+			expectResult: true,
+		},
+		{
+			name:         "gRPC error",
+			output:       nil,
+			injectError:  true,
+			expectError:  true,
+			expectResult: false,
+		},
+		{
+			name: "no create snapshot",
+			output: &csi.ControllerGetCapabilitiesResponse{
+				Capabilities: []*csi.ControllerServiceCapability{
+					{
+						Type: &csi.ControllerServiceCapability_Rpc{
+							Rpc: &csi.ControllerServiceCapability_RPC{
+								Type: csi.ControllerServiceCapability_RPC_CREATE_DELETE_VOLUME,
+							},
+						},
+					},
+				},
+			},
+			expectError:  false,
+			expectResult: false,
+		},
+		{
+			name: "empty capability",
+			output: &csi.ControllerGetCapabilitiesResponse{
+				Capabilities: []*csi.ControllerServiceCapability{
+					{
+						Type: nil,
+					},
+				},
+			},
+			expectError:  false,
+			expectResult: false,
+		},
+		{
+			name: "no capabilities",
+			output: &csi.ControllerGetCapabilitiesResponse{
+				Capabilities: []*csi.ControllerServiceCapability{},
+			},
+			expectError:  false,
+			expectResult: false,
+		},
+	}
+
+	mockController, driver, _, controllerServer, csiConn, err := createMockServer(t)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer mockController.Finish()
+	defer driver.Stop()
+	defer csiConn.Close()
+
+	for _, test := range tests {
+
+		in := &csi.ControllerGetCapabilitiesRequest{}
+
+		out := test.output
+		var injectedErr error
+		if test.injectError {
+			injectedErr = fmt.Errorf("mock error")
+		}
+
+		// Setup expectation
+		controllerServer.EXPECT().ControllerGetCapabilities(gomock.Any(), in).Return(out, injectedErr).Times(1)
+
+		ok, err := supportsControllerCreateSnapshot(context.Background(), csiConn)
+		if test.expectError && err == nil {
+			t.Errorf("test %q: Expected error, got none", test.name)
+		}
+		if !test.expectError && err != nil {
+			t.Errorf("test %q: got error: %v", test.name, err)
+		}
+		if err == nil && test.expectResult != ok {
+			t.Errorf("test fail expected result %t but got %t\n", test.expectResult, ok)
+		}
+	}
+}
+
+func createMockServer(t *testing.T) (*gomock.Controller, *driver.MockCSIDriver, *driver.MockIdentityServer, *driver.MockControllerServer, *grpc.ClientConn, error) {
+	// Start the mock server
+	mockController := gomock.NewController(t)
+	identityServer := driver.NewMockIdentityServer(mockController)
+	controllerServer := driver.NewMockControllerServer(mockController)
+	drv := driver.NewMockCSIDriver(&driver.MockCSIDriverServers{
+		Identity:   identityServer,
+		Controller: controllerServer,
+	})
+	drv.Start()
+
+	// Create a client connection to it
+	addr := drv.Address()
+	csiConn, err := connection.Connect(addr)
+	if err != nil {
+		return nil, nil, nil, nil, nil, err
+	}
+
+	return mockController, drv, identityServer, controllerServer, csiConn, nil
+}