CB-27363 Implement AWS TerminateInstances

hortonworks · Sep 30, 2024 · 6b8bef2 · 6b8bef2
1 parent 5b550db
commit 6b8bef2
Show file tree

Hide file tree

Showing 3 changed files with 94 additions and 36 deletions.
diff --git a/aws/aws.go b/aws/aws.go
@@ -266,8 +266,62 @@ func (p awsProvider) getCloudWatchClientsByRegion() map[string]cloudWatchClient
 	return cwClients
 }
 
-func (p awsProvider) TerminateInstances(*types.InstanceContainer) []error {
-	return []error{errors.New("[AWS] Termination is not supported")}
+func (p awsProvider) TerminateInstances(instances *types.InstanceContainer) []error {
+	log.Debug("[AWS] Terminating instances")
+	regionInstances := map[string][]string{}
+	for _, instance := range instances.Get(p.GetCloudType()) {
+		regionInstances[instance.Region] = append(regionInstances[instance.Region], instance.ID)
+	}
+	log.Debugf("[AWS] Terminating instances: %v", regionInstances)
+
+	wg := sync.WaitGroup{}
+	wg.Add(len(regionInstances))
+	errChan := make(chan error)
+
+	for r, i := range regionInstances {
+		go func(region string, instanceIds []string) {
+			defer wg.Done()
+
+			ec2Client := p.ec2Clients[region]
+
+			for i := 0; i < len(instanceIds); i += ctx.AwsBulkOperationSize {
+				log.Infof("[AWS] Round %d for terminate operation in region %s", (i/ctx.AwsBulkOperationSize)+1, region)
+				arrayEnd := i + ctx.AwsBulkOperationSize
+				if ctx.AwsBulkOperationSize > len(instanceIds[i:]) {
+					arrayEnd = i + len(instanceIds[i:])
+				}
+
+				instanceIdsChunk := instanceIds[i:arrayEnd]
+
+				_, err := ec2Client.TerminateInstances(&ec2.TerminateInstancesInput{
+					InstanceIds: aws.StringSlice(instanceIdsChunk),
+				})
+				if err != nil {
+					log.Errorf("[AWS] Failed to terminate instances: %v, err: %s", instanceIdsChunk, err)
+					errChan <- err
+				}
+
+				err = ec2Client.WaitUntilInstanceTerminated(&ec2.DescribeInstancesInput{
+					InstanceIds: aws.StringSlice(instanceIdsChunk),
+				})
+				if err != nil {
+					log.Errorf("[AWS] Failed to wait for terminated instances: %v, err: %s", instanceIdsChunk, err)
+					errChan <- err
+				}
+			}
+		}(r, i)
+	}
+
+	go func() {
+		wg.Wait()
+		close(errChan)
+	}()
+
+	var errs []error
+	for err := range errChan {
+		errs = append(errs, err)
+	}
+	return errs
 }
 
 func (p awsProvider) TerminateStacks(stacks *types.StackContainer) []error {
@@ -1971,17 +2025,18 @@ func newStack(cloudType types.CloudType, stack *cloudformation.Stack, region str
 
 // The low byte represents the state. The high byte is an opaque internal value
 // and should be ignored.
-//    * 0 : pending
 //
-//    * 16 : running
+//   - 0 : pending
+//
+//   - 16 : running
 //
-//    * 32 : shutting-down
+//   - 32 : shutting-down
 //
-//    * 48 : terminated
+//   - 48 : terminated
 //
-//    * 64 : stopping
+//   - 64 : stopping
 //
-//    * 80 : stopped
+//   - 80 : stopped
 func getInstanceState(instance *ec2.Instance) types.State {
 	if instance.State == nil {
 		return types.Unknown

diff --git a/azure/azure.go b/azure/azure.go
@@ -692,13 +692,15 @@ func getScaleSetInstanceState(view compute.VirtualMachineScaleSetVMInstanceView)
 }
 
 // Possible values:
-//   "PowerState/deallocated"
-//   "PowerState/deallocating"
-//   "PowerState/running"
-//   "PowerState/starting"
-//   "PowerState/stopped"
-//   "PowerState/stopping"
-//   "PowerState/unknown"
+//
+//	"PowerState/deallocated"
+//	"PowerState/deallocating"
+//	"PowerState/running"
+//	"PowerState/starting"
+//	"PowerState/stopped"
+//	"PowerState/stopping"
+//	"PowerState/unknown"
+//
 // The assumption is that the status without time is the currently active status
 func convertViewStatusToState(actualStatus compute.InstanceViewStatus) types.State {
 	switch *actualStatus.Code {

diff --git a/gcp/gcp.go b/gcp/gcp.go
@@ -971,14 +971,15 @@ func newInstance(inst *compute.Instance) *types.Instance {
 // }
 
 // Possible values:
-//   "PROVISIONING"
-//   "RUNNING"
-//   "STAGING"
-//   "STOPPED"
-//   "STOPPING"
-//   "SUSPENDED"
-//   "SUSPENDING"
-//   "TERMINATED"
+//
+//	"PROVISIONING"
+//	"RUNNING"
+//	"STAGING"
+//	"STOPPED"
+//	"STOPPING"
+//	"SUSPENDED"
+//	"SUSPENDING"
+//	"TERMINATED"
 func getInstanceState(instance *compute.Instance) types.State {
 	switch instance.Status {
 	case "PROVISIONING", "RUNNING", "STAGING":
@@ -1126,12 +1127,12 @@ func getDatabaseInstanceCreationTimeStamp(opService *sqladmin.OperationsListCall
 	return time.Time{}, errors.New(fmt.Sprintf("[GCP] Failed to get the CREATE operation of the DB instance: %s", dbName))
 }
 
-//Possible values:
-//CREATING: Disk is provisioning.
-//RESTORING: Source data is being copied into the disk.
-//FAILED: Disk creation failed.
-//READY: Disk is ready for use.
-//DELETING: Disk is deleting.
+// Possible values:
+// CREATING: Disk is provisioning.
+// RESTORING: Source data is being copied into the disk.
+// FAILED: Disk creation failed.
+// READY: Disk is ready for use.
+// DELETING: Disk is deleting.
 func getDiskStatus(gDisk *compute.Disk) types.State {
 	switch gDisk.Status {
 	case "CREATING", "RESTORING", "STAGING", "READY", "FAILED":
@@ -1147,13 +1148,13 @@ func getDiskStatus(gDisk *compute.Disk) types.State {
 	}
 }
 
-//SQL_INSTANCE_STATE_UNSPECIFIED	The state of the instance is unknown.
-//RUNNABLE	The instance is running.
-//SUSPENDED	The instance is currently offline, but it may run again in the future.
-//PENDING_DELETE	The instance is being deleted.
-//PENDING_CREATE	The instance is being created.
-//MAINTENANCE	The instance is down for maintenance.
-//FAILED	The instance failed to be created.
+// SQL_INSTANCE_STATE_UNSPECIFIED	The state of the instance is unknown.
+// RUNNABLE	The instance is running.
+// SUSPENDED	The instance is currently offline, but it may run again in the future.
+// PENDING_DELETE	The instance is being deleted.
+// PENDING_CREATE	The instance is being created.
+// MAINTENANCE	The instance is down for maintenance.
+// FAILED	The instance failed to be created.
 func getDatabaseInstanceStatus(instance *sqladmin.DatabaseInstance) types.State {
 	switch instance.State {
 	case "RUNNABLE":