Skip to content
Permalink
Browse files

Split snapshot controller using beta APIs

  • Loading branch information
xing-yang committed Oct 19, 2019
1 parent 89889f0 commit bf9a419620d414035fa92ea6af655f1cbb7d3ff8
Showing with 3,019 additions and 1,784 deletions.
  1. +0 −6 Dockerfile
  2. +2 −2 Makefile
  3. +6 −0 cmd/csi-snapshotter-common/Dockerfile
  4. +161 −0 cmd/csi-snapshotter-common/main.go
  5. 0 cmd/{csi-snapshotter → csi-snapshotter-common}/main_test.go
  6. +6 −0 cmd/csi-snapshotter-sidecar/Dockerfile
  7. +17 −33 cmd/{csi-snapshotter → csi-snapshotter-sidecar}/main.go
  8. +161 −0 cmd/csi-snapshotter-sidecar/main_test.go
  9. +26 −0 deploy/kubernetes/common-snapshot-controller.yaml
  10. +83 −0 deploy/kubernetes/rbac-common.yaml
  11. +0 −18 deploy/kubernetes/{rbac.yaml → rbac-sidecar.yaml}
  12. +9 −10 deploy/kubernetes/setup-csi-snapshotter.yaml
  13. +3 −4 examples/kubernetes/snapshot.yaml
  14. +3 −2 examples/kubernetes/snapshotclass.yaml
  15. +1 −1 examples/kubernetes/storageclass.yaml
  16. +77 −61 pkg/{controller → common_controller}/framework_test.go
  17. +1,237 −0 pkg/common_controller/snapshot_controller.go
  18. +36 −53 pkg/{controller → common_controller}/snapshot_controller_base.go
  19. +6 −5 pkg/{controller → common_controller}/snapshot_controller_test.go
  20. +59 −100 pkg/{controller → common_controller}/snapshot_create_test.go
  21. +89 −88 pkg/{controller → common_controller}/snapshot_delete_test.go
  22. +2 −2 pkg/{controller → common_controller}/snapshot_finalizer_test.go
  23. +46 −45 pkg/{controller → common_controller}/snapshot_ready_test.go
  24. +0 −1,226 pkg/controller/snapshot_controller.go
  25. +38 −11 pkg/{controller → sidecar_controller}/csi_handler.go
  26. +542 −0 pkg/sidecar_controller/snapshot_controller.go
  27. +280 −0 pkg/sidecar_controller/snapshot_controller_base.go
  28. +5 −8 pkg/snapshotter/snapshotter.go
  29. +7 −42 pkg/snapshotter/snapshotter_test.go
  30. +98 −32 pkg/{controller → utils}/util.go
  31. +16 −32 pkg/{controller → utils}/util_test.go
  32. +3 −3 vendor/modules.txt

This file was deleted.

@@ -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
@@ -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"]
@@ -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"
"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
}
File renamed without changes.
@@ -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"]
@@ -1,5 +1,5 @@
/*
Copyright 2018 The Kubernetes Authors.
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.
@@ -37,7 +37,7 @@ import (
"github.com/kubernetes-csi/csi-lib-utils/connection"
"github.com/kubernetes-csi/csi-lib-utils/leaderelection"
csirpc "github.com/kubernetes-csi/csi-lib-utils/rpc"
"github.com/kubernetes-csi/external-snapshotter/pkg/controller"
controller "github.com/kubernetes-csi/external-snapshotter/pkg/sidecar_controller"
"github.com/kubernetes-csi/external-snapshotter/pkg/snapshotter"

clientset "github.com/kubernetes-csi/external-snapshotter/pkg/client/clientset/versioned"
@@ -56,17 +56,13 @@ const (

// Command line flags
var (
snapshotterName = flag.String("snapshotter", "", "This option is deprecated.")
kubeconfig = flag.String("kubeconfig", "", "Absolute path to the kubeconfig file. Required only when running out of cluster.")
connectionTimeout = flag.Duration("connection-timeout", 0, "The --connection-timeout flag is deprecated")
csiAddress = flag.String("csi-address", "/run/csi/socket", "Address of the CSI driver socket.")
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.")
snapshotNamePrefix = flag.String("snapshot-name-prefix", "snapshot", "Prefix to apply to the name of a created snapshot")
snapshotNameUUIDLength = flag.Int("snapshot-name-uuid-length", -1, "Length in characters for the generated uuid of a created snapshot. Defaults behavior is to NOT truncate.")
showVersion = flag.Bool("version", false, "Show version.")
csiTimeout = flag.Duration("timeout", defaultCSITimeout, "The timeout for any RPCs to the CSI driver. Default is 1 minute.")
kubeconfig = flag.String("kubeconfig", "", "Absolute path to the kubeconfig file. Required only when running out of cluster.")
csiAddress = flag.String("csi-address", "/run/csi/socket", "Address of the CSI driver socket.")
resyncPeriod = flag.Duration("resync-period", 60*time.Second, "Resync interval of the controller.")
snapshotNamePrefix = flag.String("snapshot-name-prefix", "snapshot", "Prefix to apply to the name of a created snapshot")
snapshotNameUUIDLength = flag.Int("snapshot-name-uuid-length", -1, "Length in characters for the generated uuid of a created snapshot. Defaults behavior is to NOT truncate.")
showVersion = flag.Bool("version", false, "Show version.")
csiTimeout = flag.Duration("timeout", defaultCSITimeout, "The timeout for any RPCs to the CSI driver. Default is 1 minute.")

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.")
@@ -88,14 +84,6 @@ func main() {
}
klog.Infof("Version: %s", version)

if *connectionTimeout != 0 {
klog.Warning("--connection-timeout is deprecated and will have no effect")
}

if *snapshotterName != "" {
klog.Warning("--snapshotter is deprecated and will have no effect")
}

// Create the client config. Use kubeconfig if given, otherwise assume in-cluster.
config, err := buildConfig(*kubeconfig)
if err != nil {
@@ -122,7 +110,7 @@ func main() {
snapshotscheme.AddToScheme(scheme.Scheme)

// Connect to CSI.
csiConn, err := connection.Connect(*csiAddress, connection.OnConnectionLoss(connection.ExitOnConnectionLoss()))
csiConn, err := connection.Connect(*csiAddress)
if err != nil {
klog.Errorf("error connecting to CSI driver: %v", err)
os.Exit(1)
@@ -133,13 +121,13 @@ func main() {
defer cancel()

// Find driver name
*snapshotterName, err = csirpc.GetDriverName(ctx, csiConn)
snapshotterName, err := csirpc.GetDriverName(ctx, csiConn)
if err != nil {
klog.Errorf("error getting CSI driver name: %v", err)
os.Exit(1)
}

klog.V(2).Infof("CSI driver name: %q", *snapshotterName)
klog.V(2).Infof("CSI driver name: %q", snapshotterName)

// Check it's ready
if err = csirpc.ProbeForever(csiConn, *csiTimeout); err != nil {
@@ -154,7 +142,7 @@ func main() {
os.Exit(1)
}
if !supportsCreateSnapshot {
klog.Errorf("CSI driver %s does not support ControllerCreateSnapshot", *snapshotterName)
klog.Errorf("CSI driver %s does not support ControllerCreateSnapshot", snapshotterName)
os.Exit(1)
}

@@ -163,19 +151,15 @@ func main() {
os.Exit(1)
}

klog.V(2).Infof("Start NewCSISnapshotController with snapshotter [%s] kubeconfig [%s] csiTimeout [%+v] csiAddress [%s] createSnapshotContentRetryCount [%d] createSnapshotContentInterval [%+v] resyncPeriod [%+v] snapshotNamePrefix [%s] snapshotNameUUIDLength [%d]", *snapshotterName, *kubeconfig, *csiTimeout, *csiAddress, createSnapshotContentRetryCount, *createSnapshotContentInterval, *resyncPeriod, *snapshotNamePrefix, snapshotNameUUIDLength)
klog.V(2).Infof("Start NewCSISnapshotSideCarController with snapshotter [%s] kubeconfig [%s] csiTimeout [%+v] csiAddress [%s] resyncPeriod [%+v] snapshotNamePrefix [%s] snapshotNameUUIDLength [%d]", snapshotterName, *kubeconfig, *csiTimeout, *csiAddress, *resyncPeriod, *snapshotNamePrefix, snapshotNameUUIDLength)

snapShotter := snapshotter.NewSnapshotter(csiConn)
ctrl := controller.NewCSISnapshotController(
ctrl := controller.NewCSISnapshotSideCarController(
snapClient,
kubeClient,
*snapshotterName,
factory.Snapshot().V1beta1().VolumeSnapshots(),
snapshotterName,
factory.Snapshot().V1beta1().VolumeSnapshotContents(),
factory.Snapshot().V1beta1().VolumeSnapshotClasses(),
coreFactory.Core().V1().PersistentVolumeClaims(),
*createSnapshotContentRetryCount,
*createSnapshotContentInterval,
snapShotter,
*csiTimeout,
*resyncPeriod,
@@ -200,7 +184,7 @@ func main() {
if !*leaderElection {
run(context.TODO())
} else {
lockName := fmt.Sprintf("%s-%s", prefix, strings.Replace(*snapshotterName, "/", "-", -1))
lockName := fmt.Sprintf("%s-%s", prefix, strings.Replace(snapshotterName, "/", "-", -1))
le := leaderelection.NewLeaderElection(kubeClient, lockName, run)
if *leaderElectionNamespace != "" {
le.WithNamespace(*leaderElectionNamespace)

0 comments on commit bf9a419

Please sign in to comment.
You can’t perform that action at this time.