Merge pull request #163 from mesosphere/sreis/idempotent

Make NodeStageVolume idempotent

pkg/driver/driver.go

@@ -22,6 +22,7 @@ import (
 
 	csi "github.com/container-storage-interface/spec/lib/go/csi"
 	"github.com/kubernetes-sigs/aws-ebs-csi-driver/pkg/cloud"
+	"github.com/kubernetes-sigs/aws-ebs-csi-driver/pkg/driver/internal"
 	"github.com/kubernetes-sigs/aws-ebs-csi-driver/pkg/util"
 	"google.golang.org/grpc"
 	"k8s.io/klog"
@@ -40,7 +41,8 @@ type Driver struct {
 	cloud cloud.Cloud
 	srv   *grpc.Server
 
-	mounter *mount.SafeFormatAndMount
+	mounter  *mount.SafeFormatAndMount
+	inFlight *internal.InFlight
 }
 
 func NewDriver(endpoint string) (*Driver, error) {
@@ -57,6 +59,7 @@ func NewDriver(endpoint string) (*Driver, error) {
 		nodeID:   m.GetInstanceID(),
 		cloud:    cloud,
 		mounter:  newSafeMounter(),
+		inFlight: internal.NewInFlight(),
 	}, nil
 }
 

pkg/driver/fakes.go

@@ -18,6 +18,7 @@ package driver
 
 import (
 	"github.com/kubernetes-sigs/aws-ebs-csi-driver/pkg/cloud"
+	"github.com/kubernetes-sigs/aws-ebs-csi-driver/pkg/driver/internal"
 	"k8s.io/kubernetes/pkg/util/mount"
 )
 
@@ -40,5 +41,6 @@ func NewFakeDriver(endpoint string) *Driver {
 		nodeID:   cloud.GetMetadata().GetInstanceID(),
 		cloud:    cloud,
 		mounter:  NewFakeMounter(),
+		inFlight: internal.NewInFlight(),
 	}
 }

pkg/driver/internal/inflight.go

@@ -0,0 +1,70 @@
+/*
+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 internal
+
+import (
+	"sync"
+)
+
+// Idempotent is the interface required to manage in flight requests.
+type Idempotent interface {
+	// The CSI data types are generated using a protobuf.
+	// The generated structures are guaranteed to implement the Stringer interface.
+	// Example: https://github.com/container-storage-interface/spec/blob/master/lib/go/csi/csi.pb.go#L3508
+	// We can use the generated string as the key of our internal inflight database of requests.
+	String() string
+}
+
+// InFlight is a struct used to manage in flight requests.
+type InFlight struct {
+	mux      *sync.Mutex
+	inFlight map[string]bool
+}
+
+// NewInFlight instanciates a InFlight structures.
+func NewInFlight() *InFlight {
+	return &InFlight{
+		mux:      &sync.Mutex{},
+		inFlight: make(map[string]bool),
+	}
+}
+
+// Insert inserts the entry to the current list of inflight requests.
+// Returns false when the key already exists.
+func (db *InFlight) Insert(entry Idempotent) bool {
+	db.mux.Lock()
+	defer db.mux.Unlock()
+
+	hash := entry.String()
+
+	_, ok := db.inFlight[hash]
+	if ok {
+		return false
+	}
+
+	db.inFlight[hash] = true
+	return true
+}
+
+// Delete removes the entry from the inFlight entries map.
+// It doesn't return anything, and will do nothing if the specified key doesn't exist.
+func (db *InFlight) Delete(h Idempotent) {
+	db.mux.Lock()
+	defer db.mux.Unlock()
+
+	delete(db.inFlight, h.String())
+}

pkg/driver/internal/inflight_test.go

@@ -0,0 +1,210 @@
+/*
+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 internal
+
+import (
+	"testing"
+
+	"github.com/container-storage-interface/spec/lib/go/csi"
+)
+
+type testRequest struct {
+	request *csi.CreateVolumeRequest
+	expResp bool
+	delete  bool
+}
+
+var stdVolCap = []*csi.VolumeCapability{
+	{
+		AccessType: &csi.VolumeCapability_Mount{
+			Mount: &csi.VolumeCapability_MountVolume{},
+		},
+		AccessMode: &csi.VolumeCapability_AccessMode{
+			Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
+		},
+	},
+}
+var stdVolSize = int64(5 * 1024 * 1024 * 1024)
+var stdCapRange = &csi.CapacityRange{RequiredBytes: stdVolSize}
+var stdParams = map[string]string{
+	"fsType":     "ext3",
+	"volumeType": "gp2",
+}
+
+func TestInFlight(t *testing.T) {
+	testCases := []struct {
+		name     string
+		requests []testRequest
+	}{
+		{
+			name: "success normal",
+			requests: []testRequest{
+				{
+					request: &csi.CreateVolumeRequest{
+						Name:               "random-vol-name",
+						CapacityRange:      stdCapRange,
+						VolumeCapabilities: stdVolCap,
+						Parameters:         stdParams,
+					},
+					expResp: true,
+				},
+			},
+		},
+		{
+			name: "success adding request with different name",
+			requests: []testRequest{
+				{
+					request: &csi.CreateVolumeRequest{
+						Name:               "random-vol-foobar",
+						CapacityRange:      stdCapRange,
+						VolumeCapabilities: stdVolCap,
+						Parameters:         stdParams,
+					},
+					expResp: true,
+				},
+				{
+					request: &csi.CreateVolumeRequest{
+						Name:               "random-vol-name-foobar",
+						CapacityRange:      stdCapRange,
+						VolumeCapabilities: stdVolCap,
+						Parameters:         stdParams,
+					},
+					expResp: true,
+				},
+			},
+		},
+		{
+			name: "success adding request with different parameters",
+			requests: []testRequest{
+				{
+					request: &csi.CreateVolumeRequest{
+						Name:               "random-vol-name-foobar",
+						CapacityRange:      stdCapRange,
+						VolumeCapabilities: stdVolCap,
+						Parameters:         map[string]string{"foo": "bar"},
+					},
+					expResp: true,
+				},
+				{
+					request: &csi.CreateVolumeRequest{
+						Name:               "random-vol-name-foobar",
+						CapacityRange:      stdCapRange,
+						VolumeCapabilities: stdVolCap,
+					},
+					expResp: true,
+				},
+			},
+		},
+		{
+			name: "success adding request with different parameters",
+			requests: []testRequest{
+				{
+					request: &csi.CreateVolumeRequest{
+						Name:               "random-vol-name-foobar",
+						CapacityRange:      stdCapRange,
+						VolumeCapabilities: stdVolCap,
+						Parameters:         map[string]string{"foo": "bar"},
+					},
+					expResp: true,
+				},
+				{
+					request: &csi.CreateVolumeRequest{
+						Name:               "random-vol-name-foobar",
+						CapacityRange:      stdCapRange,
+						VolumeCapabilities: stdVolCap,
+						Parameters:         map[string]string{"foo": "baz"},
+					},
+					expResp: true,
+				},
+			},
+		},
+		{
+			name: "failure adding copy of request",
+			requests: []testRequest{
+				{
+					request: &csi.CreateVolumeRequest{
+						Name:               "random-vol-name",
+						CapacityRange:      stdCapRange,
+						VolumeCapabilities: stdVolCap,
+						Parameters:         stdParams,
+					},
+					expResp: true,
+				},
+				{
+					request: &csi.CreateVolumeRequest{
+						Name:               "random-vol-name",
+						CapacityRange:      stdCapRange,
+						VolumeCapabilities: stdVolCap,
+						Parameters:         stdParams,
+					},
+					expResp: false,
+				},
+			},
+		},
+		{
+			name: "success add, delete, add copy",
+			requests: []testRequest{
+				{
+					request: &csi.CreateVolumeRequest{
+						Name:               "random-vol-name",
+						CapacityRange:      stdCapRange,
+						VolumeCapabilities: stdVolCap,
+						Parameters:         stdParams,
+					},
+					expResp: true,
+				},
+				{
+					request: &csi.CreateVolumeRequest{
+						Name:               "random-vol-name",
+						CapacityRange:      stdCapRange,
+						VolumeCapabilities: stdVolCap,
+						Parameters:         stdParams,
+					},
+					expResp: false,
+					delete:  true,
+				},
+				{
+					request: &csi.CreateVolumeRequest{
+						Name:               "random-vol-name",
+						CapacityRange:      stdCapRange,
+						VolumeCapabilities: stdVolCap,
+						Parameters:         stdParams,
+					},
+					expResp: true,
+				},
+			},
+		},
+	}
+
+	for n := range testCases {
+		t.Run(testCases[n].name, func(t *testing.T) {
+			db := NewInFlight()
+			for _, r := range testCases[n].requests {
+				var resp bool
+				if r.delete {
+					db.Delete(r.request)
+				} else {
+					resp = db.Insert(r.request)
+				}
+				if r.expResp != resp {
+					t.Fatalf("expected insert to be %+v, got %+v", r.expResp, resp)
+				}
+			}
+		})
+
+	}
+}