Skip to content

Commit

Permalink
Split snapshot controller using beta APIs
Browse files Browse the repository at this point in the history
  • Loading branch information
xing-yang committed Nov 5, 2019
1 parent 89889f0 commit 4954e80
Show file tree
Hide file tree
Showing 32 changed files with 3,081 additions and 1,822 deletions.
6 changes: 0 additions & 6 deletions Dockerfile

This file was deleted.

4 changes: 2 additions & 2 deletions Makefile
Original file line number Diff line number Diff line change
Expand Up @@ -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
6 changes: 6 additions & 0 deletions cmd/csi-snapshotter-common/Dockerfile
Original file line number Diff line number Diff line change
@@ -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"]
161 changes: 161 additions & 0 deletions cmd/csi-snapshotter-common/main.go
Original file line number Diff line number Diff line change
@@ -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.
6 changes: 6 additions & 0 deletions cmd/csi-snapshotter-sidecar/Dockerfile
Original file line number Diff line number Diff line change
@@ -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"]
50 changes: 17 additions & 33 deletions cmd/csi-snapshotter/main.go → cmd/csi-snapshotter-sidecar/main.go
Original file line number Diff line number Diff line change
@@ -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.
Expand Down Expand Up @@ -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"
Expand All @@ -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.")
Expand All @@ -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 {
Expand All @@ -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)
Expand All @@ -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 {
Expand All @@ -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)
}

Expand All @@ -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,
Expand All @@ -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)
Expand Down
Loading

0 comments on commit 4954e80

Please sign in to comment.