[Digital Ocean] Handle logic for kops edit/update cluster

pkg/commands/status_discovery.go

@@ -107,5 +107,9 @@ func (s *CloudDiscoveryStatusStore) FindClusterStatus(cluster *kops.Cluster) (*k
 	if osCloud, ok := cloud.(openstack.OpenstackCloud); ok {
 		return osCloud.FindClusterStatus(cluster)
 	}
+
+	if doCloud, ok := cloud.(*digitalocean.Cloud); ok {
+		return doCloud.FindClusterStatus(cluster)
+	}
 	return nil, fmt.Errorf("etcd Status not implemented for %T", cloud)
 }

pkg/resources/digitalocean/BUILD.bazel

@@ -15,6 +15,7 @@ go_library(
         "//pkg/cloudinstances:go_default_library",
         "//pkg/resources:go_default_library",
         "//pkg/resources/digitalocean/dns:go_default_library",
+        "//protokube/pkg/etcd:go_default_library",
         "//upup/pkg/fi:go_default_library",
         "//vendor/github.com/digitalocean/godo:go_default_library",
         "//vendor/golang.org/x/oauth2:go_default_library",

pkg/resources/digitalocean/cloud.go

@@ -32,9 +32,13 @@ import (
 	"k8s.io/kops/pkg/apis/kops"
 	"k8s.io/kops/pkg/cloudinstances"
 	"k8s.io/kops/pkg/resources/digitalocean/dns"
+	"k8s.io/kops/protokube/pkg/etcd"
 	"k8s.io/kops/upup/pkg/fi"
 )
 
+const TagKubernetesClusterIndex = "k8s-index"
+const TagKubernetesClusterNamePrefix = "KubernetesCluster"
+
 // TokenSource implements oauth2.TokenSource
 type TokenSource struct {
 	AccessToken string
@@ -139,6 +143,10 @@ func (c *Cloud) LoadBalancers() godo.LoadBalancersService {
 	return c.Client.LoadBalancers
 }
 
+func (c *Cloud) GetAllLoadBalancers() ([]godo.LoadBalancer, error) {
+	return getAllLoadBalancers(c)
+}
+
 // FindVPCInfo is not implemented, it's only here to satisfy the fi.Cloud interface
 func (c *Cloud) FindVPCInfo(id string) (*fi.VPCInfo, error) {
 	return nil, errors.New("not implemented")
@@ -175,3 +183,98 @@ func (c *Cloud) GetApiIngressStatus(cluster *kops.Cluster) ([]kops.ApiIngressSta
 
 	return nil, nil
 }
+
+// FindClusterStatus discovers the status of the cluster, by looking for the tagged etcd volumes
+func (c *Cloud) FindClusterStatus(cluster *kops.Cluster) (*kops.ClusterStatus, error) {
+	etcdStatus, err := findEtcdStatus(c, cluster)
+	if err != nil {
+		return nil, err
+	}
+	status := &kops.ClusterStatus{
+		EtcdClusters: etcdStatus,
+	}
+	klog.V(2).Infof("Cluster status (from cloud): %v", fi.DebugAsJsonString(status))
+	return status, nil
+}
+
+// findEtcdStatus discovers the status of etcd, by looking for the tagged etcd volumes
+func findEtcdStatus(c *Cloud, cluster *kops.Cluster) ([]kops.EtcdClusterStatus, error) {
+	statusMap := make(map[string]*kops.EtcdClusterStatus)
+	volumes, err := getAllVolumesByRegion(c, c.RegionName)
+
+	if err != nil {
+		return nil, fmt.Errorf("failed to get all volumes by region from %s: %v", c.RegionName, err)
+	}
+
+	for _, volume := range volumes {
+		volumeID := volume.ID
+
+		etcdClusterName := ""
+		var etcdClusterSpec *etcd.EtcdClusterSpec
+
+		for _, myTag := range volume.Tags {
+			klog.V(8).Infof("findEtcdStatus status (from cloud): checking if volume with tag %q belongs to cluster", myTag)
+			// check if volume belongs to this cluster.
+			// tag will be in the format "KubernetesCluster:dev5-k8s-local" (where clusterName is dev5.k8s.local)
+			clusterName := strings.Replace(cluster.Name, ".", "-", -1)
+			if strings.Contains(myTag, fmt.Sprintf("%s:%s", TagKubernetesClusterNamePrefix, clusterName)) {
+				klog.V(10).Infof("findEtcdStatus cluster comparison matched for tag: %v", myTag)
+				// this volume belongs to our cluster, add this to our etcdClusterSpec.
+				// loop through the tags again and
+				for _, volumeTag := range volume.Tags {
+					if strings.Contains(volumeTag, TagKubernetesClusterIndex) {
+						volumeTagParts := strings.Split(volumeTag, ":")
+						if len(volumeTagParts) < 2 {
+							return nil, fmt.Errorf("volume tag split failed, too few components for tag %q on volume %q", volumeTag, volume)
+						}
+						dropletIndex := volumeTagParts[1]
+						etcdClusterSpec, err = c.getEtcdClusterSpec(volume.Name, dropletIndex)
+						if err != nil {
+							return nil, fmt.Errorf("error parsing etcd cluster tag %q on volume %q: %v", volumeTag, volumeID, err)
+						}
+
+						klog.V(10).Infof("findEtcdStatus etcdClusterSpec: %v", fi.DebugAsJsonString(etcdClusterSpec))
+						etcdClusterName = etcdClusterSpec.ClusterKey
+						status := statusMap[etcdClusterName]
+						if status == nil {
+							status = &kops.EtcdClusterStatus{
+								Name: etcdClusterName,
+							}
+							statusMap[etcdClusterName] = status
+						}
+
+						memberName := etcdClusterSpec.NodeName
+						status.Members = append(status.Members, &kops.EtcdMemberStatus{
+							Name:     memberName,
+							VolumeId: volume.ID,
+						})
+					}
+				}
+			}
+		}
+	}
+
+	status := make([]kops.EtcdClusterStatus, 0, len(statusMap))
+	for _, v := range statusMap {
+		status = append(status, *v)
+	}
+
+	return status, nil
+}
+
+func (c *Cloud) getEtcdClusterSpec(volumeName string, dropletName string) (*etcd.EtcdClusterSpec, error) {
+	var clusterKey string
+	if strings.Contains(volumeName, "etcd-main") {
+		clusterKey = "main"
+	} else if strings.Contains(volumeName, "etcd-events") {
+		clusterKey = "events"
+	} else {
+		return nil, fmt.Errorf("could not determine etcd cluster type for volume: %s", volumeName)
+	}
+
+	return &etcd.EtcdClusterSpec{
+		ClusterKey: clusterKey,
+		NodeName:   dropletName,
+		NodeNames:  []string{dropletName},
+	}, nil
+}

upup/pkg/fi/cloudup/dotasks/loadbalancer.go

@@ -21,6 +21,7 @@ import (
 	"errors"
 	"fmt"
 	"net"
+	"strings"
 	"time"
 
 	"github.com/digitalocean/godo"
@@ -49,6 +50,7 @@ func (lb *LoadBalancer) CompareWithID() *string {
 }
 
 func (lb *LoadBalancer) Find(c *fi.Context) (*LoadBalancer, error) {
+	klog.V(10).Infof("load balancer FIND - ID=%s, name=%s", fi.StringValue(lb.ID), fi.StringValue(lb.Name))
 	if fi.StringValue(lb.ID) == "" {
 		// Loadbalancer = nil if not found
 		return nil, nil
@@ -97,7 +99,6 @@ func (_ *LoadBalancer) CheckChanges(a, e, changes *LoadBalancer) error {
 }
 
 func (_ *LoadBalancer) RenderDO(t *do.DOAPITarget, a, e, changes *LoadBalancer) error {
-
 	Rules := []godo.ForwardingRule{
 		{
 			EntryProtocol:  "https",
@@ -124,6 +125,24 @@ func (_ *LoadBalancer) RenderDO(t *do.DOAPITarget, a, e, changes *LoadBalancer)
 		HealthyThreshold:       5,
 	}
 
+	// check if load balancer exist.
+	loadBalancers, err := t.Cloud.GetAllLoadBalancers()
+
+	if err != nil {
+		return fmt.Errorf("LoadBalancers.List returned error: %v", err)
+	}
+
+	for _, loadbalancer := range loadBalancers {
+		klog.V(10).Infof("load balancer retrieved=%s, e.Name=%s", loadbalancer.Name, fi.StringValue(e.Name))
+		if strings.Contains(loadbalancer.Name, fi.StringValue(e.Name)) {
+			// load balancer already exists.
+			e.ID = fi.String(loadbalancer.ID)
+			e.IPAddress = fi.String(loadbalancer.IP) // This will be empty on create, but will be filled later on FindIPAddress invokation.
+			return nil
+		}
+	}
+
+	// load balancer doesn't exist. Create one.
 	klog.V(10).Infof("Creating load balancer for DO")
 
 	loadBalancerService := t.Cloud.LoadBalancers()
@@ -150,18 +169,40 @@ func (lb *LoadBalancer) FindIPAddress(c *fi.Context) (*string, error) {
 	cloud := c.Cloud.(*digitalocean.Cloud)
 	loadBalancerService := cloud.LoadBalancers()
 
-	klog.V(10).Infof("Find IP address for load balancer ID=%s", fi.StringValue(lb.ID))
-	loadBalancer, _, err := loadBalancerService.Get(context.TODO(), fi.StringValue(lb.ID))
-	if err != nil {
-		klog.Errorf("Error fetching load balancer with Name=%s", fi.StringValue(lb.Name))
-		return nil, err
-	}
+	if len(fi.StringValue(lb.ID)) > 0 {
+		// able to retrieve ID.
+		klog.V(10).Infof("Find IP address for load balancer ID=%s", fi.StringValue(lb.ID))
+		loadBalancer, _, err := loadBalancerService.Get(context.TODO(), fi.StringValue(lb.ID))
+		if err != nil {
+			klog.Errorf("Error fetching load balancer with Name=%s", fi.StringValue(lb.Name))
+			return nil, err
+		}
+
+		address := loadBalancer.IP
+
+		if isIPv4(address) {
+			klog.V(10).Infof("load balancer address=%s", address)
+			return &address, nil
+		}
+	} else {
+		// check with the name.
+		// check if load balancer exist.
+		loadBalancers, err := cloud.GetAllLoadBalancers()
 
-	address := loadBalancer.IP
+		if err != nil {
+			return nil, fmt.Errorf("LoadBalancers.List returned error: %v", err)
+		}
 
-	if isIPv4(address) {
-		klog.V(10).Infof("load balancer address=%s", address)
-		return &address, nil
+		for _, loadbalancer := range loadBalancers {
+			if strings.Contains(loadbalancer.Name, fi.StringValue(lb.Name)) {
+				// load balancer already exists.
+				address := loadbalancer.IP
+				if isIPv4(address) {
+					klog.V(10).Infof("load balancer address=%s", address)
+					return &address, nil
+				}
+			}
+		}
 	}
 
 	const lbWaitTime = 10 * time.Second