Use AWS SDK to delete volumes during cluster down (#1954)

Author: deliahu

cli/cmd/cluster.go

@@ -25,6 +25,7 @@ import (
 	"time"
 
 	"github.com/aws/aws-sdk-go/service/autoscaling"
+	"github.com/aws/aws-sdk-go/service/ec2"
 	"github.com/aws/aws-sdk-go/service/elbv2"
 	"github.com/cortexlabs/cortex/cli/cluster"
 	"github.com/cortexlabs/cortex/cli/types/cliconfig"
@@ -421,6 +422,17 @@ var _clusterDownCmd = &cobra.Command{
 			case clusterstate.StatusNotFound:
 				exit.Error(clusterstate.ErrorClusterDoesNotExist(accessConfig.ClusterName, accessConfig.Region))
 			case clusterstate.StatusDeleteComplete:
+				// silently clean up
+				awsClient.DeleteQueuesWithPrefix(clusterconfig.SQSNamePrefix(accessConfig.ClusterName))
+				awsClient.DeletePolicy(clusterconfig.DefaultPolicyARN(accountID, accessConfig.ClusterName, accessConfig.Region))
+				if !_flagClusterDownKeepVolumes {
+					volumes, err := listPVCVolumesForCluster(awsClient, accessConfig.ClusterName)
+					if err == nil {
+						for _, volume := range volumes {
+							awsClient.DeleteVolume(*volume.VolumeId)
+						}
+					}
+				}
 				exit.Error(clusterstate.ErrorClusterAlreadyDeleted(accessConfig.ClusterName, accessConfig.Region))
 			}
 		}
@@ -437,7 +449,7 @@ var _clusterDownCmd = &cobra.Command{
 		fmt.Print("￮ deleting sqs queues ")
 		err = awsClient.DeleteQueuesWithPrefix(clusterconfig.SQSNamePrefix(accessConfig.ClusterName))
 		if err != nil {
-			fmt.Printf("\n\nfailed to delete all sqs queues; please delete queues starting with the name %s via the cloudwatch console: https://%s.console.aws.amazon.com/sqs/v2/home", clusterconfig.SQSNamePrefix(accessConfig.ClusterName), accessConfig.Region)
+			fmt.Printf("\n\nfailed to delete all sqs queues; please delete queues starting with the name %s via the cloudwatch console: https://%s.console.aws.amazon.com/sqs/v2/home\n", clusterconfig.SQSNamePrefix(accessConfig.ClusterName), accessConfig.Region)
 			errors.PrintError(err)
 			fmt.Println()
 		} else {
@@ -446,12 +458,7 @@ var _clusterDownCmd = &cobra.Command{
 
 		fmt.Print("￮ spinning down the cluster ...")
 
-		uninstallCmd := "/root/uninstall.sh"
-		if _flagClusterDownKeepVolumes {
-			uninstallCmd += " --keep-volumes"
-		}
-
-		out, exitCode, err := runManagerAccessCommand(uninstallCmd, *accessConfig, awsClient, nil, nil)
+		out, exitCode, err := runManagerAccessCommand("/root/uninstall.sh", *accessConfig, awsClient, nil, nil)
 		if err != nil {
 			errors.PrintError(err)
 			fmt.Println()
@@ -466,13 +473,41 @@ var _clusterDownCmd = &cobra.Command{
 		fmt.Printf("￮ deleting auto-generated iam policy %s ", policyARN)
 		err = awsClient.DeletePolicy(policyARN)
 		if err != nil {
-			fmt.Printf("\n\nfailed to delete auto-generated cortex policy %s; please delete the policy via the iam console: https://us-west-2.console.aws.amazon.com/iam/home#/policies", policyARN)
+			fmt.Printf("\n\nfailed to delete auto-generated cortex policy %s; please delete the policy via the iam console: https://console.aws.amazon.com/iam/home#/policies\n", policyARN)
 			errors.PrintError(err)
 			fmt.Println()
 		} else {
 			fmt.Println("✓")
 		}
 
+		// delete EBS volumes
+		if !_flagClusterDownKeepVolumes {
+			volumes, err := listPVCVolumesForCluster(awsClient, accessConfig.ClusterName)
+			if err != nil {
+				fmt.Println("\nfailed to list volumes for deletion; please delete any volumes associated with your cluster via the ec2 console: https://console.aws.amazon.com/ec2/v2/home?#Volumes")
+				errors.PrintError(err)
+				fmt.Println()
+			} else {
+				fmt.Print("￮ deleting ebs volumes ")
+				var failedToDeleteVolumes []string
+				var lastErr error
+				for _, volume := range volumes {
+					err := awsClient.DeleteVolume(*volume.VolumeId)
+					if err != nil {
+						failedToDeleteVolumes = append(failedToDeleteVolumes, *volume.VolumeId)
+						lastErr = err
+					}
+				}
+				if lastErr != nil {
+					fmt.Printf("\n\nfailed to delete %s %s; please delete %s via the ec2 console: https://console.aws.amazon.com/ec2/v2/home?#Volumes\n", s.PluralS("volume", len(failedToDeleteVolumes)), s.UserStrsAnd(failedToDeleteVolumes), s.PluralCustom("it", "them", len(failedToDeleteVolumes)))
+					errors.PrintError(lastErr)
+					fmt.Println()
+				} else {
+					fmt.Println("✓")
+				}
+			}
+		}
+
 		// best-effort deletion of cli environment(s)
 		if loadBalancer != nil {
 			envNames, isDefaultEnv, _ := getEnvNamesByOperatorEndpoint(*loadBalancer.DNSName)
@@ -1018,3 +1053,10 @@ func getAWSOperatorLoadBalancer(clusterName string, awsClient *aws.Client) (*elb
 
 	return loadBalancer, nil
 }
+
+func listPVCVolumesForCluster(awsClient *aws.Client, clusterName string) ([]ec2.Volume, error) {
+	return awsClient.ListVolumes(ec2.Tag{
+		Key:   pointer.String(fmt.Sprintf("kubernetes.io/cluster/%s", clusterName)),
+		Value: nil, // any value should be ok as long as the key is present
+	})
+}

cli/cmd/cluster_gcp.go

@@ -144,9 +144,9 @@ var _clusterGCPUpCmd = &cobra.Command{
 			exit.Error(err)
 		}
 
-		gkeClusterName := fmt.Sprintf("projects/%s/locations/%s/clusters/%s", clusterConfig.Project, clusterConfig.Zone, clusterConfig.ClusterName)
+		fullyQualifiedClusterName := fmt.Sprintf("projects/%s/locations/%s/clusters/%s", clusterConfig.Project, clusterConfig.Zone, clusterConfig.ClusterName)
 
-		clusterExists, err := gcpClient.ClusterExists(gkeClusterName)
+		clusterExists, err := gcpClient.ClusterExists(fullyQualifiedClusterName)
 		if err != nil {
 			exit.Error(err)
 		}
@@ -169,7 +169,7 @@ var _clusterGCPUpCmd = &cobra.Command{
 			exit.Error(err)
 		}
 
-		operatorLoadBalancerIP, err := getGCPOperatorLoadBalancerIP(gkeClusterName, gcpClient)
+		operatorLoadBalancerIP, err := getGCPOperatorLoadBalancerIP(fullyQualifiedClusterName, gcpClient)
 		if err != nil {
 			exit.Error(errors.Append(err, fmt.Sprintf("\n\nyou can attempt to resolve this issue and configure your cli environment by running `cortex cluster info --configure-env %s`", _flagClusterGCPUpEnv)))
 		}
@@ -244,10 +244,10 @@ var _clusterGCPDownCmd = &cobra.Command{
 			exit.Error(err)
 		}
 
-		gkeClusterName := fmt.Sprintf("projects/%s/locations/%s/clusters/%s", accessConfig.Project, accessConfig.Zone, accessConfig.ClusterName)
+		fullyQualifiedClusterName := fmt.Sprintf("projects/%s/locations/%s/clusters/%s", accessConfig.Project, accessConfig.Zone, accessConfig.ClusterName)
 		bucketName := clusterconfig.GCPBucketName(accessConfig.ClusterName, accessConfig.Project, accessConfig.Zone)
 
-		clusterExists, err := gcpClient.ClusterExists(gkeClusterName)
+		clusterExists, err := gcpClient.ClusterExists(fullyQualifiedClusterName)
 		if err != nil {
 			exit.Error(err)
 		}
@@ -257,7 +257,7 @@ var _clusterGCPDownCmd = &cobra.Command{
 		}
 
 		// updating CLI env is best-effort, so ignore errors
-		operatorLoadBalancerIP, _ := getGCPOperatorLoadBalancerIP(gkeClusterName, gcpClient)
+		operatorLoadBalancerIP, _ := getGCPOperatorLoadBalancerIP(fullyQualifiedClusterName, gcpClient)
 
 		if _flagClusterGCPDisallowPrompt {
 			fmt.Printf("your cluster named \"%s\" in %s (zone: %s) will be spun down and all apis will be deleted\n\n", accessConfig.ClusterName, accessConfig.Project, accessConfig.Zone)
@@ -291,7 +291,7 @@ var _clusterGCPDownCmd = &cobra.Command{
 			fmt.Print("￮ proceeding with best-effort deletion of the cluster ")
 		}
 
-		_, err = gcpClient.DeleteCluster(gkeClusterName)
+		_, err = gcpClient.DeleteCluster(fullyQualifiedClusterName)
 		if err != nil {
 			fmt.Print("\n\n")
 			exit.Error(err)
@@ -461,7 +461,7 @@ func createGKECluster(clusterConfig *clusterconfig.GCPConfig, gcpClient *gcp.Cli
 	fmt.Print("￮ creating GKE cluster ")
 
 	gkeClusterParent := fmt.Sprintf("projects/%s/locations/%s", clusterConfig.Project, clusterConfig.Zone)
-	gkeClusterName := fmt.Sprintf("%s/clusters/%s", gkeClusterParent, clusterConfig.ClusterName)
+	fullyQualifiedClusterName := fmt.Sprintf("%s/clusters/%s", gkeClusterParent, clusterConfig.ClusterName)
 
 	gkeClusterConfig := containerpb.Cluster{
 		Name:                  clusterConfig.ClusterName,
@@ -571,7 +571,7 @@ func createGKECluster(clusterConfig *clusterconfig.GCPConfig, gcpClient *gcp.Cli
 		fmt.Print(".")
 		time.Sleep(5 * time.Second)
 
-		cluster, err := gcpClient.GetCluster(gkeClusterName)
+		cluster, err := gcpClient.GetCluster(fullyQualifiedClusterName)
 		if err != nil {
 			return err
 		}
@@ -593,8 +593,8 @@ func createGKECluster(clusterConfig *clusterconfig.GCPConfig, gcpClient *gcp.Cli
 	return nil
 }
 
-func getGCPOperatorLoadBalancerIP(clusterName string, gcpClient *gcp.Client) (string, error) {
-	cluster, err := gcpClient.GetCluster(clusterName)
+func getGCPOperatorLoadBalancerIP(fullyQualifiedClusterName string, gcpClient *gcp.Client) (string, error) {
+	cluster, err := gcpClient.GetCluster(fullyQualifiedClusterName)
 	if err != nil {
 		return "", err
 	}

manager/uninstall.sh

@@ -40,14 +40,7 @@ function uninstall_gcp() {
 
 function uninstall_aws() {
   echo
-
   aws eks --region $CORTEX_REGION update-kubeconfig --name $CORTEX_CLUSTER_NAME >/dev/null
-
-  if [ "$arg1" != "--keep-volumes" ]; then
-    uninstall_prometheus
-    uninstall_grafana
-  fi
-
   eksctl delete cluster --wait --name=$CORTEX_CLUSTER_NAME --region=$CORTEX_REGION --timeout=$EKSCTL_TIMEOUT
   echo -e "\n✓ done spinning down the cluster"
 }

pkg/lib/aws/ec2.go

@@ -329,3 +329,63 @@ func (c *Client) DescribeVpcs() ([]ec2.Vpc, error) {
 
 	return vpcs, nil
 }
+
+func (c *Client) ListVolumes(tags ...ec2.Tag) ([]ec2.Volume, error) {
+	var volumes []ec2.Volume
+	err := c.EC2().DescribeVolumesPages(&ec2.DescribeVolumesInput{}, func(output *ec2.DescribeVolumesOutput, lastPage bool) bool {
+		if output == nil {
+			return false
+		}
+		for _, volume := range output.Volumes {
+			if volume == nil {
+				continue
+			}
+			if hasAllEC2Tags(tags, volume.Tags) {
+				volumes = append(volumes, *volume)
+			}
+		}
+
+		return true
+	})
+
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+
+	return volumes, nil
+}
+
+func (c *Client) DeleteVolume(volumeID string) error {
+	_, err := c.EC2().DeleteVolume(&ec2.DeleteVolumeInput{
+		VolumeId: aws.String(volumeID),
+	})
+	if err != nil {
+		return errors.Wrap(err)
+	}
+
+	return nil
+}
+
+func hasAllEC2Tags(queryTags []ec2.Tag, allResourceTags []*ec2.Tag) bool {
+	for _, queryTag := range queryTags {
+		if !hasEC2Tag(queryTag, allResourceTags) {
+			return false
+		}
+	}
+	return true
+}
+
+// if queryTag's value is nil, the tag will match as long as the key is present in the resource's tags
+func hasEC2Tag(queryTag ec2.Tag, allResourceTags []*ec2.Tag) bool {
+	for _, resourceTag := range allResourceTags {
+		if queryTag.Key != nil && resourceTag.Key != nil && *queryTag.Key == *resourceTag.Key {
+			if queryTag.Value == nil {
+				return true
+			}
+			if queryTag.Value != nil && resourceTag.Value != nil && *queryTag.Value == *resourceTag.Value {
+				return true
+			}
+		}
+	}
+	return false
+}

pkg/lib/gcp/clients.go

@@ -30,6 +30,7 @@ type clients struct {
 	gcs     *storage.Client
 	compute *compute.Service
 	gke     *container.ClusterManagerClient
+	disks   *compute.DisksService
 }
 
 func (c *Client) GCS() (*storage.Client, error) {
@@ -82,3 +83,18 @@ func (c *Client) GKE() (*container.ClusterManagerClient, error) {
 	}
 	return c.clients.gke, nil
 }
+
+func (c *Client) Disks() (*compute.DisksService, error) {
+	if c.clients.disks == nil {
+		comp, err := c.Compute()
+		if err != nil {
+			return nil, err
+		}
+		disks := compute.NewDisksService(comp)
+		if err != nil {
+			return nil, errors.WithStack(err)
+		}
+		c.clients.disks = disks
+	}
+	return c.clients.disks, nil
+}

pkg/lib/gcp/compute.go

@@ -330,3 +330,28 @@ func (c *Client) GetAvailableZonesForAccelerator(acceleratorType string) ([]stri
 
 	return availableAcceleratorZones.SliceSorted(), nil
 }
+
+func (c *Client) ListDisks(zone string) ([]compute.Disk, error) {
+	disksClient, err := c.Disks()
+	if err != nil {
+		return nil, err
+	}
+
+	var disks []compute.Disk
+
+	err = disksClient.List(c.ProjectID, zone).Pages(context.Background(), func(list *compute.DiskList) error {
+		for _, disk := range list.Items {
+			if disk == nil {
+				continue
+			}
+			disks = append(disks, *disk)
+		}
+		return nil
+	})
+
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+
+	return disks, nil
+}

pkg/lib/gcp/gke.go

@@ -29,13 +29,13 @@ import (
 	grpcStatus "google.golang.org/grpc/status"
 )
 
-func (c *Client) GetCluster(clusterName string) (*containerpb.Cluster, error) {
+func (c *Client) GetCluster(fullyQualifiedClusterName string) (*containerpb.Cluster, error) {
 	gke, err := c.GKE()
 	if err != nil {
 		return nil, err
 	}
 	cluster, err := gke.GetCluster(context.Background(), &containerpb.GetClusterRequest{
-		Name: clusterName,
+		Name: fullyQualifiedClusterName,
 	})
 	if err != nil {
 		return nil, errors.WithStack(err)

pkg/types/clusterconfig/cluster_config_aws.go

@@ -214,7 +214,7 @@ var CoreConfigStructFieldValidations = []*cr.StructFieldValidation{
 		StructField: "ClusterName",
 		StringValidation: &cr.StringValidation{
 			Default:   "cortex",
-			MaxLength: 63,
+			MaxLength: 54, // leaves room for 8 char uniqueness string (and "-") for bucket name (63 chars max)
 			MinLength: 3,
 			Validator: validateClusterName,
 		},
@@ -704,7 +704,7 @@ var AccessValidation = &cr.StructValidation{
 			StructField: "ClusterName",
 			StringValidation: &cr.StringValidation{
 				Default:   "cortex",
-				MaxLength: 63,
+				MaxLength: 54, // leaves room for 8 char uniqueness string (and "-") for bucket name (63 chars max)
 				MinLength: 3,
 				Validator: validateClusterName,
 			},
@@ -805,17 +805,8 @@ func (cc *Config) Validate(awsClient *aws.Client, skipQuotaVerification bool) er
 	}
 
 	if cc.Bucket == "" {
-		bucketID := hash.String(accountID + cc.Region)[:10]
-
-		defaultBucket := cc.ClusterName + "-" + bucketID
-		if len(defaultBucket) > 63 {
-			defaultBucket = defaultBucket[:63]
-		}
-		if strings.HasSuffix(defaultBucket, "-") {
-			defaultBucket = defaultBucket[:len(defaultBucket)-1]
-		}
-
-		cc.Bucket = defaultBucket
+		bucketID := hash.String(accountID + cc.Region)[:8] // this is to "guarantee" a globally unique name
+		cc.Bucket = cc.ClusterName + "-" + bucketID
 	} else {
 		bucketRegion, _ := aws.GetBucketRegion(cc.Bucket)
 		if bucketRegion != "" && bucketRegion != cc.Region { // if the bucket didn't exist, we will create it in the correct region, so there is no error