[WIP][OCPCLOUD-1381] Implement AWS Placement Group support and creation

go.mod

@@ -20,6 +20,8 @@ require (
 	sigs.k8s.io/controller-tools v0.7.0
 )
 
+replace github.com/openshift/api => github.com/JoelSpeed/api v0.0.0-20220204113201-6d46be2a3ee5
+
 require (
 	cloud.google.com/go v0.81.0 // indirect
 	github.com/Azure/go-ansiterm v0.0.0-20210617225240-d185dfc1b5a1 // indirect

pkg/actuators/machine/instances.go

@@ -5,7 +5,6 @@ import (
 	"errors"
 	"fmt"
 	"sort"
-	"strconv"
 	"strings"
 	"time"
 
@@ -353,30 +352,13 @@ func launchInstance(machine *machinev1.Machine, machineProviderConfig *machinev1
 		}
 	}
 
-	var placement *ec2.Placement
-	if machineProviderConfig.Placement.AvailabilityZone != "" && machineProviderConfig.Subnet.ID == nil {
-		placement = &ec2.Placement{
-			AvailabilityZone: aws.String(machineProviderConfig.Placement.AvailabilityZone),
-		}
+	if err := checkOrCreatePlacementGroup(client, machineProviderConfig.Placement, clusterID); err != nil {
+		return nil, err
 	}
 
-	instanceTenancy := machineProviderConfig.Placement.Tenancy
-
-	switch instanceTenancy {
-	case "":
-		// Do nothing when not set
-	case machinev1.DefaultTenancy, machinev1.DedicatedTenancy, machinev1.HostTenancy:
-		if placement == nil {
-			placement = &ec2.Placement{}
-		}
-		tenancy := string(instanceTenancy)
-		placement.Tenancy = &tenancy
-	default:
-		return nil, mapierrors.CreateMachine("invalid instance tenancy: %s. Allowed options are: %s,%s,%s",
-			instanceTenancy,
-			machinev1.DefaultTenancy,
-			machinev1.DedicatedTenancy,
-			machinev1.HostTenancy)
+	placement, err := constructInstancePlacement(machineProviderConfig)
+	if err != nil {
+		return nil, err
 	}
 
 	inputConfig := ec2.RunInstancesInput{
@@ -410,7 +392,7 @@ func launchInstance(machine *machinev1.Machine, machineProviderConfig *machinev1
 		// https://docs.aws.amazon.com/sdk-for-go/api/aws/awserr/
 		if _, ok := err.(awserr.Error); ok {
 			if reqErr, ok := err.(awserr.RequestFailure); ok {
-				if strings.HasPrefix(strconv.Itoa(reqErr.StatusCode()), "4") {
+				if reqErr.StatusCode() >= 400 && reqErr.StatusCode() < 500 {
 					klog.Infof("Error launching instance: %v", reqErr)
 					return nil, mapierrors.InvalidMachineConfiguration("error launching instance: %v", reqErr.Message())
 				}
@@ -525,3 +507,142 @@ func getInstanceMarketOptionsRequest(providerConfig *machinev1.AWSMachineProvide
 
 	return instanceMarketOptionsRequest
 }
+
+// constructInstancePlacement configures the placement options for the RunInstances request
+func constructInstancePlacement(machineProviderConfig *machinev1.AWSMachineProviderConfig) (*ec2.Placement, error) {
+	placement := &ec2.Placement{}
+	if machineProviderConfig.Placement.AvailabilityZone != "" && machineProviderConfig.Subnet.ID == nil {
+		placement.SetAvailabilityZone(machineProviderConfig.Placement.AvailabilityZone)
+	}
+
+	instanceTenancy := machineProviderConfig.Placement.Tenancy
+	switch instanceTenancy {
+	case "":
+		// Do nothing when not set
+	case machinev1.DefaultTenancy, machinev1.DedicatedTenancy, machinev1.HostTenancy:
+		placement.SetTenancy(string(instanceTenancy))
+	default:
+		return nil, mapierrors.CreateMachine("invalid instance tenancy: %s. Allowed options are: %s,%s,%s",
+			instanceTenancy,
+			machinev1.DefaultTenancy,
+			machinev1.DedicatedTenancy,
+			machinev1.HostTenancy)
+	}
+
+	if machineProviderConfig.Placement.GroupName != "" {
+		placement.SetGroupName(machineProviderConfig.Placement.GroupName)
+	}
+	if machineProviderConfig.Placement.Partition != nil && machineProviderConfig.Placement.Partition.Number != 0 {
+		placement.SetPartitionNumber(int64(machineProviderConfig.Placement.Partition.Number))
+	}
+
+	if *placement == (ec2.Placement{}) {
+		// If the placement is empty, we should just return a nil so as not to pollute the RunInstancesInput
+		return nil, nil
+	}
+
+	return placement, nil
+}
+
+func checkOrCreatePlacementGroup(client awsclient.Client, placement machinev1.Placement, clusterID string) error {
+	if placement.GroupName == "" {
+		// Nothing to do if the group name is empty
+		return nil
+	}
+
+	placementGroups, err := client.DescribePlacementGroups(&ec2.DescribePlacementGroupsInput{
+		GroupNames: []*string{aws.String(placement.GroupName)},
+	})
+	if err != nil && !isAWS4xxError(err) {
+		// Ignore a 400 error as AWS will report an unknown placement group as a 400.
+		return fmt.Errorf("could not describe placement groups: %v", err)
+	}
+
+	if len(placementGroups.PlacementGroups) == 1 {
+		return validatePlacementGroupConfig(placement, placementGroups.PlacementGroups[0])
+	}
+	if len(placementGroups.PlacementGroups) > 1 {
+		return fmt.Errorf("expected 1 placement group for name %q, got %d", placement.GroupName, len(placementGroups.PlacementGroups))
+	}
+
+	// No placement group by that name existed, so we create one
+	createPlacementGroupInput := &ec2.CreatePlacementGroupInput{
+		GroupName: aws.String(placement.GroupName),
+		TagSpecifications: []*ec2.TagSpecification{
+			{
+				ResourceType: aws.String(ec2.ResourceTypePlacementGroup),
+				Tags: []*ec2.Tag{
+					{Key: aws.String("kubernetes.io/cluster/" + clusterID), Value: aws.String("owned")},
+					{Key: aws.String("Name"), Value: aws.String(placement.GroupName)},
+				},
+			},
+		},
+	}
+	switch placement.GroupType {
+	case machinev1.AWSSpreadPlacementGroupType:
+		createPlacementGroupInput.SetStrategy(ec2.PlacementStrategySpread)
+	case machinev1.AWSClusterPlacementGroupType:
+		createPlacementGroupInput.SetStrategy(ec2.PlacementStrategyCluster)
+	case machinev1.AWSPartitionPlacementGroupType:
+		createPlacementGroupInput.SetStrategy(ec2.PlacementStrategyPartition)
+		if placement.Partition != nil && placement.Partition.Count != 0 {
+			createPlacementGroupInput.SetPartitionCount(int64(placement.Partition.Count))
+		}
+	default:
+		return fmt.Errorf("unknown placement strategy %q: valid values are %s, %s, %s", placement.GroupType, machinev1.AWSSpreadPlacementGroupType, machinev1.AWSClusterPlacementGroupType, machinev1.AWSPartitionPlacementGroupType)
+	}
+
+	if _, err := client.CreatePlacementGroup(createPlacementGroupInput); err != nil {
+		return fmt.Errorf("unable to create placement group: %v", err)
+	}
+
+	return nil
+}
+
+// validatePlacementGroupConfig validates that the configuration of the existing placement group
+// matches the configuration from the Machine
+func validatePlacementGroupConfig(placement machinev1.Placement, placementGroup *ec2.PlacementGroup) error {
+	if placementGroup == nil {
+		return fmt.Errorf("found nil placement group")
+	}
+	if aws.StringValue(placementGroup.GroupName) != placement.GroupName {
+		return fmt.Errorf("placement group name mismatch: wanted: %q, got: %q", placement.GroupName, aws.StringValue(placementGroup.GroupName))
+	}
+	var expectedPlacementGroupType string
+	switch placement.GroupType {
+	case machinev1.AWSSpreadPlacementGroupType:
+		expectedPlacementGroupType = ec2.PlacementStrategySpread
+	case machinev1.AWSClusterPlacementGroupType:
+		expectedPlacementGroupType = ec2.PlacementStrategyCluster
+	case machinev1.AWSPartitionPlacementGroupType:
+		expectedPlacementGroupType = ec2.PlacementStrategyPartition
+	default:
+		return fmt.Errorf("unknown placement strategy %q: valid values are %s, %s and %s", placement.GroupType, machinev1.AWSSpreadPlacementGroupType, machinev1.AWSClusterPlacementGroupType, machinev1.AWSPartitionPlacementGroupType)
+	}
+
+	if aws.StringValue(placementGroup.Strategy) != expectedPlacementGroupType {
+		return mapierrors.InvalidMachineConfiguration("mismatch between configured placement group type and existing placement group type: wanted: %q, got: %q", expectedPlacementGroupType, aws.StringValue(placementGroup.Strategy))
+	}
+
+	if placement.GroupType == machinev1.AWSPartitionPlacementGroupType && placement.Partition != nil {
+		if placement.Partition.Count != 0 && int64(placement.Partition.Count) != aws.Int64Value(placementGroup.PartitionCount) {
+			return mapierrors.InvalidMachineConfiguration("mismatch between configured placement group partition count and existing placement group partition count: wanted: %d, got: %d", placement.Partition.Count, aws.Int64Value(placementGroup.PartitionCount))
+		}
+		if int64(placement.Partition.Number) > aws.Int64Value(placementGroup.PartitionCount) || placement.Partition.Number < 1 {
+			return mapierrors.InvalidMachineConfiguration("placement group has %d partitions, requested partition number (%d) does not exist", aws.Int64Value(placementGroup.PartitionCount), placement.Partition.Number)
+		}
+	}
+	return nil
+}
+
+// isAWS4xxError will determine if the passed error is an AWS error with a 4xx status code.
+func isAWS4xxError(err error) bool {
+	if _, ok := err.(awserr.Error); ok {
+		if reqErr, ok := err.(awserr.RequestFailure); ok {
+			if reqErr.StatusCode() >= 400 && reqErr.StatusCode() < 500 {
+				return true
+			}
+		}
+	}
+	return false
+}