digitalocean: list/delete resources

pkg/resources/digitalocean/BUILD.bazel

@@ -4,7 +4,6 @@ go_library(
     name = "go_default_library",
     srcs = [
         "cloud.go",
-        "do.go",
         "resources.go",
     ],
     importpath = "k8s.io/kops/pkg/resources/digitalocean",

pkg/resources/digitalocean/dns/BUILD.bazel

@@ -6,6 +6,7 @@ go_library(
     importpath = "k8s.io/kops/pkg/resources/digitalocean/dns",
     visibility = ["//visibility:public"],
     deps = [
+        "//dns-controller/pkg/dns:go_default_library",
         "//dnsprovider/pkg/dnsprovider:go_default_library",
         "//dnsprovider/pkg/dnsprovider/rrstype:go_default_library",
         "//vendor/github.com/digitalocean/godo:go_default_library",

pkg/resources/digitalocean/resources.go

@@ -17,21 +17,164 @@ limitations under the License.
 package digitalocean
 
 import (
+	"context"
+	"errors"
+	"fmt"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/digitalocean/godo"
+
 	"k8s.io/kops/pkg/resources"
 	"k8s.io/kops/upup/pkg/fi"
 )
 
-type Resources struct {
-	Cloud       fi.Cloud
-	ClusterName string
+const (
+	resourceTypeDroplet = "droplet"
+	resourceTypeVolume  = "volume"
+)
+
+type listFn func(fi.Cloud, string) ([]*resources.Resource, error)
+
+func ListResources(cloud *Cloud, clusterName string) (map[string]*resources.Resource, error) {
+	resourceTrackers := make(map[string]*resources.Resource)
+
+	listFunctions := []listFn{
+		listVolumes,
+		listDroplets,
+	}
+
+	for _, fn := range listFunctions {
+		rt, err := fn(cloud, clusterName)
+		if err != nil {
+			return nil, err
+		}
+		for _, t := range rt {
+			resourceTrackers[t.Type+":"+t.ID] = t
+		}
+	}
+
+	return resourceTrackers, nil
+}
+
+func listDroplets(cloud fi.Cloud, clusterName string) ([]*resources.Resource, error) {
+	c := cloud.(*Cloud)
+	var resourceTrackers []*resources.Resource
+
+	clusterTag := "KubernetesCluster:" + strings.Replace(clusterName, ".", "-", -1)
+
+	droplets, _, err := c.Droplets().ListByTag(context.TODO(), clusterTag, &godo.ListOptions{})
+	if err != nil {
+		return nil, fmt.Errorf("failed to list droplets: %v", err)
+	}
+
+	for _, droplet := range droplets {
+		resourceTracker := &resources.Resource{
+			Name:    droplet.Name,
+			ID:      strconv.Itoa(droplet.ID),
+			Type:    resourceTypeDroplet,
+			Deleter: deleteDroplet,
+			Obj:     droplet,
+		}
+
+		resourceTrackers = append(resourceTrackers, resourceTracker)
+	}
+
+	return resourceTrackers, nil
+}
+
+func listVolumes(cloud fi.Cloud, clusterName string) ([]*resources.Resource, error) {
+	c := cloud.(*Cloud)
+	var resourceTrackers []*resources.Resource
+
+	volumeMatch := strings.Replace(clusterName, ".", "-", -1)
+
+	volumes, _, err := c.Volumes().ListVolumes(context.TODO(), &godo.ListVolumeParams{
+		Region: c.Region,
+	})
+
+	if err != nil {
+		return nil, fmt.Errorf("failed to list volumes: %s", err)
+	}
+
+	for _, volume := range volumes {
+		if strings.Contains(volume.Name, volumeMatch) {
+			resourceTracker := &resources.Resource{
+				Name:    volume.Name,
+				ID:      volume.ID,
+				Type:    resourceTypeVolume,
+				Deleter: deleteVolume,
+				Obj:     volume,
+			}
+
+			var blocks []string
+			for _, dropletID := range volume.DropletIDs {
+				blocks = append(blocks, "droplet:"+strconv.Itoa(dropletID))
+			}
+
+			resourceTracker.Blocks = blocks
+			resourceTrackers = append(resourceTrackers, resourceTracker)
+		}
+	}
+
+	return resourceTrackers, nil
 }
 
-// ListResources fetches all digitalocean resources into tracker.Resources
-func (r *Resources) ListResources() (map[string]*resources.Resource, error) {
-	return nil, nil
+func deleteDroplet(cloud fi.Cloud, t *resources.Resource) error {
+	c := cloud.(*Cloud)
+
+	dropletID, err := strconv.Atoi(t.ID)
+	if err != nil {
+		return fmt.Errorf("failed to convert droplet ID to int: %s", err)
+	}
+
+	_, err = c.Droplets().Delete(context.TODO(), dropletID)
+	if err != nil {
+		return fmt.Errorf("failed to delete droplet: %d, err: %s", dropletID, err)
+	}
+
+	return nil
 }
 
-// DeleteResources deletes all resources passed in the form in tracker.Resources
-func (r *Resources) DeleteResources(resources map[string]*resources.Resource) error {
+func deleteVolume(cloud fi.Cloud, t *resources.Resource) error {
+	c := cloud.(*Cloud)
+
+	volume := t.Obj.(godo.Volume)
+	for _, dropletID := range volume.DropletIDs {
+		action, _, err := c.VolumeActions().DetachByDropletID(context.TODO(), volume.ID, dropletID)
+		if err != nil {
+			return fmt.Errorf("failed to detach volume: %s, err: %s", volume.ID, err)
+		}
+		if err := waitForDetach(c, action); err != nil {
+			return fmt.Errorf("error while waiting for volume %s to detach: %s", volume.ID, err)
+		}
+	}
+
+	_, err := c.Volumes().DeleteVolume(context.TODO(), t.ID)
+	if err != nil {
+		return fmt.Errorf("failed to delete volume: %s, err: %s", t.ID, err)
+	}
+
 	return nil
 }
+
+func waitForDetach(cloud *Cloud, action *godo.Action) error {
+	timeout := time.After(10 * time.Second)
+	tick := time.Tick(500 * time.Millisecond)
+	for {
+		select {
+		case <-timeout:
+			return errors.New("timed out waiting for volume to detach")
+		case <-tick:
+			updatedAction, _, err := cloud.Client.Actions.Get(context.TODO(), action.ID)
+			if err != nil {
+				return err
+			}
+
+			if updatedAction.Status == godo.ActionCompleted {
+				return nil
+			}
+		}
+	}
+}

pkg/resources/ops/collector.go

@@ -36,7 +36,7 @@ func ListResources(cloud fi.Cloud, clusterName string, region string) (map[strin
 	case kops.CloudProviderAWS:
 		return aws.ListResourcesAWS(cloud.(awsup.AWSCloud), clusterName)
 	case kops.CloudProviderDO:
-		return digitalocean.ListResources(cloud, clusterName)
+		return digitalocean.ListResources(cloud.(*digitalocean.Cloud), clusterName)
 	case kops.CloudProviderGCE:
 		return gce.ListResourcesGCE(cloud.(cloudgce.GCECloud), clusterName, region)
 	case kops.CloudProviderVSphere: