Merge pull request #1886 from Jont828/async-loadbalancers

Make load balancer reconcile/delete async

Author: k8s-ci-robot

azure/scope/cluster.go

@@ -36,6 +36,7 @@ import (
 	infrav1 "sigs.k8s.io/cluster-api-provider-azure/api/v1beta1"
 	"sigs.k8s.io/cluster-api-provider-azure/azure"
 	"sigs.k8s.io/cluster-api-provider-azure/azure/services/groups"
+	"sigs.k8s.io/cluster-api-provider-azure/azure/services/loadbalancers"
 	"sigs.k8s.io/cluster-api-provider-azure/azure/services/natgateways"
 	"sigs.k8s.io/cluster-api-provider-azure/azure/services/routetables"
 	"sigs.k8s.io/cluster-api-provider-azure/azure/services/vnetpeerings"
@@ -165,11 +166,17 @@ func (s *ClusterScope) PublicIPSpecs() []azure.PublicIPSpec {
 }
 
 // LBSpecs returns the load balancer specs.
-func (s *ClusterScope) LBSpecs() []azure.LBSpec {
-	specs := []azure.LBSpec{
-		{
+func (s *ClusterScope) LBSpecs() []azure.ResourceSpecGetter {
+	specs := []azure.ResourceSpecGetter{
+		&loadbalancers.LBSpec{
 			// API Server LB
 			Name:                 s.APIServerLB().Name,
+			ResourceGroup:        s.ResourceGroup(),
+			SubscriptionID:       s.SubscriptionID(),
+			ClusterName:          s.ClusterName(),
+			Location:             s.Location(),
+			VNetName:             s.Vnet().Name,
+			VNetResourceGroup:    s.Vnet().ResourceGroup,
 			SubnetName:           s.ControlPlaneSubnet().Name,
 			FrontendIPConfigs:    s.APIServerLB().FrontendIPs,
 			APIServerPort:        s.APIServerPort(),
@@ -178,32 +185,47 @@ func (s *ClusterScope) LBSpecs() []azure.LBSpec {
 			Role:                 infrav1.APIServerRole,
 			BackendPoolName:      s.APIServerLBPoolName(s.APIServerLB().Name),
 			IdleTimeoutInMinutes: s.APIServerLB().IdleTimeoutInMinutes,
+			AdditionalTags:       s.AdditionalTags(),
 		},
 	}
 
 	// Node outbound LB
 	if s.NodeOutboundLB() != nil {
-		specs = append(specs, azure.LBSpec{
+		specs = append(specs, &loadbalancers.LBSpec{
 			Name:                 s.NodeOutboundLBName(),
+			ResourceGroup:        s.ResourceGroup(),
+			SubscriptionID:       s.SubscriptionID(),
+			ClusterName:          s.ClusterName(),
+			Location:             s.Location(),
+			VNetName:             s.Vnet().Name,
+			VNetResourceGroup:    s.Vnet().ResourceGroup,
 			FrontendIPConfigs:    s.NodeOutboundLB().FrontendIPs,
 			Type:                 s.NodeOutboundLB().Type,
 			SKU:                  s.NodeOutboundLB().SKU,
 			BackendPoolName:      s.OutboundPoolName(s.NodeOutboundLBName()),
 			IdleTimeoutInMinutes: s.NodeOutboundLB().IdleTimeoutInMinutes,
 			Role:                 infrav1.NodeOutboundRole,
+			AdditionalTags:       s.AdditionalTags(),
 		})
 	}
 
 	// Control Plane Outbound LB
 	if s.ControlPlaneOutboundLB() != nil {
-		specs = append(specs, azure.LBSpec{
+		specs = append(specs, &loadbalancers.LBSpec{
 			Name:                 s.ControlPlaneOutboundLB().Name,
+			ResourceGroup:        s.ResourceGroup(),
+			SubscriptionID:       s.SubscriptionID(),
+			ClusterName:          s.ClusterName(),
+			Location:             s.Location(),
+			VNetName:             s.Vnet().Name,
+			VNetResourceGroup:    s.Vnet().ResourceGroup,
 			FrontendIPConfigs:    s.ControlPlaneOutboundLB().FrontendIPs,
 			Type:                 s.ControlPlaneOutboundLB().Type,
 			SKU:                  s.ControlPlaneOutboundLB().SKU,
 			BackendPoolName:      s.OutboundPoolName(azure.GenerateControlPlaneOutboundLBName(s.ClusterName())),
 			IdleTimeoutInMinutes: s.NodeOutboundLB().IdleTimeoutInMinutes,
 			Role:                 infrav1.ControlPlaneOutboundRole,
+			AdditionalTags:       s.AdditionalTags(),
 		})
 	}
 
@@ -623,6 +645,7 @@ func (s *ClusterScope) PatchObject(ctx context.Context) error {
 			infrav1.VnetPeeringReadyCondition,
 			infrav1.DisksReadyCondition,
 			infrav1.NATGatewaysReadyCondition,
+			infrav1.LoadBalancersReadyCondition,
 		),
 	)
 
@@ -637,6 +660,7 @@ func (s *ClusterScope) PatchObject(ctx context.Context) error {
 			infrav1.VnetPeeringReadyCondition,
 			infrav1.DisksReadyCondition,
 			infrav1.NATGatewaysReadyCondition,
+			infrav1.LoadBalancersReadyCondition,
 		}})
 }
 

azure/services/loadbalancers/client.go

@@ -18,32 +18,28 @@ package loadbalancers
 
 import (
 	"context"
+	"encoding/json"
 
 	"github.com/Azure/azure-sdk-for-go/services/network/mgmt/2021-02-01/network"
 	"github.com/Azure/go-autorest/autorest"
+	azureautorest "github.com/Azure/go-autorest/autorest/azure"
+	"github.com/pkg/errors"
 
+	infrav1 "sigs.k8s.io/cluster-api-provider-azure/api/v1beta1"
 	"sigs.k8s.io/cluster-api-provider-azure/azure"
+	"sigs.k8s.io/cluster-api-provider-azure/util/reconciler"
 	"sigs.k8s.io/cluster-api-provider-azure/util/tele"
 )
 
-// Client wraps go-sdk.
-type Client interface {
-	Get(context.Context, string, string) (network.LoadBalancer, error)
-	CreateOrUpdate(context.Context, string, string, network.LoadBalancer) error
-	Delete(context.Context, string, string) error
-}
-
-// AzureClient contains the Azure go-sdk Client.
-type AzureClient struct {
+// azureClient contains the Azure go-sdk Client.
+type azureClient struct {
 	loadbalancers network.LoadBalancersClient
 }
 
-var _ Client = &AzureClient{}
-
-// NewClient creates a new load balancer client from subscription ID.
-func NewClient(auth azure.Authorizer) *AzureClient {
+// newClient creates a new load balancer client from subscription ID.
+func newClient(auth azure.Authorizer) *azureClient {
 	c := newLoadBalancersClient(auth.SubscriptionID(), auth.BaseURI(), auth.Authorizer())
-	return &AzureClient{c}
+	return &azureClient{c}
 }
 
 // newLoadbalancersClient creates a new load balancer client from subscription ID.
@@ -54,61 +50,129 @@ func newLoadBalancersClient(subscriptionID string, baseURI string, authorizer au
 }
 
 // Get gets the specified load balancer.
-func (ac *AzureClient) Get(ctx context.Context, resourceGroupName, lbName string) (network.LoadBalancer, error) {
-	ctx, _, done := tele.StartSpanWithLogger(ctx, "loadbalancers.AzureClient.Get")
+func (ac *azureClient) Get(ctx context.Context, spec azure.ResourceSpecGetter) (result interface{}, err error) {
+	ctx, _, done := tele.StartSpanWithLogger(ctx, "loadbalancers.azureClient.Get")
 	defer done()
 
-	return ac.loadbalancers.Get(ctx, resourceGroupName, lbName, "")
+	return ac.loadbalancers.Get(ctx, spec.ResourceGroupName(), spec.ResourceName(), "")
 }
 
-// CreateOrUpdate creates or updates a load balancer.
-func (ac *AzureClient) CreateOrUpdate(ctx context.Context, resourceGroupName string, lbName string, lb network.LoadBalancer) error {
-	ctx, _, done := tele.StartSpanWithLogger(ctx, "loadbalancers.AzureClient.CreateOrUpdate")
+// CreateOrUpdateAsync creates or updates a load balancer asynchronously.
+// It sends a PUT request to Azure and if accepted without error, the func will return a Future which can be used to track the ongoing
+// progress of the operation.
+func (ac *azureClient) CreateOrUpdateAsync(ctx context.Context, spec azure.ResourceSpecGetter, parameters interface{}) (result interface{}, future azureautorest.FutureAPI, err error) {
+	ctx, _, done := tele.StartSpanWithLogger(ctx, "loadbalancers.azureClient.CreateOrUpdate")
 	defer done()
 
+	loadBalancer, ok := parameters.(network.LoadBalancer)
+	if !ok {
+		return nil, nil, errors.Errorf("%T is not a network.LoadBalancer", parameters)
+	}
+
 	var etag string
-	if lb.Etag != nil {
-		etag = *lb.Etag
+	if loadBalancer.Etag != nil {
+		etag = *loadBalancer.Etag
 	}
 
-	req, err := ac.loadbalancers.CreateOrUpdatePreparer(ctx, resourceGroupName, lbName, lb)
+	req, err := ac.loadbalancers.CreateOrUpdatePreparer(ctx, spec.ResourceGroupName(), spec.ResourceName(), loadBalancer)
 	if err != nil {
 		err = autorest.NewErrorWithError(err, "network.LoadBalancersClient", "CreateOrUpdate", nil, "Failure preparing request")
-		return err
+		return nil, nil, err
 	}
 
 	if etag != "" {
 		req.Header.Add("If-Match", etag)
 	}
 
-	future, err := ac.loadbalancers.CreateOrUpdateSender(req)
+	createFuture, err := ac.loadbalancers.CreateOrUpdateSender(req)
 	if err != nil {
-		err = autorest.NewErrorWithError(err, "network.LoadBalancersClient", "CreateOrUpdate", future.Response(), "Failure sending request")
-		return err
+		err = autorest.NewErrorWithError(err, "network.LoadBalancersClient", "CreateOrUpdate", createFuture.Response(), "Failure sending request")
+		return nil, nil, err
 	}
 
-	err = future.WaitForCompletionRef(ctx, ac.loadbalancers.Client)
+	ctx, cancel := context.WithTimeout(ctx, reconciler.DefaultAzureCallTimeout)
+	defer cancel()
+
+	err = createFuture.WaitForCompletionRef(ctx, ac.loadbalancers.Client)
 	if err != nil {
-		return err
+		// if an error occurs, return the future.
+		// this means the long-running operation didn't finish in the specified timeout.
+		return nil, &createFuture, err
 	}
 
-	_, err = future.Result(ac.loadbalancers)
-	return err
+	result, err = createFuture.Result(ac.loadbalancers)
+	// if the operation completed, return a nil future
+	return result, nil, err
 }
 
-// Delete deletes the specified load balancer.
-func (ac *AzureClient) Delete(ctx context.Context, resourceGroupName, lbName string) error {
-	ctx, _, done := tele.StartSpanWithLogger(ctx, "loadbalancers.AzureClient.Delete")
+// DeleteAsync deletes a load balancer asynchronously. DeleteAsync sends a DELETE
+// request to Azure and if accepted without error, the func will return a Future which can be used to track the ongoing
+// progress of the operation.
+func (ac *azureClient) DeleteAsync(ctx context.Context, spec azure.ResourceSpecGetter) (future azureautorest.FutureAPI, err error) {
+	ctx, _, done := tele.StartSpanWithLogger(ctx, "loadbalancers.azureClient.Delete")
 	defer done()
 
-	future, err := ac.loadbalancers.Delete(ctx, resourceGroupName, lbName)
+	deleteFuture, err := ac.loadbalancers.Delete(ctx, spec.ResourceGroupName(), spec.ResourceName())
 	if err != nil {
-		return err
+		return nil, err
 	}
-	err = future.WaitForCompletionRef(ctx, ac.loadbalancers.Client)
+
+	ctx, cancel := context.WithTimeout(ctx, reconciler.DefaultAzureCallTimeout)
+	defer cancel()
+
+	err = deleteFuture.WaitForCompletionRef(ctx, ac.loadbalancers.Client)
+	if err != nil {
+		// if an error occurs, return the future.
+		// this means the long-running operation didn't finish in the specified timeout.
+		return &deleteFuture, err
+	}
+	_, err = deleteFuture.Result(ac.loadbalancers)
+	// if the operation completed, return a nil future.
+	return nil, err
+}
+
+// IsDone returns true if the long-running operation has completed.
+func (ac *azureClient) IsDone(ctx context.Context, future azureautorest.FutureAPI) (isDone bool, err error) {
+	ctx, _, done := tele.StartSpanWithLogger(ctx, "loadbalancers.azureClient.IsDone")
+	defer done()
+
+	isDone, err = future.DoneWithContext(ctx, ac.loadbalancers)
 	if err != nil {
-		return err
+		return false, errors.Wrap(err, "failed checking if the operation was complete")
+	}
+
+	return isDone, nil
+}
+
+// Result fetches the result of a long-running operation future.
+func (ac *azureClient) Result(ctx context.Context, future azureautorest.FutureAPI, futureType string) (result interface{}, err error) {
+	_, _, done := tele.StartSpanWithLogger(ctx, "loadbalancers.azureClient.Result")
+	defer done()
+
+	if future == nil {
+		return nil, errors.Errorf("cannot get result from nil future")
+	}
+
+	switch futureType {
+	case infrav1.PutFuture:
+		// Marshal and Unmarshal the future to put it into the correct future type so we can access the Result function.
+		// Unfortunately the FutureAPI can't be casted directly to LoadBalancersCreateOrUpdateFuture because it is a azureautorest.Future, which doesn't implement the Result function. See PR #1686 for discussion on alternatives.
+		// It was converted back to a generic azureautorest.Future from the CAPZ infrav1.Future type stored in Status: https://github.com/kubernetes-sigs/cluster-api-provider-azure/blob/main/azure/converters/futures.go#L49.
+		var createFuture *network.LoadBalancersCreateOrUpdateFuture
+		jsonData, err := future.MarshalJSON()
+		if err != nil {
+			return nil, errors.Wrap(err, "failed to marshal future")
+		}
+		if err := json.Unmarshal(jsonData, &createFuture); err != nil {
+			return nil, errors.Wrap(err, "failed to unmarshal future data")
+		}
+		return (*createFuture).Result(ac.loadbalancers)
+
+	case infrav1.DeleteFuture:
+		// Delete does not return a result load balancer
+		return nil, nil
+
+	default:
+		return nil, errors.Errorf("unknown future type %q", futureType)
 	}
-	_, err = future.Result(ac.loadbalancers)
-	return err
 }