Split snapshot controller using beta APIs
This commit is contained in:
xing-yang
6
cmd/snapshot-controller/Dockerfile
Normal file
6
cmd/snapshot-controller/Dockerfile
Normal file
@@ -0,0 +1,6 @@
|
||||
FROM gcr.io/distroless/static:latest
|
||||
LABEL maintainers="Kubernetes Authors"
|
||||
LABEL description="Snapshot Controller"
|
||||
|
||||
COPY ./bin/snapshot-controller snapshot-controller
|
||||
ENTRYPOINT ["/snapshot-controller"]
|
||||
159
cmd/snapshot-controller/main.go
Normal file
159
cmd/snapshot-controller/main.go
Normal file
@@ -0,0 +1,159 @@
|
||||
/*
|
||||
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"
|
||||
"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"
|
||||
)
|
||||
|
||||
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 := "snapshot-controller-leader"
|
||||
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
|
||||
}
|
||||
161
cmd/snapshot-controller/main_test.go
Normal file
161
cmd/snapshot-controller/main_test.go
Normal file
@@ -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
|
||||
}
|
||||
Reference in New Issue
Block a user