diff --git a/cmd/csi-snapshotter/create_crd.go b/cmd/csi-snapshotter/create_crd.go new file mode 100644 index 00000000..9e7d77a6 --- /dev/null +++ b/cmd/csi-snapshotter/create_crd.go @@ -0,0 +1,122 @@ +/* +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 ( + "reflect" + + "github.com/golang/glog" + crdv1 "github.com/kubernetes-csi/external-snapshotter/pkg/apis/volumesnapshot/v1alpha1" + apiextensionsv1beta1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1" + apiextensionsclient "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/serializer" + "k8s.io/client-go/rest" +) + +const ( + // SnapshotPVCAnnotation is "snapshot.alpha.kubernetes.io/snapshot" + SnapshotPVCAnnotation = "volumesnapshot.csi.k8s.io/snapshot" +) + +// NewClient creates a new RESTClient +func NewClient(cfg *rest.Config) (*rest.RESTClient, *runtime.Scheme, error) { + scheme := runtime.NewScheme() + if err := crdv1.AddToScheme(scheme); err != nil { + return nil, nil, err + } + + config := *cfg + config.GroupVersion = &crdv1.SchemeGroupVersion + config.APIPath = "/apis" + config.ContentType = runtime.ContentTypeJSON + config.NegotiatedSerializer = serializer.DirectCodecFactory{CodecFactory: serializer.NewCodecFactory(scheme)} + + client, err := rest.RESTClientFor(&config) + if err != nil { + return nil, nil, err + } + + return client, scheme, nil +} + +// CreateCRD creates CustomResourceDefinition +func CreateCRD(clientset apiextensionsclient.Interface) error { + crd := &apiextensionsv1beta1.CustomResourceDefinition{ + ObjectMeta: metav1.ObjectMeta{ + Name: crdv1.VolumeSnapshotClassResourcePlural + "." + crdv1.GroupName, + }, + Spec: apiextensionsv1beta1.CustomResourceDefinitionSpec{ + Group: crdv1.GroupName, + Version: crdv1.SchemeGroupVersion.Version, + Scope: apiextensionsv1beta1.ClusterScoped, + Names: apiextensionsv1beta1.CustomResourceDefinitionNames{ + Plural: crdv1.VolumeSnapshotClassResourcePlural, + Kind: reflect.TypeOf(crdv1.VolumeSnapshotClass{}).Name(), + }, + }, + } + res, err := clientset.ApiextensionsV1beta1().CustomResourceDefinitions().Create(crd) + + if err != nil && !apierrors.IsAlreadyExists(err) { + glog.Fatalf("failed to create VolumeSnapshotResource: %#v, err: %#v", + res, err) + } + + crd = &apiextensionsv1beta1.CustomResourceDefinition{ + ObjectMeta: metav1.ObjectMeta{ + Name: crdv1.VolumeSnapshotContentResourcePlural + "." + crdv1.GroupName, + }, + Spec: apiextensionsv1beta1.CustomResourceDefinitionSpec{ + Group: crdv1.GroupName, + Version: crdv1.SchemeGroupVersion.Version, + Scope: apiextensionsv1beta1.ClusterScoped, + Names: apiextensionsv1beta1.CustomResourceDefinitionNames{ + Plural: crdv1.VolumeSnapshotContentResourcePlural, + Kind: reflect.TypeOf(crdv1.VolumeSnapshotContent{}).Name(), + }, + }, + } + res, err = clientset.ApiextensionsV1beta1().CustomResourceDefinitions().Create(crd) + + if err != nil && !apierrors.IsAlreadyExists(err) { + glog.Fatalf("failed to create VolumeSnapshotContentResource: %#v, err: %#v", + res, err) + } + + crd = &apiextensionsv1beta1.CustomResourceDefinition{ + ObjectMeta: metav1.ObjectMeta{ + Name: crdv1.VolumeSnapshotResourcePlural + "." + crdv1.GroupName, + }, + Spec: apiextensionsv1beta1.CustomResourceDefinitionSpec{ + Group: crdv1.GroupName, + Version: crdv1.SchemeGroupVersion.Version, + Scope: apiextensionsv1beta1.NamespaceScoped, + Names: apiextensionsv1beta1.CustomResourceDefinitionNames{ + Plural: crdv1.VolumeSnapshotResourcePlural, + Kind: reflect.TypeOf(crdv1.VolumeSnapshot{}).Name(), + }, + }, + } + res, err = clientset.ApiextensionsV1beta1().CustomResourceDefinitions().Create(crd) + + if err != nil && !apierrors.IsAlreadyExists(err) { + glog.Fatalf("failed to create VolumeSnapshotResource: %#v, err: %#v", + res, err) + } + + return nil +} diff --git a/cmd/csi-snapshotter/main.go b/cmd/csi-snapshotter/main.go index 50e8d8d3..327a695e 100644 --- a/cmd/csi-snapshotter/main.go +++ b/cmd/csi-snapshotter/main.go @@ -1,7 +1,200 @@ +/* +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 "fmt" +import ( + "context" + "flag" + "fmt" + "os" + "os/signal" + "time" + + "github.com/golang/glog" + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/rest" + "k8s.io/client-go/tools/clientcmd" + + "github.com/kubernetes-csi/external-snapshotter/pkg/connection" + "github.com/kubernetes-csi/external-snapshotter/pkg/controller" + + clientset "github.com/kubernetes-csi/external-snapshotter/pkg/client/clientset/versioned" + informers "github.com/kubernetes-csi/external-snapshotter/pkg/client/informers/externalversions" + apiextensionsclient "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset" +) + +const ( + // Number of worker threads + threads = 10 + + // Default timeout of short CSI calls like GetPluginInfo + csiTimeout = time.Second +) + +// Command line flags +var ( + snapshotter = flag.String("snapshotter", "", "Name of the snapshotter. The snapshotter will only create snapshot content for snapshot that requests a VolumeSnapshotClass with a snapshotter field set equal to this name.") + kubeconfig = flag.String("kubeconfig", "", "Absolute path to the kubeconfig file. Required only when running out of cluster.") + connectionTimeout = flag.Duration("connection-timeout", 1*time.Minute, "Timeout for waiting for CSI driver socket.") + 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.") +) func main() { - fmt.Println("vim-go") + flag.Set("logtostderr", "true") + flag.Parse() + + // Create the client config. Use kubeconfig if given, otherwise assume in-cluster. + config, err := buildConfig(*kubeconfig) + if err != nil { + glog.Error(err.Error()) + os.Exit(1) + } + + kubeClient, err := kubernetes.NewForConfig(config) + if err != nil { + glog.Error(err.Error()) + os.Exit(1) + } + + snapClient, err := clientset.NewForConfig(config) + if err != nil { + glog.Errorf("Error building snapshot clientset: %s", err.Error()) + os.Exit(1) + } + + factory := informers.NewSharedInformerFactory(snapClient, *resyncPeriod) + + // Create CRD resource + aeclientset, err := apiextensionsclient.NewForConfig(config) + if err != nil { + glog.Error(err.Error()) + os.Exit(1) + } + + // initialize CRD resource if it does not exist + err = CreateCRD(aeclientset) + if err != nil { + glog.Error(err.Error()) + os.Exit(1) + } + + // Connect to CSI. + csiConn, err := connection.New(*csiAddress, *connectionTimeout) + if err != nil { + glog.Error(err.Error()) + os.Exit(1) + } + + // Pass a context with a timeout + ctx, cancel := context.WithTimeout(context.Background(), csiTimeout) + defer cancel() + + // Find driver name + if *snapshotter == "" { + *snapshotter, err = csiConn.GetDriverName(ctx) + if err != nil { + glog.Error(err.Error()) + os.Exit(1) + } + } + glog.V(2).Infof("CSI driver name: %q", *snapshotter) + + // Check it's ready + if err = waitForDriverReady(csiConn, *connectionTimeout); err != nil { + glog.Error(err.Error()) + os.Exit(1) + } + + // Find out if the driver supports create/delete snapshot. + supportsCreateSnapshot, err := csiConn.SupportsControllerCreateSnapshot(ctx) + if err != nil { + glog.Error(err.Error()) + os.Exit(1) + } + if !supportsCreateSnapshot { + glog.Errorf("CSI driver %s does not support ControllerCreateSnapshot", *snapshotter) + os.Exit(1) + } + + if len(*snapshotNamePrefix) == 0 { + glog.Error("Snapshot name prefix cannot be of length 0") + os.Exit(1) + } + + glog.V(2).Infof("Start NewCSISnapshotController with snapshotter [%s] kubeconfig [%s] connectionTimeout [%+v] csiAddress [%s] createSnapshotContentRetryCount [%d] createSnapshotContentInterval [%+v] resyncPeriod [%+v] snapshotNamePrefix [%s] snapshotNameUUIDLength [%d]", *snapshotter, *kubeconfig, *connectionTimeout, *csiAddress, createSnapshotContentRetryCount, *createSnapshotContentInterval, *resyncPeriod, *snapshotNamePrefix, snapshotNameUUIDLength) + + ctrl := controller.NewCSISnapshotController( + snapClient, + kubeClient, + *snapshotter, + factory.Volumesnapshot().V1alpha1().VolumeSnapshots(), + factory.Volumesnapshot().V1alpha1().VolumeSnapshotContents(), + factory.Volumesnapshot().V1alpha1().VolumeSnapshotClasses(), + *createSnapshotContentRetryCount, + *createSnapshotContentInterval, + csiConn, + *connectionTimeout, + *resyncPeriod, + *snapshotNamePrefix, + *snapshotNameUUIDLength, + ) + + // run... + stopCh := make(chan struct{}) + factory.Start(stopCh) + go ctrl.Run(threads, stopCh) + + // ...until SIGINT + c := make(chan os.Signal, 1) + signal.Notify(c, os.Interrupt) + <-c + close(stopCh) +} + +func buildConfig(kubeconfig string) (*rest.Config, error) { + if kubeconfig != "" { + return clientcmd.BuildConfigFromFlags("", kubeconfig) + } + return rest.InClusterConfig() +} + +func waitForDriverReady(csiConn connection.CSIConnection, timeout time.Duration) error { + now := time.Now() + finish := now.Add(timeout) + var err error + for { + ctx, cancel := context.WithTimeout(context.Background(), csiTimeout) + defer cancel() + err = csiConn.Probe(ctx) + if err == nil { + glog.V(2).Infof("Probe succeeded") + return nil + } + glog.V(2).Infof("Probe failed with %s", err) + + now := time.Now() + if now.After(finish) { + return fmt.Errorf("failed to probe the controller: %s", err) + } + time.Sleep(time.Second) + } } diff --git a/pkg/apis/volumesnapshot/v1alpha1/types.go b/pkg/apis/volumesnapshot/v1alpha1/types.go index 63fb4cd9..723aca27 100644 --- a/pkg/apis/volumesnapshot/v1alpha1/types.go +++ b/pkg/apis/volumesnapshot/v1alpha1/types.go @@ -19,6 +19,7 @@ package v1alpha1 import ( core_v1 "k8s.io/api/core/v1" storage "k8s.io/api/storage/v1beta1" + "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ) @@ -79,7 +80,7 @@ type VolumeSnapshotSpec struct { // Name of the VolumeSnapshotClass used by the VolumeSnapshot. If not specified, a default snapshot class will // be used if it is available. // +optional - VolumeSnapshotClassName string `json:"snapshotClassName" protobuf:"bytes,3,opt,name=snapshotClassName"` + VolumeSnapshotClassName *string `json:"snapshotClassName" protobuf:"bytes,3,opt,name=snapshotClassName"` } // VolumeSnapshotStatus is the status of the VolumeSnapshot @@ -89,18 +90,24 @@ type VolumeSnapshotStatus struct { // +optional CreationTime *metav1.Time `json:"createdAt" protobuf:"bytes,1,opt,name=createdAt"` + // When restoring volume from the snapshot, the volume size should be equal to or + // larger than the RestoreSize if it is specified. If RestoreSize is set to nil, it means + // that the storage plugin does not have this information available. + // +optional + RestoreSize *resource.Quantity `json:"restoreSize" protobuf:"bytes,2,opt,name=restoreSize"` + // Ready is set to true only if the snapshot is ready to use (e.g., finish uploading if // there is an uploading phase) and also VolumeSnapshot and its VolumeSnapshotContent // bind correctly with each other. If any of the above condition is not true, Ready is // set to false // +optional - Ready bool `json:"ready" protobuf:"varint,2,opt,name=ready"` + Ready bool `json:"ready" protobuf:"varint,3,opt,name=ready"` // The last error encountered during create snapshot operation, if any. // This field must only be set by the entity completing the create snapshot // operation, i.e. the external-snapshotter. // +optional - Error *storage.VolumeError + Error *storage.VolumeError `json:"error,omitempty" protobuf:"bytes,4,opt,name=error,casttype=VolumeError"` } // TypedLocalObjectReference contains enough information to let you locate the typed referenced object inside the same namespace. @@ -194,6 +201,11 @@ type VolumeSnapshotContentSpec struct { // taken from. It becomes non-nil when VolumeSnapshot and VolumeSnapshotContent are bound. // +optional PersistentVolumeRef *core_v1.ObjectReference `json:"persistentVolumeRef" protobuf:"bytes,3,opt,name=persistentVolumeRef"` + + // Name of the VolumeSnapshotClass used by the VolumeSnapshot. If not specified, a default snapshot class will + // be used if it is available. + // +optional + VolumeSnapshotClassName *string `json:"snapshotClassName" protobuf:"bytes,4,opt,name=snapshotClassName"` } // VolumeSnapshotSource represents the actual location and type of the snapshot. Only one of its members may be specified. @@ -206,19 +218,28 @@ type VolumeSnapshotSource struct { // Represents the source from CSI volume snapshot type CSIVolumeSnapshotSource struct { // Driver is the name of the driver to use for this snapshot. + // This MUST be the same name returned by the CSI GetPluginName() call for + // that driver. // Required. - Driver string `json:"driver"` + Driver string `json:"driver" protobuf:"bytes,1,opt,name=driver"` // SnapshotHandle is the unique snapshot id returned by the CSI volume // plugin’s CreateSnapshot to refer to the snapshot on all subsequent calls. // Required. - SnapshotHandle string `json:"snapshotHandle"` + SnapshotHandle string `json:"snapshotHandle" protobuf:"bytes,2,opt,name=snapshotHandle"` // Timestamp when the point-in-time snapshot is taken on the storage // system. This timestamp will be generated by the CSI volume driver after // the snapshot is cut. The format of this field should be a Unix nanoseconds // time encoded as an int64. On Unix, the command `date +%s%N` returns // the current time in nanoseconds since 1970-01-01 00:00:00 UTC. - // This field is REQUIRED. - CreatedAt int64 `json:"createdAt,omitempty" protobuf:"varint,3,opt,name=createdAt"` + // This field is required in the CSI spec but optional here to support static binding. + // +optional + CreationTime *int64 `json:"creationTime,omitempty" protobuf:"varint,3,opt,name=creationTime"` + + // When restoring volume from the snapshot, the volume size should be equal to or + // larger than the RestoreSize if it is specified. If RestoreSize is set to nil, it means + // that the storage plugin does not have this information available. + // +optional + RestoreSize *resource.Quantity `json:"restoreSize" protobuf:"bytes,4,opt,name=restoreSize"` } diff --git a/pkg/apis/volumesnapshot/v1alpha1/zz_generated.deepcopy.go b/pkg/apis/volumesnapshot/v1alpha1/zz_generated.deepcopy.go index a7dad6ef..2f7d25b9 100644 --- a/pkg/apis/volumesnapshot/v1alpha1/zz_generated.deepcopy.go +++ b/pkg/apis/volumesnapshot/v1alpha1/zz_generated.deepcopy.go @@ -29,6 +29,16 @@ import ( // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *CSIVolumeSnapshotSource) DeepCopyInto(out *CSIVolumeSnapshotSource) { *out = *in + if in.CreationTime != nil { + in, out := &in.CreationTime, &out.CreationTime + *out = new(int64) + **out = **in + } + if in.RestoreSize != nil { + in, out := &in.RestoreSize, &out.RestoreSize + x := (*in).DeepCopy() + *out = &x + } return } @@ -226,6 +236,11 @@ func (in *VolumeSnapshotContentSpec) DeepCopyInto(out *VolumeSnapshotContentSpec *out = new(v1.ObjectReference) **out = **in } + if in.VolumeSnapshotClassName != nil { + in, out := &in.VolumeSnapshotClassName, &out.VolumeSnapshotClassName + *out = new(string) + **out = **in + } return } @@ -278,7 +293,7 @@ func (in *VolumeSnapshotSource) DeepCopyInto(out *VolumeSnapshotSource) { if in.CSI != nil { in, out := &in.CSI, &out.CSI *out = new(CSIVolumeSnapshotSource) - **out = **in + (*in).DeepCopyInto(*out) } return } @@ -301,6 +316,11 @@ func (in *VolumeSnapshotSpec) DeepCopyInto(out *VolumeSnapshotSpec) { *out = new(TypedLocalObjectReference) **out = **in } + if in.VolumeSnapshotClassName != nil { + in, out := &in.VolumeSnapshotClassName, &out.VolumeSnapshotClassName + *out = new(string) + **out = **in + } return } @@ -321,6 +341,11 @@ func (in *VolumeSnapshotStatus) DeepCopyInto(out *VolumeSnapshotStatus) { in, out := &in.CreationTime, &out.CreationTime *out = (*in).DeepCopy() } + if in.RestoreSize != nil { + in, out := &in.RestoreSize, &out.RestoreSize + x := (*in).DeepCopy() + *out = &x + } if in.Error != nil { in, out := &in.Error, &out.Error *out = new(v1beta1.VolumeError) diff --git a/pkg/connection/connection.go b/pkg/connection/connection.go new file mode 100644 index 00000000..3789f287 --- /dev/null +++ b/pkg/connection/connection.go @@ -0,0 +1,264 @@ +/* +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 connection + +import ( + "context" + "fmt" + "net" + "strings" + "time" + + "github.com/container-storage-interface/spec/lib/go/csi/v0" + "github.com/golang/glog" + "google.golang.org/grpc" + "google.golang.org/grpc/connectivity" + "k8s.io/api/core/v1" +) + +// CSIConnection is gRPC connection to a remote CSI driver and abstracts all +// CSI calls. +type CSIConnection interface { + // GetDriverName returns driver name as discovered by GetPluginInfo() + // gRPC call. + GetDriverName(ctx context.Context) (string, error) + + // SupportsControllerCreateSnapshot returns true if the CSI driver reports + // CREATE_DELETE_SNAPSHOT in ControllerGetCapabilities() gRPC call. + SupportsControllerCreateSnapshot(ctx context.Context) (bool, error) + + // SupportsControllerListSnapshots returns true if the CSI driver reports + // LIST_SNAPSHOTS in ControllerGetCapabilities() gRPC call. + SupportsControllerListSnapshots(ctx context.Context) (bool, error) + + // CreateSnapshot creates a snapshot for a volume + CreateSnapshot(ctx context.Context, snapshotName string, volume *v1.PersistentVolume, parameters map[string]string, snapshotterCredentials map[string]string) (driverName string, snapshotId string, timestamp int64, size int64, status *csi.SnapshotStatus, err error) + + // DeleteSnapshot deletes a snapshot from a volume + DeleteSnapshot(ctx context.Context, snapshotID string, snapshotterCredentials map[string]string) (err error) + + // GetSnapshotStatus lists snapshot from a volume + GetSnapshotStatus(ctx context.Context, snapshotID string) (*csi.SnapshotStatus, int64, error) + + // Probe checks that the CSI driver is ready to process requests + Probe(ctx context.Context) error + + // Close the connection + Close() error +} + +type csiConnection struct { + conn *grpc.ClientConn +} + +var ( + _ CSIConnection = &csiConnection{} +) + +func New(address string, timeout time.Duration) (CSIConnection, error) { + conn, err := connect(address, timeout) + if err != nil { + return nil, err + } + return &csiConnection{ + conn: conn, + }, nil +} + +func connect(address string, timeout time.Duration) (*grpc.ClientConn, error) { + glog.V(2).Infof("Connecting to %s", address) + dialOptions := []grpc.DialOption{ + grpc.WithInsecure(), + grpc.WithBackoffMaxDelay(time.Second), + grpc.WithUnaryInterceptor(logGRPC), + } + if strings.HasPrefix(address, "/") { + dialOptions = append(dialOptions, grpc.WithDialer(func(addr string, timeout time.Duration) (net.Conn, error) { + return net.DialTimeout("unix", addr, timeout) + })) + } + conn, err := grpc.Dial(address, dialOptions...) + + if err != nil { + return nil, err + } + ctx, cancel := context.WithTimeout(context.Background(), timeout) + defer cancel() + for { + if !conn.WaitForStateChange(ctx, conn.GetState()) { + glog.V(4).Infof("Connection timed out") + // subsequent GetPluginInfo will show the real connection error + return conn, nil + } + if conn.GetState() == connectivity.Ready { + glog.V(3).Infof("Connected") + return conn, nil + } + glog.V(4).Infof("Still trying, connection is %s", conn.GetState()) + } +} + +func (c *csiConnection) GetDriverName(ctx context.Context) (string, error) { + client := csi.NewIdentityClient(c.conn) + + req := csi.GetPluginInfoRequest{} + + rsp, err := client.GetPluginInfo(ctx, &req) + if err != nil { + return "", err + } + name := rsp.GetName() + if name == "" { + return "", fmt.Errorf("name is empty") + } + return name, nil +} + +func (c *csiConnection) Probe(ctx context.Context) error { + client := csi.NewIdentityClient(c.conn) + + req := csi.ProbeRequest{} + + _, err := client.Probe(ctx, &req) + if err != nil { + return err + } + return nil +} + +func (c *csiConnection) SupportsControllerCreateSnapshot(ctx context.Context) (bool, error) { + client := csi.NewControllerClient(c.conn) + req := csi.ControllerGetCapabilitiesRequest{} + + rsp, err := client.ControllerGetCapabilities(ctx, &req) + if err != nil { + return false, err + } + caps := rsp.GetCapabilities() + for _, cap := range caps { + if cap == nil { + continue + } + rpc := cap.GetRpc() + if rpc == nil { + continue + } + if rpc.GetType() == csi.ControllerServiceCapability_RPC_CREATE_DELETE_SNAPSHOT { + return true, nil + } + } + return false, nil +} + +func (c *csiConnection) SupportsControllerListSnapshots(ctx context.Context) (bool, error) { + client := csi.NewControllerClient(c.conn) + req := csi.ControllerGetCapabilitiesRequest{} + + rsp, err := client.ControllerGetCapabilities(ctx, &req) + if err != nil { + return false, err + } + caps := rsp.GetCapabilities() + for _, cap := range caps { + if cap == nil { + continue + } + rpc := cap.GetRpc() + if rpc == nil { + continue + } + if rpc.GetType() == csi.ControllerServiceCapability_RPC_LIST_SNAPSHOTS { + return true, nil + } + } + return false, nil +} + +func (c *csiConnection) CreateSnapshot(ctx context.Context, snapshotName string, volume *v1.PersistentVolume, parameters map[string]string, snapshotterCredentials map[string]string) (string, string, int64, int64, *csi.SnapshotStatus, error) { + glog.V(5).Infof("CSI CreateSnapshot: %s", snapshotName) + if volume.Spec.CSI == nil { + return "", "", 0, 0, nil, fmt.Errorf("CSIPersistentVolumeSource not defined in spec") + } + + client := csi.NewControllerClient(c.conn) + + driverName, err := c.GetDriverName(ctx) + if err != nil { + return "", "", 0, 0, nil, err + } + + req := csi.CreateSnapshotRequest{ + SourceVolumeId: volume.Spec.CSI.VolumeHandle, + Name: snapshotName, + Parameters: parameters, + CreateSnapshotSecrets: snapshotterCredentials, + } + + rsp, err := client.CreateSnapshot(ctx, &req) + if err != nil { + return "", "", 0, 0, nil, err + } + + glog.V(5).Infof("CSI CreateSnapshot: %s driver name [%s] snapshot ID [%s] time stamp [%d] size [%d] status [%s]", snapshotName, driverName, rsp.Snapshot.Id, rsp.Snapshot.CreatedAt, rsp.Snapshot.SizeBytes, *rsp.Snapshot.Status) + return driverName, rsp.Snapshot.Id, rsp.Snapshot.CreatedAt, rsp.Snapshot.SizeBytes, rsp.Snapshot.Status, nil +} + +func (c *csiConnection) DeleteSnapshot(ctx context.Context, snapshotID string, snapshotterCredentials map[string]string) (err error) { + client := csi.NewControllerClient(c.conn) + + req := csi.DeleteSnapshotRequest{ + SnapshotId: snapshotID, + DeleteSnapshotSecrets: snapshotterCredentials, + } + + if _, err := client.DeleteSnapshot(ctx, &req); err != nil { + return err + } + + return nil +} + +func (c *csiConnection) GetSnapshotStatus(ctx context.Context, snapshotID string) (*csi.SnapshotStatus, int64, error) { + client := csi.NewControllerClient(c.conn) + + req := csi.ListSnapshotsRequest{ + SnapshotId: snapshotID, + } + + rsp, err := client.ListSnapshots(ctx, &req) + if err != nil { + return nil, 0, err + } + + if rsp.Entries == nil || len(rsp.Entries) == 0 { + return nil, 0, fmt.Errorf("can not find snapshot for snapshotID %s", snapshotID) + } + + return rsp.Entries[0].Snapshot.Status, rsp.Entries[0].Snapshot.CreatedAt, nil +} + +func (c *csiConnection) Close() error { + return c.conn.Close() +} + +func logGRPC(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error { + glog.V(5).Infof("GRPC call: %s", method) + err := invoker(ctx, method, req, reply, cc, opts...) + glog.V(5).Infof("GRPC response: %+v", reply) + glog.V(5).Infof("GRPC error: %v", err) + return err +} diff --git a/pkg/controller/csi_handler.go b/pkg/controller/csi_handler.go new file mode 100644 index 00000000..437f4f8f --- /dev/null +++ b/pkg/controller/csi_handler.go @@ -0,0 +1,112 @@ +/* +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 controller + +import ( + "context" + "fmt" + "strings" + "time" + + "github.com/container-storage-interface/spec/lib/go/csi/v0" + crdv1 "github.com/kubernetes-csi/external-snapshotter/pkg/apis/volumesnapshot/v1alpha1" + "github.com/kubernetes-csi/external-snapshotter/pkg/connection" + "k8s.io/api/core/v1" +) + +// Handler is responsible for handling VolumeSnapshot events from informer. +type Handler interface { + CreateSnapshot(snapshot *crdv1.VolumeSnapshot, volume *v1.PersistentVolume, parameters map[string]string, snapshotterCredentials map[string]string) (string, string, int64, int64, *csi.SnapshotStatus, error) + DeleteSnapshot(content *crdv1.VolumeSnapshotContent, snapshotterCredentials map[string]string) error + GetSnapshotStatus(content *crdv1.VolumeSnapshotContent) (*csi.SnapshotStatus, int64, error) +} + +// csiHandler is a handler that calls CSI to create/delete volume snapshot. +type csiHandler struct { + csiConnection connection.CSIConnection + timeout time.Duration + snapshotNamePrefix string + snapshotNameUUIDLength int +} + +func NewCSIHandler( + csiConnection connection.CSIConnection, + timeout time.Duration, + snapshotNamePrefix string, + snapshotNameUUIDLength int, +) Handler { + return &csiHandler{ + csiConnection: csiConnection, + timeout: timeout, + snapshotNamePrefix: snapshotNamePrefix, + snapshotNameUUIDLength: snapshotNameUUIDLength, + } +} + +func (handler *csiHandler) CreateSnapshot(snapshot *crdv1.VolumeSnapshot, volume *v1.PersistentVolume, parameters map[string]string, snapshotterCredentials map[string]string) (string, string, int64, int64, *csi.SnapshotStatus, error) { + + ctx, cancel := context.WithTimeout(context.Background(), handler.timeout) + defer cancel() + + snapshotName, err := makeSnapshotName(handler.snapshotNamePrefix, string(snapshot.UID), handler.snapshotNameUUIDLength) + if err != nil { + return "", "", 0, 0, nil, err + } + return handler.csiConnection.CreateSnapshot(ctx, snapshotName, volume, parameters, snapshotterCredentials) +} + +func (handler *csiHandler) DeleteSnapshot(content *crdv1.VolumeSnapshotContent, snapshotterCredentials map[string]string) error { + if content.Spec.CSI == nil { + return fmt.Errorf("CSISnapshot not defined in spec") + } + ctx, cancel := context.WithTimeout(context.Background(), handler.timeout) + defer cancel() + + err := handler.csiConnection.DeleteSnapshot(ctx, content.Spec.CSI.SnapshotHandle, snapshotterCredentials) + if err != nil { + return fmt.Errorf("failed to delete snapshot data %s: %q", content.Name, err) + } + + return nil +} + +func (handler *csiHandler) GetSnapshotStatus(content *crdv1.VolumeSnapshotContent) (*csi.SnapshotStatus, int64, error) { + if content.Spec.CSI == nil { + return nil, 0, fmt.Errorf("CSISnapshot not defined in spec") + } + ctx, cancel := context.WithTimeout(context.Background(), handler.timeout) + defer cancel() + + csiSnapshotStatus, timestamp, err := handler.csiConnection.GetSnapshotStatus(ctx, content.Spec.CSI.SnapshotHandle) + if err != nil { + return nil, 0, fmt.Errorf("failed to list snapshot data %s: %q", content.Name, err) + } + return csiSnapshotStatus, timestamp, nil +} + +func makeSnapshotName(prefix, snapshotUID string, snapshotNameUUIDLength int) (string, error) { + // create persistent name based on a volumeNamePrefix and volumeNameUUIDLength + // of PVC's UID + if len(snapshotUID) == 0 { + return "", fmt.Errorf("Corrupted snapshot object, it is missing UID") + } + if snapshotNameUUIDLength == -1 { + // Default behavior is to not truncate or remove dashes + return fmt.Sprintf("%s-%s", prefix, snapshotUID), nil + } + return fmt.Sprintf("%s-%s", prefix, strings.Replace(snapshotUID, "-", "", -1)[0:snapshotNameUUIDLength]), nil +} diff --git a/pkg/controller/snapshot_controller.go b/pkg/controller/snapshot_controller.go new file mode 100644 index 00000000..2401eda6 --- /dev/null +++ b/pkg/controller/snapshot_controller.go @@ -0,0 +1,855 @@ +/* +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 controller + +import ( + "fmt" + "strings" + "time" + + "github.com/container-storage-interface/spec/lib/go/csi/v0" + "github.com/golang/glog" + crdv1 "github.com/kubernetes-csi/external-snapshotter/pkg/apis/volumesnapshot/v1alpha1" + "k8s.io/api/core/v1" + storagev1 "k8s.io/api/storage/v1" + storage "k8s.io/api/storage/v1beta1" + apierrs "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/api/resource" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/labels" + "k8s.io/client-go/kubernetes/scheme" + ref "k8s.io/client-go/tools/reference" + "k8s.io/kubernetes/pkg/util/goroutinemap" + "k8s.io/kubernetes/pkg/util/goroutinemap/exponentialbackoff" +) + +// ================================================================== +// PLEASE DO NOT ATTEMPT TO SIMPLIFY THIS CODE. +// KEEP THE SPACE SHUTTLE FLYING. +// ================================================================== + +// Design: +// +// The fundamental key to this design is the bi-directional "pointer" between +// VolumeSnapshots and VolumeSnapshotContents, which is represented here +// as snapshot.Spec.SnapshotContentName and content.Spec.VolumeSnapshotRef. +// The bi-directionality is complicated to manage in a transactionless system, but +// without it we can't ensure sane behavior in the face of different forms of +// trouble. For example, a rogue HA controller instance could end up racing +// and making multiple bindings that are indistinguishable, resulting in +// potential data loss. +// +// This controller is designed to work in active-passive high availability +// mode. It *could* work also in active-active HA mode, all the object +// transitions are designed to cope with this, however performance could be +// lower as these two active controllers will step on each other toes +// frequently. +// +// This controller supports both dynamic snapshot creation and pre-bound snapshot. +// In pre-bound mode, objects are created with pre-defined pointers: a VolumeSnapshot +// points to a specific VolumeSnapshotContent and the VolumeSnapshotContent also +// points back for this VolumeSnapshot. +// +// The dynamic snapshot creation is multi-step process: first controller triggers +// snapshot creation though csi volume plugin which should return a snapshot after +// it is created successfully (however, the snapshot might not be ready to use yet if +// there is an uploading phase). The creationTimestamp will be updated according to +// VolumeSnapshot, and then a VolumeSnapshotContent object is created to represent +// this snapshot. After that, the controller will keep checking the snapshot status +// though csi snapshot calls. When the snapshot is ready to use, the controller set +// the status "Bound" to true to indicate the snapshot is bound and ready to use. +// If the createtion failed for any reason, the Error status is set accordingly. +// In alpha version, the controller not retry to create the snapshot after it failed. +// In the future version, a retry policy will be added. + +const pvcKind = "PersistentVolumeClaim" +const controllerUpdateFailMsg = "snapshot controller failed to update" + +const IsDefaultSnapshotClassAnnotation = "snapshot.storage.kubernetes.io/is-default-class" + +// syncContent deals with one key off the queue. It returns false when it's time to quit. +func (ctrl *csiSnapshotController) syncContent(content *crdv1.VolumeSnapshotContent) error { + glog.V(5).Infof("synchronizing VolumeSnapshotContent[%s]", content.Name) + + // VolumeSnapshotContent is not bound to any VolumeSnapshot, this case rare and we just return err + if content.Spec.VolumeSnapshotRef == nil { + // content is not bound + glog.V(4).Infof("synchronizing VolumeSnapshotContent[%s]: VolumeSnapshotContent is not bound to any VolumeSnapshot", content.Name) + ctrl.eventRecorder.Event(content, v1.EventTypeWarning, "SnapshotContentNotBound", "VolumeSnapshotContent is not bound to any VolumeSnapshot") + return fmt.Errorf("volumeSnapshotContent %s is not bound to any VolumeSnapshot", content.Name) + } else { + glog.V(4).Infof("synchronizing VolumeSnapshotContent[%s]: content is bound to snapshot %s", content.Name, snapshotRefKey(content.Spec.VolumeSnapshotRef)) + // The VolumeSnapshotContent is reserved for a VolumeSnapshot; + // that VolumeSnapshot has not yet been bound to this VolumeSnapshotContent; the VolumeSnapshot sync will handle it. + if content.Spec.VolumeSnapshotRef.UID == "" { + glog.V(4).Infof("synchronizing VolumeSnapshotContent[%s]: VolumeSnapshotContent is pre-bound to VolumeSnapshot %s", content.Name, snapshotRefKey(content.Spec.VolumeSnapshotRef)) + return nil + } + // Get the VolumeSnapshot by _name_ + var snapshot *crdv1.VolumeSnapshot + snapshotName := snapshotRefKey(content.Spec.VolumeSnapshotRef) + obj, found, err := ctrl.snapshotStore.GetByKey(snapshotName) + if err != nil { + return err + } + if !found { + glog.V(4).Infof("synchronizing VolumeSnapshotContent[%s]: snapshot %s not found", content.Name, snapshotRefKey(content.Spec.VolumeSnapshotRef)) + // Fall through with snapshot = nil + } else { + var ok bool + snapshot, ok = obj.(*crdv1.VolumeSnapshot) + if !ok { + return fmt.Errorf("cannot convert object from snapshot cache to snapshot %q!?: %#v", content.Name, obj) + } + glog.V(4).Infof("synchronizing VolumeSnapshotContent[%s]: snapshot %s found", content.Name, snapshotRefKey(content.Spec.VolumeSnapshotRef)) + } + if snapshot != nil && snapshot.UID != content.Spec.VolumeSnapshotRef.UID { + // The snapshot that the content was pointing to was deleted, and another + // with the same name created. + glog.V(4).Infof("synchronizing VolumeSnapshotContent[%s]: content %s has different UID, the old one must have been deleted", content.Name, snapshotRefKey(content.Spec.VolumeSnapshotRef)) + // Treat the volume as bound to a missing claim. + snapshot = nil + } + if snapshot == nil { + ctrl.deleteSnapshotContent(content) + } + } + return nil +} + +// syncSnapshot is the main controller method to decide what to do with a snapshot. +// It's invoked by appropriate cache.Controller callbacks when a snapshot is +// created, updated or periodically synced. We do not differentiate between +// these events. +// For easier readability, it is split into syncUnreadySnapshot and syncReadySnapshot +func (ctrl *csiSnapshotController) syncSnapshot(snapshot *crdv1.VolumeSnapshot) error { + glog.V(5).Infof("synchonizing VolumeSnapshot[%s]: %s", snapshotKey(snapshot), getSnapshotStatusForLogging(snapshot)) + + if !snapshot.Status.Ready { + return ctrl.syncUnreadySnapshot(snapshot) + } else { + return ctrl.syncReadySnapshot(snapshot) + } +} + +// syncReadySnapshot checks the snapshot which has been bound to snapshot content succesfully before. +// If there is any problem with the binding (e.g., snapshot points to a non-exist snapshot content), update the snapshot status and emit event. +func (ctrl *csiSnapshotController) syncReadySnapshot(snapshot *crdv1.VolumeSnapshot) error { + if snapshot.Spec.SnapshotContentName == "" { + if err := ctrl.updateSnapshotErrorStatusWithEvent(snapshot, v1.EventTypeWarning, "SnapshotLost", "Bound snapshot has lost reference to VolumeSnapshotContent"); err != nil { + return err + } + return nil + } + obj, found, err := ctrl.contentStore.GetByKey(snapshot.Spec.SnapshotContentName) + if err != nil { + return err + } + if !found { + if err = ctrl.updateSnapshotErrorStatusWithEvent(snapshot, v1.EventTypeWarning, "SnapshotContentMissing", "VolumeSnapshotContent is missing"); err != nil { + return err + } + return nil + } else { + content, ok := obj.(*crdv1.VolumeSnapshotContent) + if !ok { + return fmt.Errorf("Cannot convert object from snapshot content store to VolumeSnapshotContent %q!?: %#v", snapshot.Spec.SnapshotContentName, obj) + } + + glog.V(5).Infof("syncCompleteSnapshot[%s]: VolumeSnapshotContent %q found", snapshotKey(snapshot), content.Name) + if !IsSnapshotBound(snapshot, content) { + // snapshot is bound but content is not bound to snapshot correctly + if err = ctrl.updateSnapshotErrorStatusWithEvent(snapshot, v1.EventTypeWarning, "SnapshotMisbound", "VolumeSnapshotContent is not bound to the VolumeSnapshot correctly"); err != nil { + return err + } + return nil + } + // Snapshot is correctly bound. + return nil + } +} + +// syncUnreadySnapshot is the main controller method to decide what to do with a snapshot which is not set to ready. +func (ctrl *csiSnapshotController) syncUnreadySnapshot(snapshot *crdv1.VolumeSnapshot) error { + uniqueSnapshotName := snapshotKey(snapshot) + glog.V(5).Infof("syncUnreadySnapshot %s", uniqueSnapshotName) + + if snapshot.Spec.SnapshotContentName != "" { + contentObj, found, err := ctrl.contentStore.GetByKey(snapshot.Spec.SnapshotContentName) + if err != nil { + return err + } + if !found { + // snapshot is bound to a non-existing content. + ctrl.updateSnapshotErrorStatusWithEvent(snapshot, v1.EventTypeWarning, "SnapshotContentMissing", "VolumeSnapshotContent is missing") + glog.V(4).Infof("synchronizing unready snapshot[%s]: snapshotcontent %q requested and not found, will try again next time", uniqueSnapshotName, snapshot.Spec.SnapshotContentName) + return fmt.Errorf("snapshot %s is bound to a non-existing content %s", uniqueSnapshotName, snapshot.Spec.SnapshotContentName) + } + content, ok := contentObj.(*crdv1.VolumeSnapshotContent) + if !ok { + return fmt.Errorf("expected volume snapshot content, got %+v", contentObj) + } + + if err := ctrl.checkandBindSnapshotContent(snapshot, content); err != nil { + // snapshot is bound but content is not bound to snapshot correctly + ctrl.updateSnapshotErrorStatusWithEvent(snapshot, v1.EventTypeWarning, "SnapshotBindFailed", fmt.Sprintf("Snapshot failed to bind VolumeSnapshotContent, %v", err)) + return fmt.Errorf("snapshot %s is bound, but VolumeSnapshotContent %s is not bound to the VolumeSnapshot correctly, %v", uniqueSnapshotName, content.Name, err) + } + + // snapshot is already bound correctly, check the status and update if it is ready. + glog.V(5).Infof("Check and update snapshot %s status", uniqueSnapshotName) + if err = ctrl.checkandUpdateSnapshotStatus(snapshot, content); err != nil { + return err + } + return nil + } else { // snapshot.Spec.SnapshotContentName == nil + if contentObj := ctrl.getMatchSnapshotContent(snapshot); contentObj != nil { + glog.V(5).Infof("Find VolumeSnapshotContent object %s for snapshot %s", contentObj.Name, uniqueSnapshotName) + newSnapshot, err := ctrl.bindandUpdateVolumeSnapshot(contentObj, snapshot) + if err != nil { + return err + } + glog.V(5).Infof("bindandUpdateVolumeSnapshot %v", newSnapshot) + return nil + } else if snapshot.Status.Error == nil || isControllerUpdateFailError(snapshot.Status.Error) { // Try to create snapshot if no error status is set + if err := ctrl.createSnapshot(snapshot); err != nil { + ctrl.updateSnapshotErrorStatusWithEvent(snapshot, v1.EventTypeWarning, "SnapshotCreationFailed", fmt.Sprintf("Failed to create snapshot with error %v", err)) + return err + } + return nil + } + return nil + } +} + +// getMatchSnapshotContent looks up VolumeSnapshotContent for a VolumeSnapshot named snapshotName +func (ctrl *csiSnapshotController) getMatchSnapshotContent(snapshot *crdv1.VolumeSnapshot) *crdv1.VolumeSnapshotContent { + var snapshotContentObj *crdv1.VolumeSnapshotContent + var found bool + + objs := ctrl.contentStore.List() + for _, obj := range objs { + content := obj.(*crdv1.VolumeSnapshotContent) + if content.Spec.VolumeSnapshotRef != nil && + content.Spec.VolumeSnapshotRef.Name == snapshot.Name && + content.Spec.VolumeSnapshotRef.Namespace == snapshot.Namespace && + content.Spec.VolumeSnapshotRef.UID == snapshot.UID && + content.Spec.VolumeSnapshotClassName != nil && snapshot.Spec.VolumeSnapshotClassName != nil && + *(content.Spec.VolumeSnapshotClassName) == *(snapshot.Spec.VolumeSnapshotClassName) { + found = true + snapshotContentObj = content + break + } + } + + if !found { + glog.V(4).Infof("No VolumeSnapshotContent for VolumeSnapshot %s found", snapshotKey(snapshot)) + return nil + } + + return snapshotContentObj +} + +// deleteSnapshotContent starts delete action. +func (ctrl *csiSnapshotController) deleteSnapshotContent(content *crdv1.VolumeSnapshotContent) { + operationName := fmt.Sprintf("delete-%s[%s]", content.Name, string(content.UID)) + glog.V(5).Infof("Snapshotter is about to delete volume snapshot and the operation named %s", operationName) + ctrl.scheduleOperation(operationName, func() error { + return ctrl.deleteSnapshotContentOperation(content) + }) +} + +// scheduleOperation starts given asynchronous operation on given volume. It +// makes sure the operation is already not running. +func (ctrl *csiSnapshotController) scheduleOperation(operationName string, operation func() error) { + glog.V(5).Infof("scheduleOperation[%s]", operationName) + + err := ctrl.runningOperations.Run(operationName, operation) + if err != nil { + switch { + case goroutinemap.IsAlreadyExists(err): + glog.V(4).Infof("operation %q is already running, skipping", operationName) + case exponentialbackoff.IsExponentialBackoff(err): + glog.V(4).Infof("operation %q postponed due to exponential backoff", operationName) + default: + glog.Errorf("error scheduling operation %q: %v", operationName, err) + } + } +} + +func (ctrl *csiSnapshotController) storeSnapshotUpdate(snapshot interface{}) (bool, error) { + return storeObjectUpdate(ctrl.snapshotStore, snapshot, "snapshot") +} + +func (ctrl *csiSnapshotController) storeContentUpdate(content interface{}) (bool, error) { + return storeObjectUpdate(ctrl.contentStore, content, "content") +} + +func (ctrl *csiSnapshotController) storeClassUpdate(content interface{}) (bool, error) { + return storeObjectUpdate(ctrl.classStore, content, "class") +} + +// createSnapshot starts new asynchronous operation to create snapshot +func (ctrl *csiSnapshotController) createSnapshot(snapshot *crdv1.VolumeSnapshot) error { + glog.V(5).Infof("createSnapshot[%s]: started", snapshotKey(snapshot)) + opName := fmt.Sprintf("create-%s[%s]", snapshotKey(snapshot), string(snapshot.UID)) + ctrl.scheduleOperation(opName, func() error { + snapshotObj, err := ctrl.createSnapshotOperation(snapshot) + if err != nil { + ctrl.updateSnapshotErrorStatusWithEvent(snapshot, v1.EventTypeWarning, "SnapshotCreationFailed", fmt.Sprintf("Failed to create snapshot: %v", err)) + glog.Errorf("createSnapshot [%s]: error occurred in createSnapshotOperation: %v", opName, err) + return err + } + _, updateErr := ctrl.storeSnapshotUpdate(snapshotObj) + if updateErr != nil { + // We will get an "snapshot update" event soon, this is not a big error + glog.V(4).Infof("createSnapshot [%s]: cannot update internal cache: %v", snapshotKey(snapshotObj), updateErr) + } + + return nil + }) + return nil +} + +func (ctrl *csiSnapshotController) checkandUpdateSnapshotStatus(snapshot *crdv1.VolumeSnapshot, content *crdv1.VolumeSnapshotContent) error { + glog.V(5).Infof("checkandUpdateSnapshotStatus[%s] started", snapshotKey(snapshot)) + opName := fmt.Sprintf("check-%s[%s]", snapshotKey(snapshot), string(snapshot.UID)) + ctrl.scheduleOperation(opName, func() error { + snapshotObj, err := ctrl.checkandUpdateSnapshotStatusOperation(snapshot, content) + if err != nil { + ctrl.updateSnapshotErrorStatusWithEvent(snapshot, v1.EventTypeWarning, "SnapshotCheckandUpdateFailed", fmt.Sprintf("Failed to check and update snapshot: %v", err)) + glog.Errorf("checkandUpdateSnapshotStatus [%s]: error occured %v", snapshotKey(snapshot), err) + return err + } + _, updateErr := ctrl.storeSnapshotUpdate(snapshotObj) + if updateErr != nil { + // We will get an "snapshot update" event soon, this is not a big error + glog.V(4).Infof("checkandUpdateSnapshotStatus [%s]: cannot update internal cache: %v", snapshotKey(snapshotObj), updateErr) + } + + return nil + }) + return nil +} + +// updateSnapshotStatusWithEvent saves new snapshot.Status to API server and emits +// given event on the snapshot. It saves the status and emits the event only when +// the status has actually changed from the version saved in API server. +// Parameters: +// snapshot - snapshot to update +// eventtype, reason, message - event to send, see EventRecorder.Event() +func (ctrl *csiSnapshotController) updateSnapshotErrorStatusWithEvent(snapshot *crdv1.VolumeSnapshot, eventtype, reason, message string) error { + glog.V(5).Infof("updateSnapshotStatusWithEvent[%s]", snapshotKey(snapshot)) + + if snapshot.Status.Error != nil && snapshot.Status.Error.Message == message { + glog.V(4).Infof("updateSnapshotStatusWithEvent[%s]: the same error %v is already set", snapshot.Name, snapshot.Status.Error) + return nil + } + snapshotClone := snapshot.DeepCopy() + statusError := &storage.VolumeError{ + Time: metav1.Time{ + Time: time.Now(), + }, + Message: message, + } + snapshotClone.Status.Error = statusError + + snapshotClone.Status.Ready = false + newSnapshot, err := ctrl.clientset.VolumesnapshotV1alpha1().VolumeSnapshots(snapshotClone.Namespace).Update(snapshotClone) + if err != nil { + glog.V(4).Infof("updating VolumeSnapshot[%s] error status failed %v", snapshotKey(snapshot), err) + return err + } + + _, err = ctrl.storeSnapshotUpdate(newSnapshot) + if err != nil { + glog.V(4).Infof("updating VolumeSnapshot[%s] error status: cannot update internal cache %v", snapshotKey(snapshot), err) + return err + } + // Emit the event only when the status change happens + ctrl.eventRecorder.Event(newSnapshot, eventtype, reason, message) + + return nil +} + +// Stateless functions +func getSnapshotStatusForLogging(snapshot *crdv1.VolumeSnapshot) string { + return fmt.Sprintf("bound to: %q, Completed: %v", snapshot.Spec.SnapshotContentName, snapshot.Status.Ready) +} + +func IsSnapshotBound(snapshot *crdv1.VolumeSnapshot, content *crdv1.VolumeSnapshotContent) bool { + if content.Spec.VolumeSnapshotRef != nil && content.Spec.VolumeSnapshotRef.Name == snapshot.Name && + content.Spec.VolumeSnapshotRef.UID == snapshot.UID { + return true + } + return false +} + +// The function checks whether the volumeSnapshotRef in snapshot content matches the given snapshot. If match, it binds the content with the snapshot +func (ctrl *csiSnapshotController) checkandBindSnapshotContent(snapshot *crdv1.VolumeSnapshot, content *crdv1.VolumeSnapshotContent) error { + if content.Spec.VolumeSnapshotRef == nil || content.Spec.VolumeSnapshotRef.Name != snapshot.Name { + return fmt.Errorf("Could not bind snapshot %s and content %s, the VolumeSnapshotRef does not match", snapshot.Name, content.Name) + } else if content.Spec.VolumeSnapshotRef.UID != "" && content.Spec.VolumeSnapshotRef.UID != snapshot.UID { + return fmt.Errorf("Could not bind snapshot %s and content %s, the VolumeSnapshotRef does not match", snapshot.Name, content.Name) + } else if content.Spec.VolumeSnapshotRef.UID == "" { + contentClone := content.DeepCopy() + contentClone.Spec.VolumeSnapshotRef.UID = snapshot.UID + className := *(snapshot.Spec.VolumeSnapshotClassName) + contentClone.Spec.VolumeSnapshotClassName = &className + newContent, err := ctrl.clientset.VolumesnapshotV1alpha1().VolumeSnapshotContents().Update(contentClone) + if err != nil { + glog.V(4).Infof("updating VolumeSnapshotContent[%s] error status failed %v", newContent.Name, err) + return err + } + _, err = ctrl.storeContentUpdate(newContent) + if err != nil { + glog.V(4).Infof("updating VolumeSnapshotContent[%s] error status: cannot update internal cache %v", newContent.Name, err) + return err + } + } + return nil +} + +func (ctrl *csiSnapshotController) checkandUpdateSnapshotStatusOperation(snapshot *crdv1.VolumeSnapshot, content *crdv1.VolumeSnapshotContent) (*crdv1.VolumeSnapshot, error) { + status, _, err := ctrl.handler.GetSnapshotStatus(content) + if err != nil { + return nil, fmt.Errorf("failed to check snapshot status %s with error %v", snapshot.Name, err) + } + + newSnapshot, err := ctrl.updateSnapshotStatus(snapshot, status, time.Now(), nil, IsSnapshotBound(snapshot, content)) + if err != nil { + return nil, err + } + return newSnapshot, nil +} + +// The function goes through the whole snapshot creation process. +// 1. Trigger the snapshot through csi storage provider. +// 2. Update VolumeSnapshot status with creationtimestamp information +// 3. Create the VolumeSnapshotContent object with the snapshot id information. +// 4. Bind the VolumeSnapshot and VolumeSnapshotContent object +func (ctrl *csiSnapshotController) createSnapshotOperation(snapshot *crdv1.VolumeSnapshot) (*crdv1.VolumeSnapshot, error) { + glog.Infof("createSnapshot: Creating snapshot %s through the plugin ...", snapshotKey(snapshot)) + + if snapshot.Status.Error != nil && !isControllerUpdateFailError(snapshot.Status.Error) { + glog.V(4).Infof("error is already set in snapshot, do not retry to create: %s", snapshot.Status.Error.Message) + return snapshot, nil + } + + className := snapshot.Spec.VolumeSnapshotClassName + glog.V(5).Infof("createSnapshotOperation [%s]: VolumeSnapshotClassName [%s]", snapshot.Name, *className) + var class *crdv1.VolumeSnapshotClass + var err error + if className != nil { + class, err = ctrl.GetSnapshotClass(*className) + if err != nil { + glog.Errorf("createSnapshotOperation failed to getClassFromVolumeSnapshot %s", err) + return nil, err + } + } else { + glog.Errorf("failed to take snapshot %s without a snapshot class", snapshot.Name) + return nil, fmt.Errorf("failed to take snapshot %s without a snapshot class", snapshot.Name) + } + + volume, err := ctrl.getVolumeFromVolumeSnapshot(snapshot) + if err != nil { + glog.Errorf("createSnapshotOperation failed to get PersistentVolume object [%s]: Error: [%#v]", snapshot.Name, err) + return nil, err + } + + // Create VolumeSnapshotContent name + contentName := GetSnapshotContentNameForSnapshot(snapshot) + + // Resolve snapshotting secret credentials. + snapshotterSecretRef, err := GetSecretReference(class.Parameters, contentName, snapshot) + if err != nil { + return nil, err + } + snapshotterCredentials, err := GetCredentials(ctrl.client, snapshotterSecretRef) + if err != nil { + return nil, err + } + + driverName, snapshotID, timestamp, size, csiSnapshotStatus, err := ctrl.handler.CreateSnapshot(snapshot, volume, class.Parameters, snapshotterCredentials) + if err != nil { + return nil, fmt.Errorf("failed to take snapshot of the volume, %s: %q", volume.Name, err) + } + glog.V(5).Infof("Created snapshot: driver %s, snapshotId %s, timestamp %d, size %d, csiSnapshotStatus %v", driverName, snapshotID, timestamp, size, csiSnapshotStatus) + + var newSnapshot *crdv1.VolumeSnapshot + // Update snapshot status with timestamp + for i := 0; i < ctrl.createSnapshotContentRetryCount; i++ { + glog.V(5).Infof("createSnapshot [%s]: trying to update snapshot creation timestamp", snapshotKey(snapshot)) + newSnapshot, err = ctrl.updateSnapshotStatus(snapshot, csiSnapshotStatus, time.Unix(0, timestamp), resource.NewQuantity(size, resource.BinarySI), false) + if err == nil { + break + } + glog.V(4).Infof("failed to update snapshot %s creation timestamp: %v", snapshotKey(snapshot), err) + } + + if err != nil { + return nil, err + } + // Create VolumeSnapshotContent in the database + volumeRef, err := ref.GetReference(scheme.Scheme, volume) + + snapshotContent := &crdv1.VolumeSnapshotContent{ + ObjectMeta: metav1.ObjectMeta{ + Name: contentName, + }, + Spec: crdv1.VolumeSnapshotContentSpec{ + VolumeSnapshotRef: &v1.ObjectReference{ + Kind: "VolumeSnapshot", + Namespace: snapshot.Namespace, + Name: snapshot.Name, + UID: snapshot.UID, + APIVersion: "snapshot.storage.k8s.io/v1alpha1", + }, + PersistentVolumeRef: volumeRef, + VolumeSnapshotSource: crdv1.VolumeSnapshotSource{ + CSI: &crdv1.CSIVolumeSnapshotSource{ + Driver: driverName, + SnapshotHandle: snapshotID, + CreationTime: ×tamp, + RestoreSize: resource.NewQuantity(size, resource.BinarySI), + }, + }, + VolumeSnapshotClassName: &(class.Name), + }, + } + // Try to create the VolumeSnapshotContent object several times + for i := 0; i < ctrl.createSnapshotContentRetryCount; i++ { + glog.V(5).Infof("createSnapshot [%s]: trying to save volume snapshot content %s", snapshotKey(snapshot), snapshotContent.Name) + if _, err = ctrl.clientset.VolumesnapshotV1alpha1().VolumeSnapshotContents().Create(snapshotContent); err == nil || apierrs.IsAlreadyExists(err) { + // Save succeeded. + if err != nil { + glog.V(3).Infof("volume snapshot content %q for snapshot %q already exists, reusing", snapshotContent.Name, snapshotKey(snapshot)) + err = nil + } else { + glog.V(3).Infof("volume snapshot content %q for snapshot %q saved", snapshotContent.Name, snapshotKey(snapshot)) + } + break + } + // Save failed, try again after a while. + glog.V(3).Infof("failed to save volume snapshot content %q for snapshot %q: %v", snapshotContent.Name, snapshotKey(snapshot), err) + time.Sleep(ctrl.createSnapshotContentInterval) + } + + if err != nil { + // Save failed. Now we have a storage asset outside of Kubernetes, + // but we don't have appropriate volumesnapshot content object for it. + // Emit some event here and controller should try to create the content in next sync period. + strerr := fmt.Sprintf("Error creating volume snapshot content object for snapshot %s: %v.", snapshotKey(snapshot), err) + glog.Error(strerr) + ctrl.eventRecorder.Event(newSnapshot, v1.EventTypeWarning, "CreateSnapshotContentFailed", strerr) + return nil, newControllerUpdateError(snapshotKey(snapshot), err.Error()) + } + + // save succeeded, bind and update status for snapshot. + result, err := ctrl.bindandUpdateVolumeSnapshot(snapshotContent, newSnapshot) + if err != nil { + return nil, err + } + return result, nil +} + +// Delete a snapshot +// 1. Find the SnapshotContent corresponding to Snapshot +// 1a: Not found => finish (it's been deleted already) +// 2. Ask the backend to remove the snapshot device +// 3. Delete the SnapshotContent object +// 4. Remove the Snapshot from store +// 5. Finish +func (ctrl *csiSnapshotController) deleteSnapshotContentOperation(content *crdv1.VolumeSnapshotContent) error { + glog.V(5).Infof("deleteSnapshotOperation [%s] started", content.Name) + + // get secrets if VolumeSnapshotClass specifies it + var snapshotterCredentials map[string]string + snapshotClassName := content.Spec.VolumeSnapshotClassName + if snapshotClassName != nil { + if snapshotClass, err := ctrl.classLister.Get(*snapshotClassName); err == nil { + // Resolve snapshotting secret credentials. + // No VolumeSnapshot is provided when resolving delete secret names, since the VolumeSnapshot may or may not exist at delete time. + snapshotterSecretRef, err := GetSecretReference(snapshotClass.Parameters, content.Name, nil) + if err != nil { + return err + } + snapshotterCredentials, err = GetCredentials(ctrl.client, snapshotterSecretRef) + if err != nil { + return err + } + } + } + + err := ctrl.handler.DeleteSnapshot(content, snapshotterCredentials) + if err != nil { + ctrl.eventRecorder.Event(content, v1.EventTypeWarning, "SnapshotDeleteError", "Failed to delete snapshot") + return fmt.Errorf("failed to delete snapshot %#v, err: %v", content.Name, err) + } + + err = ctrl.clientset.VolumesnapshotV1alpha1().VolumeSnapshotContents().Delete(content.Name, &metav1.DeleteOptions{}) + if err != nil { + ctrl.eventRecorder.Event(content, v1.EventTypeWarning, "SnapshotContentObjectDeleteError", "Failed to delete snapshot content API object") + return fmt.Errorf("failed to delete VolumeSnapshotContent %s from API server: %q", content.Name, err) + } + + return nil +} + +func (ctrl *csiSnapshotController) bindandUpdateVolumeSnapshot(snapshotContent *crdv1.VolumeSnapshotContent, snapshot *crdv1.VolumeSnapshot) (*crdv1.VolumeSnapshot, error) { + glog.V(5).Infof("bindandUpdateVolumeSnapshot for snapshot [%s]: snapshotContent [%s]", snapshot.Name, snapshotContent.Name) + snapshotObj, err := ctrl.clientset.VolumesnapshotV1alpha1().VolumeSnapshots(snapshot.Namespace).Get(snapshot.Name, metav1.GetOptions{}) + if err != nil { + return nil, fmt.Errorf("error get snapshot %s from api server: %v", snapshotKey(snapshot), err) + } + + // Copy the snapshot object before updating it + snapshotCopy := snapshotObj.DeepCopy() + + if snapshotObj.Spec.SnapshotContentName == snapshotContent.Name { + glog.Infof("bindVolumeSnapshotContentToVolumeSnapshot: VolumeSnapshot %s already bind to volumeSnapshotContent [%s]", snapshot.Name, snapshotContent.Name) + } else { + glog.Infof("bindVolumeSnapshotContentToVolumeSnapshot: before bind VolumeSnapshot %s to volumeSnapshotContent [%s]", snapshot.Name, snapshotContent.Name) + snapshotCopy.Spec.SnapshotContentName = snapshotContent.Name + updateSnapshot, err := ctrl.clientset.VolumesnapshotV1alpha1().VolumeSnapshots(snapshot.Namespace).Update(snapshotCopy) + if err != nil { + glog.Infof("bindVolumeSnapshotContentToVolumeSnapshot: Error binding VolumeSnapshot %s to volumeSnapshotContent [%s]. Error [%#v]", snapshot.Name, snapshotContent.Name, err) + return nil, newControllerUpdateError(snapshotKey(snapshot), err.Error()) + } + snapshotCopy = updateSnapshot + _, err = ctrl.storeSnapshotUpdate(snapshotCopy) + if err != nil { + glog.Errorf("%v", err) + } + } + + glog.V(5).Infof("bindandUpdateVolumeSnapshot for snapshot completed [%#v]", snapshotCopy) + return snapshotCopy, nil +} + +// UpdateSnapshotStatus converts snapshot status to crdv1.VolumeSnapshotCondition +func (ctrl *csiSnapshotController) updateSnapshotStatus(snapshot *crdv1.VolumeSnapshot, csistatus *csi.SnapshotStatus, timestamp time.Time, size *resource.Quantity, bound bool) (*crdv1.VolumeSnapshot, error) { + glog.V(5).Infof("updating VolumeSnapshot[]%s, set status %v, timestamp %v", snapshotKey(snapshot), csistatus, timestamp) + status := snapshot.Status + change := false + timeAt := &metav1.Time{ + Time: timestamp, + } + + snapshotClone := snapshot.DeepCopy() + switch csistatus.Type { + case csi.SnapshotStatus_READY: + if bound { + status.Ready = true + // Remove the error if checking snapshot is already bound and ready + status.Error = nil + change = true + } + if status.CreationTime == nil { + status.CreationTime = timeAt + change = true + } + case csi.SnapshotStatus_ERROR_UPLOADING: + if status.Error == nil { + status.Error = &storage.VolumeError{ + Time: *timeAt, + Message: "Failed to upload the snapshot", + } + change = true + ctrl.eventRecorder.Event(snapshot, v1.EventTypeWarning, "SnapshotUploadError", "Failed to upload the snapshot") + + } + case csi.SnapshotStatus_UPLOADING: + if status.CreationTime == nil { + status.CreationTime = timeAt + change = true + } + } + if change { + if size != nil { + status.RestoreSize = size + } + snapshotClone.Status = status + newSnapshotObj, err := ctrl.clientset.VolumesnapshotV1alpha1().VolumeSnapshots(snapshotClone.Namespace).Update(snapshotClone) + if err != nil { + return nil, newControllerUpdateError(snapshotKey(snapshot), err.Error()) + } else { + return newSnapshotObj, nil + } + } + return snapshot, nil +} + +// getVolumeFromVolumeSnapshot is a helper function to get PV from VolumeSnapshot. +func (ctrl *csiSnapshotController) getVolumeFromVolumeSnapshot(snapshot *crdv1.VolumeSnapshot) (*v1.PersistentVolume, error) { + pvc, err := ctrl.getClaimFromVolumeSnapshot(snapshot) + if err != nil { + return nil, err + } + + if pvc.Status.Phase != v1.ClaimBound { + return nil, fmt.Errorf("the PVC %s is not yet bound to a PV, will not attempt to take a snapshot", pvc.Name) + } + + pvName := pvc.Spec.VolumeName + pv, err := ctrl.client.CoreV1().PersistentVolumes().Get(pvName, metav1.GetOptions{}) + if err != nil { + return nil, fmt.Errorf("failed to retrieve PV %s from the API server: %q", pvName, err) + } + + glog.V(5).Infof("getVolumeFromVolumeSnapshot: snapshot [%s] PV name [%s]", snapshot.Name, pvName) + + return pv, nil +} + +func (ctrl *csiSnapshotController) getStorageClassFromVolumeSnapshot(snapshot *crdv1.VolumeSnapshot) (*storagev1.StorageClass, error) { + // Get storage class from PVC or PV + pvc, err := ctrl.getClaimFromVolumeSnapshot(snapshot) + if err != nil { + return nil, err + } + storageclassName := *pvc.Spec.StorageClassName + if len(storageclassName) == 0 { + volume, err := ctrl.getVolumeFromVolumeSnapshot(snapshot) + if err != nil { + return nil, err + } + storageclassName = volume.Spec.StorageClassName + } + if len(storageclassName) == 0 { + return nil, fmt.Errorf("cannot figure out the snapshot class automatically, please specify one in snapshot spec.") + } + storageclass, err := ctrl.client.StorageV1().StorageClasses().Get(*pvc.Spec.StorageClassName, metav1.GetOptions{}) + if err != nil { + return nil, err + } + return storageclass, nil +} + +// GetSnapshotClass is a helper function to get snapshot class from the class name. +func (ctrl *csiSnapshotController) GetSnapshotClass(className string) (*crdv1.VolumeSnapshotClass, error) { + glog.V(5).Infof("getSnapshotClass: VolumeSnapshotClassName [%s]", className) + + obj, found, err := ctrl.classStore.GetByKey(className) + if found { + class, ok := obj.(*crdv1.VolumeSnapshotClass) + if ok { + return class, nil + } + } + class, err := ctrl.classLister.Get(className) + if err != nil { + glog.Errorf("failed to retrieve snapshot class %s from the API server: %q", className, err) + return nil, fmt.Errorf("failed to retrieve snapshot class %s from the API server: %q", className, err) + } + _, updateErr := ctrl.storeClassUpdate(class) + if updateErr != nil { + glog.V(4).Infof("getSnapshotClass [%s]: cannot update internal cache: %v", class.Name, updateErr) + } + return class, nil +} + +// SetDefaultSnapshotClass is a helper function to figure out the default snapshot class from +// PVC/PV StorageClass and update VolumeSnapshot with this snapshot class name. +func (ctrl *csiSnapshotController) SetDefaultSnapshotClass(snapshot *crdv1.VolumeSnapshot) (*crdv1.VolumeSnapshotClass, *crdv1.VolumeSnapshot, error) { + glog.V(5).Infof("SetDefaultSnapshotClass for snapshot [%s]", snapshot.Name) + + storageclass, err := ctrl.getStorageClassFromVolumeSnapshot(snapshot) + if err != nil { + return nil, nil, err + } + // Find default snapshot class if available + list, err := ctrl.classLister.List(labels.Everything()) + if err != nil { + return nil, nil, err + } + defaultClasses := []*crdv1.VolumeSnapshotClass{} + + for _, class := range list { + if IsDefaultAnnotation(class.ObjectMeta) && storageclass.Provisioner == class.Snapshotter && ctrl.snapshotterName == class.Snapshotter { + defaultClasses = append(defaultClasses, class) + glog.V(5).Infof("get defaultClass added: %s", class.Name) + } + } + if len(defaultClasses) == 0 { + return nil, nil, fmt.Errorf("cannot find default snapshot class") + } + if len(defaultClasses) > 1 { + glog.V(4).Infof("get DefaultClass %d defaults found", len(defaultClasses)) + return nil, nil, fmt.Errorf("%d default snapshot classes were found", len(defaultClasses)) + } + glog.V(5).Infof("setDefaultSnapshotClass [%s]: default VolumeSnapshotClassName [%s]", snapshot.Name, defaultClasses[0].Name) + snapshotClone := snapshot.DeepCopy() + snapshotClone.Spec.VolumeSnapshotClassName = &(defaultClasses[0].Name) + newSnapshot, err := ctrl.clientset.VolumesnapshotV1alpha1().VolumeSnapshots(snapshotClone.Namespace).Update(snapshotClone) + if err != nil { + glog.V(4).Infof("updating VolumeSnapshot[%s] default class failed %v", snapshotKey(snapshot), err) + } + _, updateErr := ctrl.storeSnapshotUpdate(newSnapshot) + if updateErr != nil { + // We will get an "snapshot update" event soon, this is not a big error + glog.V(4).Infof("setDefaultSnapshotClass [%s]: cannot update internal cache: %v", snapshotKey(snapshot), updateErr) + } + _, updateErr = ctrl.storeClassUpdate(defaultClasses[0]) + if updateErr != nil { + glog.V(4).Infof("setDefaultSnapshotClass [%s]: cannot update internal cache: %v", defaultClasses[0].Name, updateErr) + } + return defaultClasses[0], newSnapshot, nil +} + +// getClaimFromVolumeSnapshot is a helper function to get PVC from VolumeSnapshot. +func (ctrl *csiSnapshotController) getClaimFromVolumeSnapshot(snapshot *crdv1.VolumeSnapshot) (*v1.PersistentVolumeClaim, error) { + if snapshot.Spec.Source == nil || snapshot.Spec.Source.Kind != pvcKind { + return nil, fmt.Errorf("The snapshot source is not the right type. Expected %s, Got %v", pvcKind, snapshot.Spec.Source) + } + pvcName := snapshot.Spec.Source.Name + if pvcName == "" { + return nil, fmt.Errorf("the PVC name is not specified in snapshot %s", snapshotKey(snapshot)) + } + + pvc, err := ctrl.client.CoreV1().PersistentVolumeClaims(snapshot.Namespace).Get(pvcName, metav1.GetOptions{}) + if err != nil { + return nil, fmt.Errorf("failed to retrieve PVC %s from the API server: %q", pvcName, err) + } + + return pvc, nil +} + +var _ error = controllerUpdateError{} + +type controllerUpdateError struct { + message string +} + +func newControllerUpdateError(name, message string) error { + return controllerUpdateError{ + message: fmt.Sprintf("%s %s on API server: %s", controllerUpdateFailMsg, name, message), + } +} + +func (e controllerUpdateError) Error() string { + return e.message +} + +func isControllerUpdateFailError(err *storage.VolumeError) bool { + if err != nil { + if strings.Contains(err.Message, controllerUpdateFailMsg) { + return true + } + } + return false +} diff --git a/pkg/controller/snapshot_controller_base.go b/pkg/controller/snapshot_controller_base.go new file mode 100644 index 00000000..9a5c519a --- /dev/null +++ b/pkg/controller/snapshot_controller_base.go @@ -0,0 +1,472 @@ +/* +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 controller + +import ( + "fmt" + "time" + + "github.com/golang/glog" + crdv1 "github.com/kubernetes-csi/external-snapshotter/pkg/apis/volumesnapshot/v1alpha1" + clientset "github.com/kubernetes-csi/external-snapshotter/pkg/client/clientset/versioned" + storageinformers "github.com/kubernetes-csi/external-snapshotter/pkg/client/informers/externalversions/volumesnapshot/v1alpha1" + storagelisters "github.com/kubernetes-csi/external-snapshotter/pkg/client/listers/volumesnapshot/v1alpha1" + "github.com/kubernetes-csi/external-snapshotter/pkg/connection" + "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/labels" + "k8s.io/apimachinery/pkg/util/wait" + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/kubernetes/scheme" + corev1 "k8s.io/client-go/kubernetes/typed/core/v1" + "k8s.io/client-go/tools/cache" + "k8s.io/client-go/tools/record" + "k8s.io/client-go/util/workqueue" + "k8s.io/kubernetes/pkg/util/goroutinemap" +) + +type csiSnapshotController struct { + clientset clientset.Interface + client kubernetes.Interface + snapshotterName string + eventRecorder record.EventRecorder + snapshotQueue workqueue.RateLimitingInterface + contentQueue workqueue.RateLimitingInterface + + snapshotLister storagelisters.VolumeSnapshotLister + snapshotListerSynced cache.InformerSynced + contentLister storagelisters.VolumeSnapshotContentLister + contentListerSynced cache.InformerSynced + classLister storagelisters.VolumeSnapshotClassLister + classListerSynced cache.InformerSynced + + snapshotStore cache.Store + contentStore cache.Store + classStore cache.Store + + handler Handler + // Map of scheduled/running operations. + runningOperations goroutinemap.GoRoutineMap + + createSnapshotContentRetryCount int + createSnapshotContentInterval time.Duration + resyncPeriod time.Duration +} + +// NewCSISnapshotController returns a new *csiSnapshotController +func NewCSISnapshotController( + clientset clientset.Interface, + client kubernetes.Interface, + snapshotterName string, + volumeSnapshotInformer storageinformers.VolumeSnapshotInformer, + volumeSnapshotContentInformer storageinformers.VolumeSnapshotContentInformer, + volumeSnapshotClassInformer storageinformers.VolumeSnapshotClassInformer, + createSnapshotContentRetryCount int, + createSnapshotContentInterval time.Duration, + conn connection.CSIConnection, + timeout time.Duration, + resyncPeriod time.Duration, + snapshotNamePrefix string, + snapshotNameUUIDLength int, +) *csiSnapshotController { + broadcaster := record.NewBroadcaster() + broadcaster.StartRecordingToSink(&corev1.EventSinkImpl{Interface: client.Core().Events(v1.NamespaceAll)}) + var eventRecorder record.EventRecorder + eventRecorder = broadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: fmt.Sprintf("csi-snapshotter %s", snapshotterName)}) + + ctrl := &csiSnapshotController{ + clientset: clientset, + client: client, + snapshotterName: snapshotterName, + eventRecorder: eventRecorder, + handler: NewCSIHandler(conn, timeout, snapshotNamePrefix, snapshotNameUUIDLength), + runningOperations: goroutinemap.NewGoRoutineMap(true), + createSnapshotContentRetryCount: createSnapshotContentRetryCount, + createSnapshotContentInterval: createSnapshotContentInterval, + resyncPeriod: resyncPeriod, + snapshotStore: cache.NewStore(cache.DeletionHandlingMetaNamespaceKeyFunc), + contentStore: cache.NewStore(cache.DeletionHandlingMetaNamespaceKeyFunc), + classStore: cache.NewStore(cache.DeletionHandlingMetaNamespaceKeyFunc), + snapshotQueue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "csi-snapshotter-snapshot"), + contentQueue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "csi-snapshotter-content"), + } + + volumeSnapshotInformer.Informer().AddEventHandlerWithResyncPeriod( + cache.ResourceEventHandlerFuncs{ + AddFunc: func(obj interface{}) { ctrl.enqueueSnapshotWork(obj) }, + UpdateFunc: func(oldObj, newObj interface{}) { ctrl.enqueueSnapshotWork(newObj) }, + DeleteFunc: func(obj interface{}) { ctrl.enqueueSnapshotWork(obj) }, + }, + ctrl.resyncPeriod, + ) + ctrl.snapshotLister = volumeSnapshotInformer.Lister() + ctrl.snapshotListerSynced = volumeSnapshotInformer.Informer().HasSynced + + volumeSnapshotContentInformer.Informer().AddEventHandlerWithResyncPeriod( + cache.ResourceEventHandlerFuncs{ + AddFunc: func(obj interface{}) { ctrl.enqueueContentWork(obj) }, + UpdateFunc: func(oldObj, newObj interface{}) { ctrl.enqueueContentWork(newObj) }, + DeleteFunc: func(obj interface{}) { ctrl.enqueueContentWork(obj) }, + }, + ctrl.resyncPeriod, + ) + ctrl.contentLister = volumeSnapshotContentInformer.Lister() + ctrl.contentListerSynced = volumeSnapshotContentInformer.Informer().HasSynced + + ctrl.classLister = volumeSnapshotClassInformer.Lister() + ctrl.classListerSynced = volumeSnapshotClassInformer.Informer().HasSynced + + return ctrl +} + +func (ctrl *csiSnapshotController) Run(workers int, stopCh <-chan struct{}) { + defer ctrl.snapshotQueue.ShutDown() + defer ctrl.contentQueue.ShutDown() + + glog.Infof("Starting CSI snapshotter") + defer glog.Infof("Shutting CSI snapshotter") + + if !cache.WaitForCacheSync(stopCh, ctrl.snapshotListerSynced, ctrl.contentListerSynced) { + glog.Errorf("Cannot sync caches") + return + } + + ctrl.initializeCaches(ctrl.snapshotLister, ctrl.contentLister) + + for i := 0; i < workers; i++ { + go wait.Until(ctrl.snapshotWorker, 0, stopCh) + go wait.Until(ctrl.contentWorker, 0, stopCh) + } + + <-stopCh +} + +// enqueueSnapshotWork adds snapshot to given work queue. +func (ctrl *csiSnapshotController) enqueueSnapshotWork(obj interface{}) { + // Beware of "xxx deleted" events + if unknown, ok := obj.(cache.DeletedFinalStateUnknown); ok && unknown.Obj != nil { + obj = unknown.Obj + } + if snapshot, ok := obj.(*crdv1.VolumeSnapshot); ok { + objName, err := cache.DeletionHandlingMetaNamespaceKeyFunc(snapshot) + if err != nil { + glog.Errorf("failed to get key from object: %v, %v", err, snapshot) + return + } + glog.V(5).Infof("enqueued %q for sync", objName) + ctrl.snapshotQueue.Add(objName) + } +} + +// enqueueContentWork adds snapshot content to given work queue. +func (ctrl *csiSnapshotController) enqueueContentWork(obj interface{}) { + // Beware of "xxx deleted" events + if unknown, ok := obj.(cache.DeletedFinalStateUnknown); ok && unknown.Obj != nil { + obj = unknown.Obj + } + if content, ok := obj.(*crdv1.VolumeSnapshotContent); ok { + objName, err := cache.DeletionHandlingMetaNamespaceKeyFunc(content) + if err != nil { + glog.Errorf("failed to get key from object: %v, %v", err, content) + return + } + glog.V(5).Infof("enqueued %q for sync", objName) + ctrl.contentQueue.Add(objName) + } +} + +// snapshotWorker processes items from snapshotQueue. It must run only once, +// syncSnapshot is not assured to be reentrant. +func (ctrl *csiSnapshotController) snapshotWorker() { + workFunc := func() bool { + keyObj, quit := ctrl.snapshotQueue.Get() + if quit { + return true + } + defer ctrl.snapshotQueue.Done(keyObj) + key := keyObj.(string) + glog.V(5).Infof("snapshotWorker[%s]", key) + + namespace, name, err := cache.SplitMetaNamespaceKey(key) + glog.V(5).Infof("snapshotWorker: snapshot namespace [%s] name [%s]", namespace, name) + if err != nil { + glog.Errorf("error getting namespace & name of snapshot %q to get snapshot from informer: %v", key, err) + return false + } + snapshot, err := ctrl.snapshotLister.VolumeSnapshots(namespace).Get(name) + if err == nil { + // The volume snapshot still exists in informer cache, the event must have + // been add/update/sync + newSnapshot, err := ctrl.checkAndUpdateSnapshotClass(snapshot) + if err == nil { + glog.V(5).Infof("passed checkAndUpdateSnapshotClass for snapshot %q", key) + ctrl.updateSnapshot(newSnapshot) + } + return false + } + if err != nil && !errors.IsNotFound(err) { + glog.V(2).Infof("error getting snapshot %q from informer: %v", key, err) + return false + } + // The snapshot is not in informer cache, the event must have been "delete" + vsObj, found, err := ctrl.snapshotStore.GetByKey(key) + if err != nil { + glog.V(2).Infof("error getting snapshot %q from cache: %v", key, err) + return false + } + if !found { + // The controller has already processed the delete event and + // deleted the snapshot from its cache + glog.V(2).Infof("deletion of snapshot %q was already processed", key) + return false + } + snapshot, ok := vsObj.(*crdv1.VolumeSnapshot) + if !ok { + glog.Errorf("expected vs, got %+v", vsObj) + return false + } + newSnapshot, err := ctrl.checkAndUpdateSnapshotClass(snapshot) + if err == nil { + ctrl.deleteSnapshot(newSnapshot) + } + return false + } + + for { + if quit := workFunc(); quit { + glog.Infof("snapshot worker queue shutting down") + return + } + } +} + +// contentWorker processes items from contentQueue. It must run only once, +// syncContent is not assured to be reentrant. +func (ctrl *csiSnapshotController) contentWorker() { + workFunc := func() bool { + keyObj, quit := ctrl.contentQueue.Get() + if quit { + return true + } + defer ctrl.contentQueue.Done(keyObj) + key := keyObj.(string) + glog.V(5).Infof("contentWorker[%s]", key) + + _, name, err := cache.SplitMetaNamespaceKey(key) + if err != nil { + glog.V(4).Infof("error getting name of snapshotContent %q to get snapshotContent from informer: %v", key, err) + return false + } + content, err := ctrl.contentLister.Get(name) + if err == nil { + // Skip update if content is for another CSI driver + snapshotClassName := content.Spec.VolumeSnapshotClassName + if snapshotClassName != nil { + if snapshotClass, err := ctrl.classLister.Get(*snapshotClassName); err == nil { + if snapshotClass.Snapshotter != ctrl.snapshotterName { + return false + } + } + } + + // The content still exists in informer cache, the event must have + // been add/update/sync + ctrl.updateContent(content) + return false + } + if !errors.IsNotFound(err) { + glog.V(2).Infof("error getting content %q from informer: %v", key, err) + return false + } + + // The content is not in informer cache, the event must have been + // "delete" + contentObj, found, err := ctrl.contentStore.GetByKey(key) + if err != nil { + glog.V(2).Infof("error getting content %q from cache: %v", key, err) + return false + } + if !found { + // The controller has already processed the delete event and + // deleted the volume from its cache + glog.V(2).Infof("deletion of content %q was already processed", key) + return false + } + content, ok := contentObj.(*crdv1.VolumeSnapshotContent) + if !ok { + glog.Errorf("expected content, got %+v", content) + return false + } + ctrl.deleteContent(content) + return false + } + + for { + if quit := workFunc(); quit { + glog.Infof("content worker queue shutting down") + return + } + } +} + +// checkAndUpdateSnapshotClass gets the VolumeSnapshotClass from VolumeSnapshot. If it is not set, +// gets it from default VolumeSnapshotClass and sets it. It also detects if snapshotter in the +// VolumeSnapshotClass is the same as the snapshotter in external controller. +func (ctrl *csiSnapshotController) checkAndUpdateSnapshotClass(snapshot *crdv1.VolumeSnapshot) (*crdv1.VolumeSnapshot, error) { + className := snapshot.Spec.VolumeSnapshotClassName + var class *crdv1.VolumeSnapshotClass + var err error + newSnapshot := snapshot + if className != nil { + glog.V(5).Infof("checkAndUpdateSnapshotClass [%s]: VolumeSnapshotClassName [%s]", snapshot.Name, *className) + class, err = ctrl.GetSnapshotClass(*className) + if err != nil { + glog.Errorf("checkAndUpdateSnapshotClass failed to getSnapshotClass %v", err) + ctrl.updateSnapshotErrorStatusWithEvent(snapshot, v1.EventTypeWarning, "GetSnapshotClassFailed", fmt.Sprintf("Failed to get snapshot class with error %v", err)) + return nil, err + } + } else { + glog.V(5).Infof("checkAndUpdateSnapshotClass [%s]: SetDefaultSnapshotClass", snapshot.Name) + class, newSnapshot, err = ctrl.SetDefaultSnapshotClass(snapshot) + if err != nil { + glog.Errorf("checkAndUpdateSnapshotClass failed to setDefaultClass %v", err) + ctrl.updateSnapshotErrorStatusWithEvent(snapshot, v1.EventTypeWarning, "SetDefaultSnapshotClassFailed", fmt.Sprintf("Failed to set default snapshot class with error %v", err)) + return nil, err + } + } + + glog.V(5).Infof("VolumeSnapshotClass Snapshotter [%s] Snapshot Controller snapshotterName [%s]", class.Snapshotter, ctrl.snapshotterName) + if class.Snapshotter != ctrl.snapshotterName { + glog.V(4).Infof("Skipping VolumeSnapshot %s for snapshotter [%s] in VolumeSnapshotClass because it does not match with the snapshotter for controller [%s]", snapshotKey(snapshot), class.Snapshotter, ctrl.snapshotterName) + return nil, fmt.Errorf("volumeSnapshotClass does not match with the snapshotter for controller") + } + return newSnapshot, nil +} + +// updateSnapshot runs in worker thread and handles "snapshot added", +// "snapshot updated" and "periodic sync" events. +func (ctrl *csiSnapshotController) updateSnapshot(snapshot *crdv1.VolumeSnapshot) { + // Store the new snapshot version in the cache and do not process it if this is + // an old version. + glog.V(5).Infof("updateSnapshot %q", snapshotKey(snapshot)) + newSnapshot, err := ctrl.storeSnapshotUpdate(snapshot) + if err != nil { + glog.Errorf("%v", err) + } + if !newSnapshot { + return + } + err = ctrl.syncSnapshot(snapshot) + if err != nil { + if errors.IsConflict(err) { + // Version conflict error happens quite often and the controller + // recovers from it easily. + glog.V(3).Infof("could not sync claim %q: %+v", snapshotKey(snapshot), err) + } else { + glog.Errorf("could not sync volume %q: %+v", snapshotKey(snapshot), err) + } + } +} + +// updateContent runs in worker thread and handles "content added", +// "content updated" and "periodic sync" events. +func (ctrl *csiSnapshotController) updateContent(content *crdv1.VolumeSnapshotContent) { + // Store the new content version in the cache and do not process it if this is + // an old version. + new, err := ctrl.storeContentUpdate(content) + if err != nil { + glog.Errorf("%v", err) + } + if !new { + return + } + err = ctrl.syncContent(content) + if err != nil { + if errors.IsConflict(err) { + // Version conflict error happens quite often and the controller + // recovers from it easily. + glog.V(3).Infof("could not sync content %q: %+v", content.Name, err) + } else { + glog.Errorf("could not sync content %q: %+v", content.Name, err) + } + } +} + +// deleteSnapshot runs in worker thread and handles "snapshot deleted" event. +func (ctrl *csiSnapshotController) deleteSnapshot(snapshot *crdv1.VolumeSnapshot) { + _ = ctrl.snapshotStore.Delete(snapshot) + glog.V(4).Infof("snapshot %q deleted", snapshotKey(snapshot)) + + snapshotContentName := snapshot.Spec.SnapshotContentName + if snapshotContentName == "" { + glog.V(5).Infof("deleteSnapshot[%q]: content not bound", snapshotKey(snapshot)) + return + } + // sync the content when its snapshot is deleted. Explicitly sync'ing the + // content here in response to snapshot deletion prevents the content from + // waiting until the next sync period for its Release. + glog.V(5).Infof("deleteSnapshot[%q]: scheduling sync of content %s", snapshotKey(snapshot), snapshotContentName) + ctrl.contentQueue.Add(snapshotContentName) +} + +// deleteContent runs in worker thread and handles "snapshot deleted" event. +func (ctrl *csiSnapshotController) deleteContent(content *crdv1.VolumeSnapshotContent) { + _ = ctrl.contentStore.Delete(content) + glog.V(4).Infof("content %q deleted", content.Name) + + snapshotName := snapshotRefKey(content.Spec.VolumeSnapshotRef) + if snapshotName == "" { + glog.V(5).Infof("deleteContent[%q]: content not bound", content.Name) + return + } + // sync the snapshot when its content is deleted. Explicitly sync'ing the + // snapshot here in response to content deletion prevents the snapshot from + // waiting until the next sync period for its Release. + glog.V(5).Infof("deleteContent[%q]: scheduling sync of snapshot %s", content.Name, snapshotName) + ctrl.snapshotQueue.Add(snapshotName) +} + +// initializeCaches fills all controller caches with initial data from etcd in +// order to have the caches already filled when first addSnapshot/addContent to +// perform initial synchronization of the controller. +func (ctrl *csiSnapshotController) initializeCaches(snapshotLister storagelisters.VolumeSnapshotLister, contentLister storagelisters.VolumeSnapshotContentLister) { + snapshotList, err := snapshotLister.List(labels.Everything()) + if err != nil { + glog.Errorf("CSISnapshotController can't initialize caches: %v", err) + return + } + for _, snapshot := range snapshotList { + snapshotClone := snapshot.DeepCopy() + if _, err = ctrl.storeSnapshotUpdate(snapshotClone); err != nil { + glog.Errorf("error updating volume snapshot cache: %v", err) + } + } + + contentList, err := contentLister.List(labels.Everything()) + if err != nil { + glog.Errorf("CSISnapshotController can't initialize caches: %v", err) + return + } + for _, content := range contentList { + contentClone := content.DeepCopy() + if _, err = ctrl.storeSnapshotUpdate(contentClone); err != nil { + glog.Errorf("error updating volume snapshot cache: %v", err) + } + } + + glog.V(4).Infof("controller initialized") +} diff --git a/pkg/controller/util.go b/pkg/controller/util.go new file mode 100644 index 00000000..ec0a5ad3 --- /dev/null +++ b/pkg/controller/util.go @@ -0,0 +1,236 @@ +/* +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 controller + +import ( + "fmt" + "github.com/golang/glog" + crdv1 "github.com/kubernetes-csi/external-snapshotter/pkg/apis/volumesnapshot/v1alpha1" + "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/api/meta" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/util/sets" + "k8s.io/apimachinery/pkg/util/validation" + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/tools/cache" + "os" + "strconv" +) + +var ( + keyFunc = cache.DeletionHandlingMetaNamespaceKeyFunc +) + +const snapshotterSecretNameKey = "csiSnapshotterSecretName" +const snapshotterSecretNamespaceKey = "csiSnapshotterSecretNamespace" + +func snapshotKey(vs *crdv1.VolumeSnapshot) string { + return fmt.Sprintf("%s/%s", vs.Namespace, vs.Name) +} + +func snapshotRefKey(vsref *v1.ObjectReference) string { + return fmt.Sprintf("%s/%s", vsref.Namespace, vsref.Name) +} + +// storeObjectUpdate updates given cache with a new object version from Informer +// callback (i.e. with events from etcd) or with an object modified by the +// controller itself. Returns "true", if the cache was updated, false if the +// object is an old version and should be ignored. +func storeObjectUpdate(store cache.Store, obj interface{}, className string) (bool, error) { + objName, err := keyFunc(obj) + if err != nil { + return false, fmt.Errorf("Couldn't get key for object %+v: %v", obj, err) + } + oldObj, found, err := store.Get(obj) + if err != nil { + return false, fmt.Errorf("Error finding %s %q in controller cache: %v", className, objName, err) + } + + objAccessor, err := meta.Accessor(obj) + if err != nil { + return false, err + } + + if !found { + // This is a new object + glog.V(4).Infof("storeObjectUpdate: adding %s %q, version %s", className, objName, objAccessor.GetResourceVersion()) + if err = store.Add(obj); err != nil { + return false, fmt.Errorf("error adding %s %q to controller cache: %v", className, objName, err) + } + return true, nil + } + + oldObjAccessor, err := meta.Accessor(oldObj) + if err != nil { + return false, err + } + + objResourceVersion, err := strconv.ParseInt(objAccessor.GetResourceVersion(), 10, 64) + if err != nil { + return false, fmt.Errorf("error parsing ResourceVersion %q of %s %q: %s", objAccessor.GetResourceVersion(), className, objName, err) + } + oldObjResourceVersion, err := strconv.ParseInt(oldObjAccessor.GetResourceVersion(), 10, 64) + if err != nil { + return false, fmt.Errorf("error parsing old ResourceVersion %q of %s %q: %s", oldObjAccessor.GetResourceVersion(), className, objName, err) + } + + // Throw away only older version, let the same version pass - we do want to + // get periodic sync events. + if oldObjResourceVersion > objResourceVersion { + glog.V(4).Infof("storeObjectUpdate: ignoring %s %q version %s", className, objName, objAccessor.GetResourceVersion()) + return false, nil + } + + glog.V(4).Infof("storeObjectUpdate updating %s %q with version %s", className, objName, objAccessor.GetResourceVersion()) + if err = store.Update(obj); err != nil { + return false, fmt.Errorf("error updating %s %q in controller cache: %v", className, objName, err) + } + return true, nil +} + +// GetSnapshotContentNameForSnapshot returns SnapshotContent.Name for the create VolumeSnapshotContent. +// The name must be unique. +func GetSnapshotContentNameForSnapshot(snapshot *crdv1.VolumeSnapshot) string { + return "snapcontent-" + string(snapshot.UID) +} + +// IsDefaultAnnotation returns a boolean if +// the annotation is set +func IsDefaultAnnotation(obj metav1.ObjectMeta) bool { + if obj.Annotations[IsDefaultSnapshotClassAnnotation] == "true" { + return true + } + + return false +} + +// GetSecretReference returns a reference to the secret specified in the given nameKey and namespaceKey parameters, or an error if the parameters are not specified correctly. +// if neither the name or namespace parameter are set, a nil reference and no error is returned. +// no lookup of the referenced secret is performed, and the secret may or may not exist. +// +// supported tokens for name resolution: +// - ${volumesnapshotcontent.name} +// - ${volumesnapshot.namespace} +// - ${volumesnapshot.name} +// - ${volumesnapshot.annotations['ANNOTATION_KEY']} (e.g. ${pvc.annotations['example.com/snapshot-create-secret-name']}) +// +// supported tokens for namespace resolution: +// - ${volumesnapshotcontent.name} +// - ${volumesnapshot.namespace} +// +// an error is returned in the following situations: +// - only one of name or namespace is provided +// - the name or namespace parameter contains a token that cannot be resolved +// - the resolved name is not a valid secret name +// - the resolved namespace is not a valid namespace name +func GetSecretReference(snapshotClassParams map[string]string, snapContentName string, snapshot *crdv1.VolumeSnapshot) (*v1.SecretReference, error) { + nameTemplate, hasName := snapshotClassParams[snapshotterSecretNameKey] + namespaceTemplate, hasNamespace := snapshotClassParams[snapshotterSecretNamespaceKey] + + if !hasName && !hasNamespace { + return nil, nil + } + + if len(nameTemplate) == 0 || len(namespaceTemplate) == 0 { + return nil, fmt.Errorf("%s and %s parameters must be specified together", snapshotterSecretNameKey, snapshotterSecretNamespaceKey) + } + + ref := &v1.SecretReference{} + + // Secret namespace template can make use of the VolumeSnapshotContent name or the VolumeSnapshot namespace. + // Note that neither of those things are under the control of the VolumeSnapshot user. + namespaceParams := map[string]string{"volumesnapshotcontent.name": snapContentName} + // snapshot may be nil when resolving create/delete snapshot secret names because the + // snapshot may or may not exist at delete time + if snapshot != nil { + namespaceParams["volumesnapshot.namespace"] = snapshot.Namespace + } + + resolvedNamespace, err := resolveTemplate(namespaceTemplate, namespaceParams) + if err != nil { + return nil, fmt.Errorf("error resolving %s value %q: %v", snapshotterSecretNamespaceKey, namespaceTemplate, err) + } + glog.V(4).Infof("GetSecretReference namespaceTemplate %s, namespaceParams: %+v, resolved %s", namespaceTemplate, namespaceParams, resolvedNamespace) + + if len(validation.IsDNS1123Label(resolvedNamespace)) > 0 { + if namespaceTemplate != resolvedNamespace { + return nil, fmt.Errorf("%s parameter %q resolved to %q which is not a valid namespace name", snapshotterSecretNamespaceKey, namespaceTemplate, resolvedNamespace) + } + return nil, fmt.Errorf("%s parameter %q is not a valid namespace name", snapshotterSecretNamespaceKey, namespaceTemplate) + } + ref.Namespace = resolvedNamespace + + // Secret name template can make use of the VolumeSnapshotContent name, VolumeSnapshot name or namespace, + // or a VolumeSnapshot annotation. + // Note that VolumeSnapshot name and annotations are under the VolumeSnapshot user's control. + nameParams := map[string]string{"volumesnapshotcontent.name": snapContentName} + if snapshot != nil { + nameParams["volumesnapshot.name"] = snapshot.Name + nameParams["volumesnapshot.namespace"] = snapshot.Namespace + for k, v := range snapshot.Annotations { + nameParams["volumesnapshot.annotations['"+k+"']"] = v + } + } + resolvedName, err := resolveTemplate(nameTemplate, nameParams) + if err != nil { + return nil, fmt.Errorf("error resolving %s value %q: %v", snapshotterSecretNameKey, nameTemplate, err) + } + if len(validation.IsDNS1123Subdomain(resolvedName)) > 0 { + if nameTemplate != resolvedName { + return nil, fmt.Errorf("%s parameter %q resolved to %q which is not a valid secret name", snapshotterSecretNameKey, nameTemplate, resolvedName) + } + return nil, fmt.Errorf("%s parameter %q is not a valid secret name", snapshotterSecretNameKey, nameTemplate) + } + ref.Name = resolvedName + + glog.V(4).Infof("GetSecretReference validated Secret: %+v", ref) + return ref, nil +} + +// resolveTemplate resolves the template by checking if the value is missing for a key +func resolveTemplate(template string, params map[string]string) (string, error) { + missingParams := sets.NewString() + resolved := os.Expand(template, func(k string) string { + v, ok := params[k] + if !ok { + missingParams.Insert(k) + } + return v + }) + if missingParams.Len() > 0 { + return "", fmt.Errorf("invalid tokens: %q", missingParams.List()) + } + return resolved, nil +} + +// GetCredentials retrieves credentials stored in v1.SecretReference +func GetCredentials(k8s kubernetes.Interface, ref *v1.SecretReference) (map[string]string, error) { + if ref == nil { + return nil, nil + } + + secret, err := k8s.CoreV1().Secrets(ref.Namespace).Get(ref.Name, metav1.GetOptions{}) + if err != nil { + return nil, fmt.Errorf("error getting secret %s in namespace %s: %v", ref.Name, ref.Namespace, err) + } + + credentials := map[string]string{} + for key, value := range secret.Data { + credentials[key] = string(value) + } + return credentials, nil +}