digitalocean: add protokube support

pkg/client/simple/vfsclientset/clientset.go

@@ -112,6 +112,11 @@ func DeleteAllClusterState(basePath vfs.Path) error {
 		if err != nil {
 			return err
 		}
+
+		if relativePath == "" {
+			continue
+		}
+
 		if relativePath == "config" || relativePath == "cluster.spec" {
 			continue
 		}

pkg/resources/digitalocean/dns/dns.go

@@ -379,18 +379,29 @@ func (r *resourceRecordChangeset) Apply() error {
 			}
 
 			if !found {
-				return fmt.Errorf("could not find desired record to upsert")
-			}
+				recordCreateRequest := &godo.DomainRecordEditRequest{
+					Name: record.Name(),
+					Data: record.Rrdatas()[0],
+					TTL:  int(record.Ttl()),
+					Type: string(record.Type()),
+				}
+				err := createRecord(r.client, r.zone.Name(), recordCreateRequest)
+				if err != nil {
+					return fmt.Errorf("could not upsert records: %v", err)
+				}
 
-			domainEditRequest := &godo.DomainRecordEditRequest{
-				Name: record.Name(),
-				Data: record.Rrdatas()[0],
-				TTL:  int(record.Ttl()),
-				Type: string(record.Type()),
-			}
-			err := editRecord(r.client, r.zone.Name(), desiredRecord.ID, domainEditRequest)
-			if err != nil {
-				return fmt.Errorf("failed to edit record: %v", err)
+			} else {
+
+				domainEditRequest := &godo.DomainRecordEditRequest{
+					Name: record.Name(),
+					Data: record.Rrdatas()[0],
+					TTL:  int(record.Ttl()),
+					Type: string(record.Type()),
+				}
+				err := editRecord(r.client, r.zone.Name(), desiredRecord.ID, domainEditRequest)
+				if err != nil {
+					return fmt.Errorf("failed to edit record: %v", err)
+				}
 			}
 		}
 

protokube/cmd/protokube/main.go

@@ -68,7 +68,7 @@ func run() error {
 	flag.BoolVar(&containerized, "containerized", containerized, "Set if we are running containerized.")
 	flag.BoolVar(&initializeRBAC, "initialize-rbac", initializeRBAC, "Set if we should initialize RBAC")
 	flag.BoolVar(&master, "master", master, "Whether or not this node is a master")
-	flag.StringVar(&cloud, "cloud", "aws", "CloudProvider we are using (aws,gce)")
+	flag.StringVar(&cloud, "cloud", "aws", "CloudProvider we are using (aws,digitalocean,gce)")
 	flag.StringVar(&clusterID, "cluster-id", clusterID, "Cluster ID")
 	flag.StringVar(&dnsInternalSuffix, "dns-internal-suffix", dnsInternalSuffix, "DNS suffix for internal domain names")
 	flag.StringVar(&dnsServer, "dns-server", dnsServer, "DNS Server")
@@ -82,7 +82,7 @@ func run() error {
 	flag.StringVar(&tlsCert, "tls-cert", tlsCert, "Path to a file containing the certificate for etcd server")
 	flag.StringVar(&tlsKey, "tls-key", tlsKey, "Path to a file containing the private key for etcd server")
 	flags.StringSliceVarP(&zones, "zone", "z", []string{}, "Configure permitted zones and their mappings")
-	flags.StringVar(&dnsProviderID, "dns", "aws-route53", "DNS provider we should use (aws-route53, google-clouddns, coredns)")
+	flags.StringVar(&dnsProviderID, "dns", "aws-route53", "DNS provider we should use (aws-route53, google-clouddns, coredns, digitalocean)")
 	flags.StringVar(&etcdBackupImage, "etcd-backup-image", "", "Set to override the image for (experimental) etcd backups")
 	flags.StringVar(&etcdBackupStore, "etcd-backup-store", "", "Set to enable (experimental) etcd backups")
 	flags.StringVar(&etcdImageSource, "etcd-image", "k8s.gcr.io/etcd:2.2.1", "Etcd Source Container Registry")
@@ -114,6 +114,28 @@ func run() error {
 		if internalIP == nil {
 			internalIP = awsVolumes.InternalIP()
 		}
+	} else if cloud == "digitalocean" {
+		if clusterID == "" {
+			glog.Error("digitalocean requires --cluster-id")
+			os.Exit(1)
+		}
+
+		doVolumes, err := protokube.NewDOVolumes(clusterID)
+		if err != nil {
+			glog.Errorf("Error initializing DigitalOcean: %q", err)
+			os.Exit(1)
+		}
+
+		volumes = doVolumes
+
+		if internalIP == nil {
+			internalIP, err = protokube.GetDropletInternalIP()
+			if err != nil {
+				glog.Errorf("Error getting droplet internal IP: %s", err)
+				os.Exit(1)
+			}
+		}
+
 	} else if cloud == "gce" {
 		gceVolumes, err := protokube.NewGCEVolumes()
 		if err != nil {

protokube/pkg/protokube/BUILD.bazel

@@ -6,6 +6,7 @@ go_library(
         "aws_volume.go",
         "baremetal_volume.go",
         "channels.go",
+        "do_volume.go",
         "etcd_cluster.go",
         "etcd_manifest.go",
         "gce_volume.go",
@@ -30,6 +31,7 @@ go_library(
         "//dns-controller/pkg/dns:go_default_library",
         "//pkg/k8scodecs:go_default_library",
         "//pkg/kubemanifest:go_default_library",
+        "//pkg/resources/digitalocean:go_default_library",
         "//protokube/pkg/etcd:go_default_library",
         "//protokube/pkg/gossip:go_default_library",
         "//protokube/pkg/gossip/aws:go_default_library",
@@ -45,6 +47,7 @@ go_library(
         "//vendor/github.com/aws/aws-sdk-go/aws/request:go_default_library",
         "//vendor/github.com/aws/aws-sdk-go/aws/session:go_default_library",
         "//vendor/github.com/aws/aws-sdk-go/service/ec2:go_default_library",
+        "//vendor/github.com/digitalocean/godo:go_default_library",
         "//vendor/github.com/golang/glog:go_default_library",
         "//vendor/golang.org/x/net/context:go_default_library",
         "//vendor/golang.org/x/oauth2/google:go_default_library",

protokube/pkg/protokube/do_volume.go

@@ -0,0 +1,250 @@
+/*
+Copyright 2016 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 protokube
+
+import (
+	"context"
+	"fmt"
+	"io/ioutil"
+	"net"
+	"net/http"
+	"os"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/digitalocean/godo"
+
+	"k8s.io/kops/pkg/resources/digitalocean"
+	"k8s.io/kops/protokube/pkg/etcd"
+)
+
+const (
+	dropletRegionMetadataURL     = "http://169.254.169.254/metadata/v1/region"
+	dropletNameMetadataURL       = "http://169.254.169.254/metadata/v1/hostname"
+	dropletIDMetadataURL         = "http://169.254.169.254/metadata/v1/id"
+	dropletInternalIPMetadataURL = "http://169.254.169.254/metadata/v1/interfaces/private/0/ipv4/address"
+	localDevicePrefix            = "/dev/disk/by-id/scsi-0DO_Volume_"
+)
+
+type DOVolumes struct {
+	ClusterID string
+	Cloud     *digitalocean.Cloud
+
+	region      string
+	dropletName string
+	dropletID   int
+}
+
+var _ Volumes = &DOVolumes{}
+
+func NewDOVolumes(clusterID string) (*DOVolumes, error) {
+	region, err := getMetadataRegion()
+	if err != nil {
+		return nil, fmt.Errorf("failed to get droplet region: %s", err)
+	}
+
+	dropletID, err := getMetadataDropletID()
+	if err != nil {
+		return nil, fmt.Errorf("failed to get droplet id: %s", err)
+	}
+
+	dropletIDInt, err := strconv.Atoi(dropletID)
+	if err != nil {
+		return nil, fmt.Errorf("failed to convert droplet ID to int: %s", err)
+	}
+
+	dropletName, err := getMetadataDropletName()
+	if err != nil {
+		return nil, fmt.Errorf("failed to get droplet name: %s", err)
+	}
+
+	cloud, err := digitalocean.NewCloud(region)
+	if err != nil {
+		return nil, fmt.Errorf("failed to initialize digitalocean cloud: %s", err)
+	}
+
+	return &DOVolumes{
+		Cloud:       cloud,
+		ClusterID:   clusterID,
+		dropletID:   dropletIDInt,
+		dropletName: dropletName,
+		region:      region,
+	}, nil
+}
+
+func (d *DOVolumes) AttachVolume(volume *Volume) error {
+	for {
+		action, _, err := d.Cloud.VolumeActions().Attach(context.TODO(), volume.ID, d.dropletID)
+		if err != nil {
+			return fmt.Errorf("error attaching volume: %s", err)
+		}
+
+		if action.Status != godo.ActionInProgress && action.Status != godo.ActionCompleted {
+			return fmt.Errorf("invalid status for digitalocean volume: %s", volume.ID)
+		}
+
+		doVolume, err := d.getVolumeByID(volume.ID)
+		if err != nil {
+			return fmt.Errorf("error getting volume status: %s", err)
+		}
+
+		if len(doVolume.DropletIDs) == 1 {
+			if doVolume.DropletIDs[0] != d.dropletID {
+				return fmt.Errorf("digitalocean volume %s is attached to another droplet", doVolume.ID)
+			}
+
+			volume.LocalDevice = getLocalDeviceName(doVolume)
+			return nil
+		}
+
+		time.Sleep(10 * time.Second)
+	}
+
+}
+
+func (d *DOVolumes) FindVolumes() ([]*Volume, error) {
+	doVolumes, _, err := d.Cloud.Volumes().ListVolumes(context.TODO(), &godo.ListVolumeParams{Region: d.region})
+
+	if err != nil {
+		return nil, fmt.Errorf("failed to list volumes: %s", err)
+	}
+
+	var volumes []*Volume
+	for _, doVolume := range doVolumes {
+		// determine if this volume belongs to this cluster
+		// check for string d.ClusterID but with strings "." replaced with "-"
+		if !strings.Contains(doVolume.Name, strings.Replace(d.ClusterID, ".", "-", -1)) {
+			continue
+		}
+
+		vol := &Volume{
+			ID: doVolume.ID,
+			Info: VolumeInfo{
+				Description: doVolume.Description,
+			},
+		}
+
+		if len(doVolume.DropletIDs) == 1 {
+			vol.AttachedTo = strconv.Itoa(doVolume.DropletIDs[0])
+			vol.LocalDevice = getLocalDeviceName(&doVolume)
+		}
+
+		etcdClusterSpec, err := d.getEtcdClusterSpec(doVolume)
+		if err != nil {
+			return nil, fmt.Errorf("failed to get etcd cluster spec: %s", err)
+		}
+
+		vol.Info.EtcdClusters = append(vol.Info.EtcdClusters, etcdClusterSpec)
+		volumes = append(volumes, vol)
+	}
+
+	return volumes, nil
+}
+
+func (d *DOVolumes) FindMountedVolume(volume *Volume) (string, error) {
+	device := volume.LocalDevice
+
+	_, err := os.Stat(pathFor(device))
+	if err == nil {
+		return device, nil
+	}
+
+	if !os.IsNotExist(err) {
+		return "", fmt.Errorf("error checking for device %q: %v", device, err)
+	}
+
+	return "", nil
+}
+
+func (d *DOVolumes) getVolumeByID(id string) (*godo.Volume, error) {
+	vol, _, err := d.Cloud.Volumes().GetVolume(context.TODO(), id)
+	return vol, err
+
+}
+
+// getEtcdClusterSpec returns etcd.EtcdClusterSpec which holds
+// necessary information required for starting an etcd server.
+// DigitalOcean support on kops only supports single master setup for now
+// but in the future when it supports multiple masters this method be
+// updated to handle that case.
+// TODO: use tags once it's supported for volumes
+func (d *DOVolumes) getEtcdClusterSpec(vol godo.Volume) (*etcd.EtcdClusterSpec, error) {
+	nodeName := d.dropletName
+
+	var clusterKey string
+	if strings.Contains(vol.Name, "etcd-main") {
+		clusterKey = "main"
+	} else if strings.Contains(vol.Name, "etcd-events") {
+		clusterKey = "events"
+	} else {
+		return nil, fmt.Errorf("could not determine etcd cluster type for volume: %s", vol.Name)
+	}
+
+	return &etcd.EtcdClusterSpec{
+		ClusterKey: clusterKey,
+		NodeName:   nodeName,
+		NodeNames:  []string{nodeName},
+	}, nil
+}
+
+func getLocalDeviceName(vol *godo.Volume) string {
+	return localDevicePrefix + vol.Name
+}
+
+// GetDropletInternalIP gets the private IP of the droplet running this program
+// This function is exported so it can be called from protokube
+func GetDropletInternalIP() (net.IP, error) {
+	addr, err := getMetadata(dropletInternalIPMetadataURL)
+	if err != nil {
+		return nil, err
+	}
+
+	return net.ParseIP(addr), nil
+}
+
+func getMetadataRegion() (string, error) {
+	return getMetadata(dropletRegionMetadataURL)
+}
+
+func getMetadataDropletName() (string, error) {
+	return getMetadata(dropletNameMetadataURL)
+}
+
+func getMetadataDropletID() (string, error) {
+	return getMetadata(dropletIDMetadataURL)
+}
+
+func getMetadata(url string) (string, error) {
+	resp, err := http.Get(url)
+	if err != nil {
+		return "", err
+	}
+
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		return "", fmt.Errorf("droplet metadata returned non-200 status code: %d", resp.StatusCode)
+	}
+
+	bodyBytes, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		return "", err
+	}
+
+	return string(bodyBytes), nil
+}

tests/integration/update_cluster/additional_user-data/cloudformation.json.extracted.yaml

@@ -35,6 +35,8 @@ Resources.AWSAutoScalingLaunchConfigurationmasterustest1amastersadditionaluserda
 
 
 
+
+
   function ensure-install-dir() {
     INSTALL_DIR="/var/cache/kubernetes-install"
     # On ContainerOS, we install to /var/lib/toolbox install (because of noexec)
@@ -333,6 +335,8 @@ Resources.AWSAutoScalingLaunchConfigurationnodesadditionaluserdataexamplecom.Pro
 
 
 
+
+
   function ensure-install-dir() {
     INSTALL_DIR="/var/cache/kubernetes-install"
     # On ContainerOS, we install to /var/lib/toolbox install (because of noexec)

tests/integration/update_cluster/minimal-cloudformation/cloudformation.json.extracted.yaml

@@ -26,6 +26,8 @@ Resources.AWSAutoScalingLaunchConfigurationmasterustest1amastersminimalexampleco
 
 
 
+
+
   function ensure-install-dir() {
     INSTALL_DIR="/var/cache/kubernetes-install"
     # On ContainerOS, we install to /var/lib/toolbox install (because of noexec)
@@ -303,6 +305,8 @@ Resources.AWSAutoScalingLaunchConfigurationnodesminimalexamplecom.Properties.Use
 
 
 
+
+
   function ensure-install-dir() {
     INSTALL_DIR="/var/cache/kubernetes-install"
     # On ContainerOS, we install to /var/lib/toolbox install (because of noexec)