refactor CloudProvider.Create() to return an error

aws · Dec 20, 2021 · 75492a4 · 75492a4
1 parent 4b4ce94
commit 75492a4
Show file tree

Hide file tree

Showing 5 changed files with 31 additions and 49 deletions.
diff --git a/pkg/cloudprovider/aws/cloudprovider.go b/pkg/cloudprovider/aws/cloudprovider.go
@@ -108,13 +108,7 @@ func withUserAgent(sess *session.Session) *session.Session {
 }
 
 // Create a node given the constraints.
-func (c *CloudProvider) Create(ctx context.Context, constraints *v1alpha5.Constraints, instanceTypes []cloudprovider.InstanceType, quantity int, callback func(*v1.Node) error) <-chan error {
-	return c.creationQueue.Add(func() error {
-		return c.create(ctx, constraints, instanceTypes, quantity, callback)
-	})
-}
-
-func (c *CloudProvider) create(ctx context.Context, constraints *v1alpha5.Constraints, instanceTypes []cloudprovider.InstanceType, quantity int, callback func(*v1.Node) error) error {
+func (c *CloudProvider) Create(ctx context.Context, constraints *v1alpha5.Constraints, instanceTypes []cloudprovider.InstanceType, quantity int, callback func(*v1.Node) error) error {
 	vendorConstraints, err := v1alpha1.Deserialize(constraints)
 	if err != nil {
 		return err

diff --git a/pkg/cloudprovider/fake/cloudprovider.go b/pkg/cloudprovider/fake/cloudprovider.go
@@ -22,6 +22,7 @@ import (
 	"github.com/Pallinder/go-randomdata"
 	"github.com/aws/karpenter/pkg/apis/provisioning/v1alpha5"
 	"github.com/aws/karpenter/pkg/cloudprovider"
+	"go.uber.org/multierr"
 	"knative.dev/pkg/apis"
 
 	v1 "k8s.io/api/core/v1"
@@ -33,8 +34,8 @@ type CloudProvider struct {
 	InstanceTypes []cloudprovider.InstanceType
 }
 
-func (c *CloudProvider) Create(_ context.Context, constraints *v1alpha5.Constraints, instanceTypes []cloudprovider.InstanceType, quantity int, bind func(*v1.Node) error) <-chan error {
-	err := make(chan error)
+func (c *CloudProvider) Create(_ context.Context, constraints *v1alpha5.Constraints, instanceTypes []cloudprovider.InstanceType, quantity int, bind func(*v1.Node) error) error {
+	var err error
 	for i := 0; i < quantity; i++ {
 		name := strings.ToLower(randomdata.SillyName())
 		instance := instanceTypes[0]
@@ -49,32 +50,30 @@ func (c *CloudProvider) Create(_ context.Context, constraints *v1alpha5.Constrai
 			}
 		}
 
-		go func() {
-			err <- bind(&v1.Node{
-				ObjectMeta: metav1.ObjectMeta{
-					Name: name,
-					Labels: map[string]string{
-						v1.LabelTopologyZone:       zone,
-						v1.LabelInstanceTypeStable: instance.Name(),
-						v1alpha5.LabelCapacityType: capacityType,
-					},
+		err = multierr.Append(err, bind(&v1.Node{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: name,
+				Labels: map[string]string{
+					v1.LabelTopologyZone:       zone,
+					v1.LabelInstanceTypeStable: instance.Name(),
+					v1alpha5.LabelCapacityType: capacityType,
 				},
-				Spec: v1.NodeSpec{
-					ProviderID: fmt.Sprintf("fake:///%s/%s", name, zone),
+			},
+			Spec: v1.NodeSpec{
+				ProviderID: fmt.Sprintf("fake:///%s/%s", name, zone),
+			},
+			Status: v1.NodeStatus{
+				NodeInfo: v1.NodeSystemInfo{
+					Architecture:    instance.Architecture(),
+					OperatingSystem: v1alpha5.OperatingSystemLinux,
 				},
-				Status: v1.NodeStatus{
-					NodeInfo: v1.NodeSystemInfo{
-						Architecture:    instance.Architecture(),
-						OperatingSystem: v1alpha5.OperatingSystemLinux,
-					},
-					Allocatable: v1.ResourceList{
-						v1.ResourcePods:   *instance.Pods(),
-						v1.ResourceCPU:    *instance.CPU(),
-						v1.ResourceMemory: *instance.Memory(),
-					},
+				Allocatable: v1.ResourceList{
+					v1.ResourcePods:   *instance.Pods(),
+					v1.ResourceCPU:    *instance.CPU(),
+					v1.ResourceMemory: *instance.Memory(),
 				},
-			})
-		}()
+			},
+		}))
 	}
 	return err
 }

diff --git a/pkg/cloudprovider/metrics/cloudprovider.go b/pkg/cloudprovider/metrics/cloudprovider.go
@@ -66,19 +66,9 @@ func Decorate(cloudProvider cloudprovider.CloudProvider) cloudprovider.CloudProv
 	return &decorator{cloudProvider}
 }
 
-func (d *decorator) Create(ctx context.Context, constraints *v1alpha5.Constraints, instanceTypes []cloudprovider.InstanceType, quantity int, callback func(*v1.Node) error) <-chan error {
-	recordLatency := metrics.Measure(methodDurationHistogramVec.WithLabelValues(getControllerName(ctx), "Create", d.Name()))
-	out := make(chan error)
-	go func(in <-chan error) {
-		defer recordLatency()
-		select {
-		case err := <-in:
-			out <- err
-		case <-ctx.Done():
-		}
-		close(out)
-	}(d.CloudProvider.Create(ctx, constraints, instanceTypes, quantity, callback))
-	return out
+func (d *decorator) Create(ctx context.Context, constraints *v1alpha5.Constraints, instanceTypes []cloudprovider.InstanceType, quantity int, callback func(*v1.Node) error) error {
+	defer metrics.Measure(methodDurationHistogramVec.WithLabelValues(getControllerName(ctx), "Create", d.Name()))()
+	return d.CloudProvider.Create(ctx, constraints, instanceTypes, quantity, callback)
 }
 
 func (d *decorator) Delete(ctx context.Context, node *v1.Node) error {

diff --git a/pkg/cloudprovider/types.go b/pkg/cloudprovider/types.go
@@ -30,9 +30,8 @@ type CloudProvider interface {
 	// Create a set of nodes for each of the given constraints. This API uses a
 	// callback pattern to enable cloudproviders to batch capacity creation
 	// requests. The callback must be called with a theoretical node object that
-	// is fulfilled by the cloud providers capacity creation request. This API
-	// is called in parallel and then waits for all channels to return nil or error.
-	Create(context.Context, *v1alpha5.Constraints, []InstanceType, int, func(*v1.Node) error) <-chan error
+	// is fulfilled by the cloud providers capacity creation request.
+	Create(context.Context, *v1alpha5.Constraints, []InstanceType, int, func(*v1.Node) error) error
 	// Delete node in cloudprovider
 	Delete(context.Context, *v1.Node) error
 	// GetInstanceTypes returns instance types supported by the cloudprovider.

diff --git a/pkg/controllers/provisioning/provisioner.go b/pkg/controllers/provisioning/provisioner.go
@@ -199,7 +199,7 @@ func (p *Provisioner) launch(ctx context.Context, constraints *v1alpha5.Constrai
 	for _, ps := range packing.Pods {
 		pods <- ps
 	}
-	return <-p.cloudProvider.Create(ctx, constraints, packing.InstanceTypeOptions, packing.NodeQuantity, func(node *v1.Node) error {
+	return p.cloudProvider.Create(ctx, constraints, packing.InstanceTypeOptions, packing.NodeQuantity, func(node *v1.Node) error {
 		node.Labels = functional.UnionStringMaps(node.Labels, constraints.Labels)
 		node.Spec.Taints = append(node.Spec.Taints, constraints.Taints...)
 		return p.bind(ctx, node, <-pods)