Topology support

Gopkg.toml

@@ -76,7 +76,7 @@
   non-go = true
 
 [[constraint]]
-  branch = "master"
+  branch = "v0.3.0"
   name = "github.com/kubernetes-csi/csi-test"
 
 [[constraint]]

cmd/main.go

@@ -16,11 +16,14 @@ package main
 
 import (
 	"flag"
+	"math/rand"
 	"os"
+	"time"
 
 	"github.com/golang/glog"
 
-	gce "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/gce-cloud-provider"
+	gce "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/gce-cloud-provider/compute"
+	metadataservice "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/gce-cloud-provider/metadata"
 	driver "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/gce-pd-csi-driver"
 	mountmanager "sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/mount-manager"
 )
@@ -38,6 +41,7 @@ var (
 
 func main() {
 	flag.Parse()
+	rand.Seed(time.Now().UnixNano())
 	handle()
 	os.Exit(0)
 }
@@ -59,7 +63,12 @@ func handle() {
 	mounter := mountmanager.NewSafeMounter()
 	deviceUtils := mountmanager.NewDeviceUtils()
 
-	err = gceDriver.SetupGCEDriver(cloudProvider, mounter, deviceUtils, *driverName, *nodeID, vendorVersion)
+	ms, err := metadataservice.NewMetadataService()
+	if err != nil {
+		glog.Fatalf("Failed to set up metadata service: %v", err)
+	}
+
+	err = gceDriver.SetupGCEDriver(cloudProvider, mounter, deviceUtils, ms, *driverName, *nodeID, vendorVersion)
 	if err != nil {
 		glog.Fatalf("Failed to initialize GCE CSI Driver: %v", err)
 	}

pkg/common/constants.go

@@ -0,0 +1,26 @@
+/*
+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 common
+
+const (
+	// Keys for Storage Class Parameters
+	ParameterKeyZone = "zone"
+	ParameterKeyType = "type"
+
+	// Keys for Topology. This key will be shared amonst drivers from GCP
+	TopologyKeyZone = "com.google.topology/zone"
+)

pkg/utils/utils.go → pkg/common/utils.go

@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package utils
+package common
 
 import (
 	"fmt"

pkg/gce-cloud-provider/fake-gce.go → pkg/gce-cloud-provider/compute/fake-gce.go

@@ -21,9 +21,10 @@ import (
 	"github.com/golang/glog"
 	"golang.org/x/net/context"
 	compute "google.golang.org/api/compute/v1"
+	"google.golang.org/api/googleapi"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
-	"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/utils"
+	"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/common"
 )
 
 type FakeCloudProvider struct {
@@ -58,7 +59,7 @@ func (cloud *FakeCloudProvider) GetZone() string {
 func (cloud *FakeCloudProvider) GetDiskOrError(ctx context.Context, volumeZone, volumeName string) (*compute.Disk, error) {
 	disk, ok := cloud.disks[volumeName]
 	if !ok {
-		return nil, fmt.Errorf("Disk %v not found", volumeName)
+		return nil, notFoundError()
 	}
 	return disk, nil
 }
@@ -71,12 +72,12 @@ func (cloud *FakeCloudProvider) GetAndValidateExistingDisk(ctx context.Context,
 	}
 	if disk != nil {
 		// Check that disk is the same
-		requestValid := utils.GbToBytes(disk.SizeGb) >= reqBytes || reqBytes == 0
-		responseValid := utils.GbToBytes(disk.SizeGb) <= limBytes || limBytes == 0
+		requestValid := common.GbToBytes(disk.SizeGb) >= reqBytes || reqBytes == 0
+		responseValid := common.GbToBytes(disk.SizeGb) <= limBytes || limBytes == 0
 		if !requestValid || !responseValid {
 			return true, status.Error(codes.AlreadyExists, fmt.Sprintf(
 				"Disk already exists with incompatible capacity. Need %v (Required) < %v (Existing) < %v (Limit)",
-				reqBytes, utils.GbToBytes(disk.SizeGb), limBytes))
+				reqBytes, common.GbToBytes(disk.SizeGb), limBytes))
 		}
 
 		respType := strings.Split(disk.Type, "/")
@@ -171,7 +172,7 @@ func (cloud *FakeCloudProvider) InsertInstance(instance *compute.Instance, insta
 func (cloud *FakeCloudProvider) GetInstanceOrError(ctx context.Context, instanceZone, instanceName string) (*compute.Instance, error) {
 	instance, ok := cloud.instances[instanceName]
 	if !ok {
-		return nil, fmt.Errorf("Could not find instance %v", instanceName)
+		return nil, notFoundError()
 	}
 	return instance, nil
 }
@@ -180,3 +181,13 @@ func (cloud *FakeCloudProvider) GetInstanceOrError(ctx context.Context, instance
 func (cloud *FakeCloudProvider) WaitForOp(ctx context.Context, op *compute.Operation, zone string) error {
 	return nil
 }
+
+func notFoundError() *googleapi.Error {
+	return &googleapi.Error{
+		Errors: []googleapi.ErrorItem{
+			{
+				Reason: "notFound",
+			},
+		},
+	}
+}

pkg/gce-cloud-provider/gce-compute.go → pkg/gce-cloud-provider/compute/gce-compute.go

@@ -25,7 +25,7 @@ import (
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 	"k8s.io/apimachinery/pkg/util/wait"
-	"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/utils"
+	"sigs.k8s.io/gcp-compute-persistent-disk-csi-driver/pkg/common"
 )
 
 type GCECompute interface {
@@ -62,11 +62,7 @@ func (cloud *CloudProvider) GetDiskOrError(ctx context.Context, volumeZone, volu
 	glog.Infof("Getting disk %v from zone %v", volumeName, volumeZone)
 	disk, err := svc.Disks.Get(project, volumeZone, volumeName).Context(ctx).Do()
 	if err != nil {
-		if IsGCEError(err, "notFound") {
-			return nil, status.Error(codes.NotFound, fmt.Sprintf("disk %v does not exist", volumeName))
-		}
-
-		return nil, status.Error(codes.Internal, fmt.Sprintf("unknown disk GET error: %v", err))
+		return nil, err
 	}
 	glog.Infof("Got disk %v from zone %v", volumeName, volumeZone)
 	return disk, nil
@@ -86,12 +82,12 @@ func (cloud *CloudProvider) GetAndValidateExistingDisk(ctx context.Context, conf
 
 	if resp != nil {
 		// Disk already exists
-		requestValid := utils.GbToBytes(resp.SizeGb) >= reqBytes && reqBytes != 0
-		responseValid := utils.GbToBytes(resp.SizeGb) <= limBytes && limBytes != 0
+		requestValid := common.GbToBytes(resp.SizeGb) >= reqBytes && reqBytes != 0
+		responseValid := common.GbToBytes(resp.SizeGb) <= limBytes && limBytes != 0
 		if !requestValid || !responseValid {
 			return true, status.Error(codes.AlreadyExists, fmt.Sprintf(
 				"Disk already exists with incompatible capacity. Need %v (Required) < %v (Existing) < %v (Limit)",
-				reqBytes, utils.GbToBytes(resp.SizeGb), limBytes))
+				reqBytes, common.GbToBytes(resp.SizeGb), limBytes))
 		}
 
 		respType := strings.Split(resp.Type, "/")
@@ -190,11 +186,7 @@ func (cloud *CloudProvider) GetInstanceOrError(ctx context.Context, instanceZone
 	glog.Infof("Getting instance %v from zone %v", instanceName, instanceZone)
 	instance, err := svc.Instances.Get(project, instanceZone, instanceName).Do()
 	if err != nil {
-		if IsGCEError(err, "notFound") {
-			return nil, status.Error(codes.NotFound, fmt.Sprintf("instance %v does not exist", instanceName))
-		}
-
-		return nil, status.Error(codes.Internal, fmt.Sprintf("unknown instance GET error: %v", err))
+		return nil, err
 	}
 	glog.Infof("Got instance %v from zone %v", instanceName, instanceZone)
 	return instance, nil

pkg/gce-cloud-provider/gce.go → pkg/gce-cloud-provider/compute/gce.go

@@ -51,7 +51,6 @@ func CreateCloudProvider(vendorVersion string) (*CloudProvider, error) {
 	if err != nil {
 		return nil, err
 	}
-	// TODO: Use metadata server or flags to retrieve project and zone. Fallback on flag if necessary
 
 	project, zone, err := getProjectAndZoneFromMetadata()
 	if err != nil {

pkg/gce-cloud-provider/metadata/fake.go

@@ -0,0 +1,33 @@
+/*
+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 metadata
+
+type fakeServiceManager struct{}
+
+var _ MetadataService = &fakeServiceManager{}
+
+func NewFakeService() MetadataService {
+	return &fakeServiceManager{}
+}
+
+func (manager *fakeServiceManager) GetZone() string {
+	return "test-location"
+}
+
+func (manager *fakeServiceManager) GetProject() string {
+	return "test-project"
+}

pkg/gce-cloud-provider/metadata/metadata.go

@@ -0,0 +1,62 @@
+/*
+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 metadata
+
+import (
+	"fmt"
+
+	"cloud.google.com/go/compute/metadata"
+)
+
+// MetadataService is a fakeable interface exposing necessary data
+// from the GCE Metadata service
+type MetadataService interface {
+	GetZone() string
+	GetProject() string
+}
+
+type metadataServiceManager struct {
+	// Current zone the driver is running in
+	zone    string
+	project string
+}
+
+var _ MetadataService = &metadataServiceManager{}
+
+func NewMetadataService() (MetadataService, error) {
+	zone, err := metadata.Zone()
+	if err != nil {
+		return nil, fmt.Errorf("failed to get current zone: %v", err)
+	}
+	projectID, err := metadata.ProjectID()
+	if err != nil {
+		return nil, fmt.Errorf("failed to get project: %v", err)
+	}
+
+	return &metadataServiceManager{
+		project: projectID,
+		zone:    zone,
+	}, nil
+}
+
+func (manager *metadataServiceManager) GetZone() string {
+	return manager.zone
+}
+
+func (manager *metadataServiceManager) GetProject() string {
+	return manager.project
+}