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 7261d2e5e10d29d3119b3ba1fb73b8f130578699
Showing with 2,412 additions and 1,157 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. +10 −22 cmd/{csi-snapshotter → csi-snapshotter-sidecar}/main.go
  8. +161 −0 cmd/csi-snapshotter-sidecar/main_test.go
  9. +40 −0 deploy/kubernetes/common-csi-snapshotter.yaml
  10. +83 −0 deploy/kubernetes/rbac-common.yaml
  11. +0 −3 deploy/kubernetes/rbac.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. +74 −58 pkg/{controller → common_controller}/framework_test.go
  17. +595 −634 pkg/{controller → 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. +84 −83 pkg/{controller → common_controller}/snapshot_delete_test.go
  22. +1 −1 pkg/{controller → common_controller}/snapshot_finalizer_test.go
  23. +46 −45 pkg/{controller → common_controller}/snapshot_ready_test.go
  24. +38 −11 pkg/{controller → sidecar_controller}/csi_handler.go
  25. +576 −0 pkg/sidecar_controller/snapshot_controller.go
  26. +295 −0 pkg/sidecar_controller/snapshot_controller_base.go
  27. +5 −8 pkg/snapshotter/snapshotter.go
  28. +7 −42 pkg/snapshotter/snapshotter_test.go
  29. +86 −32 pkg/{controller → utils}/util.go
  30. +16 −32 pkg/{controller → utils}/util_test.go
  31. +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,9 +56,7 @@ 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.")
@@ -88,14 +86,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 +112,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 +123,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 +144,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,17 +153,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] createSnapshotContentRetryCount [%d] createSnapshotContentInterval [%+v] resyncPeriod [%+v] snapshotNamePrefix [%s] snapshotNameUUIDLength [%d]", snapshotterName, *kubeconfig, *csiTimeout, *csiAddress, createSnapshotContentRetryCount, *createSnapshotContentInterval, *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,
@@ -200,7 +188,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 7261d2e

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