Merge pull request kubernetes#59930 from zetaab/shutdoaw

Automatic merge from submit-queue (batch tested with PRs 67368, 59930, 68074). If you want to cherry-pick this change to another branch, please follow the instructions here: https://github.com/kubernetes/community/blob/master/contributors/devel/cherry-picks.md.

implement InstanceShutdownByProviderID to aws cloudprovider

**What this PR does / why we need it**: implement InstanceShutdownByProviderID to aws cloudprovider

**Which issue(s) this PR fixes**:
Fixes kubernetes#59925

**Special notes for your reviewer**:

**Release note**:
```release-note
NONE
```

Author: Kubernetes Submit Queue

pkg/cloudprovider/providers/aws/aws.go

@@ -1336,7 +1336,7 @@ func (c *Cloud) NodeAddressesByProviderID(ctx context.Context, providerID string
 	return extractNodeAddresses(instance)
 }
 
-// InstanceExistsByProviderID returns true if the instance with the given provider id still exists and is running.
+// InstanceExistsByProviderID returns true if the instance with the given provider id still exists.
 // If false is returned with no error, the instance will be immediately deleted by the cloud controller manager.
 func (c *Cloud) InstanceExistsByProviderID(ctx context.Context, providerID string) (bool, error) {
 	instanceID, err := kubernetesInstanceID(providerID).mapToAWSInstanceID()
@@ -1359,18 +1359,41 @@ func (c *Cloud) InstanceExistsByProviderID(ctx context.Context, providerID strin
 		return false, fmt.Errorf("multiple instances found for instance: %s", instanceID)
 	}
 
-	state := instances[0].State.Name
-	if *state != "running" {
-		glog.Warningf("the instance %s is not running", instanceID)
-		return false, nil
-	}
-
 	return true, nil
 }
 
 // InstanceShutdownByProviderID returns true if the instance is in safe state to detach volumes
 func (c *Cloud) InstanceShutdownByProviderID(ctx context.Context, providerID string) (bool, error) {
-	return false, cloudprovider.NotImplemented
+	instanceID, err := kubernetesInstanceID(providerID).mapToAWSInstanceID()
+	if err != nil {
+		return false, err
+	}
+
+	request := &ec2.DescribeInstancesInput{
+		InstanceIds: []*string{instanceID.awsString()},
+	}
+
+	instances, err := c.ec2.DescribeInstances(request)
+	if err != nil {
+		return false, err
+	}
+	if len(instances) == 0 {
+		glog.Warningf("the instance %s does not exist anymore", providerID)
+		return true, nil
+	}
+	if len(instances) > 1 {
+		return false, fmt.Errorf("multiple instances found for instance: %s", instanceID)
+	}
+
+	instance := instances[0]
+	if instance.State != nil {
+		state := aws.StringValue(instance.State.Name)
+		// valid state for detaching volumes
+		if state == ec2.InstanceStateNameStopped || state == ec2.InstanceStateNameTerminated {
+			return true, nil
+		}
+	}
+	return false, nil
 }
 
 // InstanceID returns the cloud provider ID of the node with the specified nodeName.