implement volume task

upup/pkg/fi/cloudup/openstack/cloud.go

@@ -18,25 +18,86 @@ package openstack
 
 import (
 	"fmt"
+	"time"
 
+	"github.com/golang/glog"
+	"github.com/gophercloud/gophercloud"
+	os "github.com/gophercloud/gophercloud/openstack"
+	cinder "github.com/gophercloud/gophercloud/openstack/blockstorage/v2/volumes"
 	"k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/kops/pkg/apis/kops"
 	"k8s.io/kops/pkg/cloudinstances"
 	"k8s.io/kops/upup/pkg/fi"
+	"k8s.io/kops/util/pkg/vfs"
 	"k8s.io/kubernetes/federation/pkg/dnsprovider"
 )
 
+// readBackoff is the backoff strategy for openstack read retries.
+var readBackoff = wait.Backoff{
+	Duration: time.Second,
+	Factor:   1.5,
+	Jitter:   0.1,
+	Steps:    4,
+}
+
+// writeBackoff is the backoff strategy for openstack write retries.
+var writeBackoff = wait.Backoff{
+	Duration: time.Second,
+	Factor:   1.5,
+	Jitter:   0.1,
+	Steps:    5,
+}
+
 type OpenstackCloud interface {
 	fi.Cloud
+
+	// SetVolumeTags will set the tags for the Cinder volume
+	SetVolumeTags(id string, tags map[string]string) error
+
+	// GetCloudTags will return the tags attached on cloud
+	GetCloudTags() map[string]string
+
+	// ListVolumes will return the Cinder volumes which match the options
+	ListVolumes(opt cinder.ListOpts) ([]cinder.Volume, error)
+
+	// CreateVolume will create a new Cinder Volume
+	CreateVolume(opt cinder.CreateOpts) (*cinder.Volume, error)
 }
 
 type openstackCloud struct {
+	cinderClient *gophercloud.ServiceClient
+	tags         map[string]string
 }
 
 var _ fi.Cloud = &openstackCloud{}
 
-func NewOpenstackCloud() (OpenstackCloud, error) {
+func NewOpenstackCloud(tags map[string]string) (OpenstackCloud, error) {
-	return &openstackCloud{}, nil
+	config := vfs.OpenstackConfig{}
+
+	authOption, err := config.GetCredential()
+	if err != nil {
+		return nil, err
+	}
+	provider, err := os.AuthenticatedClient(authOption)
+	if err != nil {
+		return nil, fmt.Errorf("error building openstack authenticated client: %v", err)
+	}
+
+	endpointOpt, err := config.GetServiceConfig("Cinder")
+	if err != nil {
+		return nil, err
+	}
+	cinderClient, err := os.NewBlockStorageV2(provider, endpointOpt)
+	if err != nil {
+		return nil, fmt.Errorf("error building swift client: %v", err)
+	}
+
+	c := &openstackCloud{
+		cinderClient: cinderClient,
+		tags:         tags,
+	}
+	return c, nil
 }
 
 func (c *openstackCloud) ProviderID() kops.CloudProviderID {
@@ -62,3 +123,79 @@ func (c *openstackCloud) DeleteGroup(g *cloudinstances.CloudInstanceGroup) error
 func (c *openstackCloud) GetCloudGroups(cluster *kops.Cluster, instancegroups []*kops.InstanceGroup, warnUnmatched bool, nodes []v1.Node) (map[string]*cloudinstances.CloudInstanceGroup, error) {
 	return nil, fmt.Errorf("openstackCloud::GetCloudGroups not implemented")
 }
+
+func (c *openstackCloud) SetVolumeTags(id string, tags map[string]string) error {
+	if len(tags) == 0 {
+		return nil
+	}
+	if id == "" {
+		return fmt.Errorf("error setting tags to unknown volume")
+	}
+	glog.V(4).Infof("setting tags to cinder volume %q: %v", id, tags)
+
+	opt := cinder.UpdateOpts{Metadata: tags}
+	done, err := vfs.RetryWithBackoff(writeBackoff, func() (bool, error) {
+		_, err := cinder.Update(c.cinderClient, id, opt).Extract()
+		if err != nil {
+			return false, fmt.Errorf("error setting tags to cinder volume %q: %v", id, err)
+		}
+		return true, nil
+	})
+	if err != nil {
+		return err
+	} else if done {
+		return nil
+	} else {
+		return wait.ErrWaitTimeout
+	}
+}
+
+func (c *openstackCloud) GetCloudTags() map[string]string {
+	return c.tags
+}
+
+func (c *openstackCloud) ListVolumes(opt cinder.ListOpts) ([]cinder.Volume, error) {
+	var volumes []cinder.Volume
+
+	done, err := vfs.RetryWithBackoff(readBackoff, func() (bool, error) {
+		allPages, err := cinder.List(c.cinderClient, opt).AllPages()
+		if err != nil {
+			return false, fmt.Errorf("error listing volumes %v: %v", opt, err)
+		}
+
+		vs, err := cinder.ExtractVolumes(allPages)
+		if err != nil {
+			return false, fmt.Errorf("error extracting volumes: %v", err)
+		}
+		volumes = vs
+		return true, nil
+	})
+	if err != nil {
+		return volumes, err
+	} else if done {
+		return volumes, nil
+	} else {
+		return volumes, wait.ErrWaitTimeout
+	}
+}
+
+func (c *openstackCloud) CreateVolume(opt cinder.CreateOpts) (*cinder.Volume, error) {
+	var volume *cinder.Volume
+
+	done, err := vfs.RetryWithBackoff(writeBackoff, func() (bool, error) {
+		v, err := cinder.Create(c.cinderClient, opt).Extract()
+		if err != nil {
+			return false, fmt.Errorf("error creating volume %v: %v", opt, err)
+		}
+		volume = v
+		return true, nil
+	})
+	if err != nil {
+		return volume, err
+	} else if done {
+		return volume, nil
+	} else {
+		return volume, wait.ErrWaitTimeout
+	}
+
+}

upup/pkg/fi/cloudup/openstacktasks/volume.go

@@ -0,0 +1,145 @@
+/*
+Copyright 2017 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 openstacktasks
+
+import (
+	"fmt"
+
+	"github.com/golang/glog"
+	cinder "github.com/gophercloud/gophercloud/openstack/blockstorage/v2/volumes"
+	"k8s.io/kops/upup/pkg/fi"
+	"k8s.io/kops/upup/pkg/fi/cloudup/openstack"
+)
+
+type Volume struct {
+	ID               *string
+	Name             *string
+	AvailabilityZone *string
+	VolumeType       *string
+	SizeGB           *int64
+	Tags             map[string]string
+	Lifecycle        *fi.Lifecycle
+}
+
+var _ fi.CompareWithID = &Volume{}
+
+func (c *Volume) CompareWithID() *string {
+	return c.ID
+}
+
+func (c *Volume) Find(context *fi.Context) (*Volume, error) {
+	cloud := context.Cloud.(openstack.OpenstackCloud)
+	opt := cinder.ListOpts{
+		Name:     fi.StringValue(c.Name),
+		Metadata: cloud.GetCloudTags(),
+	}
+	volumes, err := cloud.ListVolumes(opt)
+	if err != nil {
+		return nil, err
+	}
+	n := len(volumes)
+	if n == 0 {
+		return nil, nil
+	} else if n != 1 {
+		return nil, fmt.Errorf("found multiple Volumes with name: %s", fi.StringValue(c.Name))
+	}
+	v := volumes[0]
+	actual := &Volume{
+		ID:               fi.String(v.ID),
+		Name:             fi.String(v.Name),
+		AvailabilityZone: fi.String(v.AvailabilityZone),
+		VolumeType:       fi.String(v.VolumeType),
+		SizeGB:           fi.Int64(int64(v.Size)),
+		Tags:             v.Metadata,
+		Lifecycle:        c.Lifecycle,
+	}
+	return actual, nil
+}
+
+func (c *Volume) Run(context *fi.Context) error {
+	cloud := context.Cloud.(openstack.OpenstackCloud)
+	for k, v := range cloud.GetCloudTags() {
+		c.Tags[k] = v
+	}
+
+	return fi.DefaultDeltaRunMethod(c, context)
+}
+
+func (_ *Volume) CheckChanges(a, e, changes *Volume) error {
+	if a == nil {
+		if e.Name == nil {
+			return fi.RequiredField("Name")
+		}
+		if e.AvailabilityZone == nil {
+			return fi.RequiredField("AvailabilityZone")
+		}
+		if e.VolumeType == nil {
+			return fi.RequiredField("VolumeType")
+		}
+		if e.SizeGB == nil {
+			return fi.RequiredField("SizeGB")
+		}
+	} else {
+		if changes.ID != nil {
+			return fi.CannotChangeField("ID")
+		}
+		if changes.AvailabilityZone != nil {
+			return fi.CannotChangeField("AvailabilityZone")
+		}
+		if changes.VolumeType != nil {
+			return fi.CannotChangeField("VolumeType")
+		}
+		if changes.SizeGB != nil {
+			return fi.CannotChangeField("SizeGB")
+		}
+	}
+	return nil
+}
+
+func (_ *Volume) RenderOpenstack(t *openstack.OpenstackAPITarget, a, e, changes *Volume) error {
+	if a == nil {
+		glog.V(2).Infof("Creating PersistentVolume with Name:%q", fi.StringValue(e.Name))
+
+		opt := cinder.CreateOpts{
+			Size:             int(*e.SizeGB),
+			AvailabilityZone: fi.StringValue(e.AvailabilityZone),
+			Metadata:         e.Tags,
+			Name:             fi.StringValue(e.Name),
+			VolumeType:       fi.StringValue(e.VolumeType),
+		}
+
+		v, err := t.Cloud.CreateVolume(opt)
+		if err != nil {
+			return fmt.Errorf("error creating PersistentVolume: %v", err)
+		}
+
+		e.ID = fi.String(v.ID)
+		return nil
+	}
+
+	if changes != nil && changes.Tags != nil {
+		glog.V(2).Infof("Update the tags on volume %q: %v, the differences are %v", fi.StringValue(e.ID), e.Tags, changes.Tags)
+
+		err := t.Cloud.SetVolumeTags(fi.StringValue(e.ID), e.Tags)
+		if err != nil {
+			return fmt.Errorf("error updating the tags on volume %q: %v", fi.StringValue(e.ID), err)
+		}
+	}
+
+	glog.V(2).Infof("Openstack task Volume::RenderOpenstack did nothing")
+	return nil
+}