Add operation poller unit tests (#40)

Author: astaticvoid

pkg/cloudmap/api.go

@@ -2,6 +2,7 @@ package cloudmap
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"github.com/aws/aws-cloud-map-mcs-controller-for-k8s/pkg/model"
 	"github.com/aws/aws-sdk-go-v2/aws"
@@ -16,10 +17,10 @@ import (
 // internal data structures. It manages all interactions with the AWS SDK.
 type ServiceDiscoveryApi interface {
 	// ListNamespaces returns a list of all namespaces.
-	ListNamespaces(ctx context.Context) (namespaces []*Resource, err error)
+	ListNamespaces(ctx context.Context) (namespaces []*model.Resource, err error)
 
 	// ListServices returns a list of services for a given namespace.
-	ListServices(ctx context.Context, namespaceId string) (services []*Resource, err error)
+	ListServices(ctx context.Context, namespaceId string) (services []*model.Resource, err error)
 
 	// ListInstances returns a list of service instances registered to a given service.
 	ListInstances(ctx context.Context, serviceId string) ([]*model.Endpoint, error)
@@ -51,12 +52,6 @@ type serviceDiscoveryApi struct {
 	awsFacade AwsFacade
 }
 
-// Resource encapsulates a ID/name pair
-type Resource struct {
-	Id   string
-	Name string
-}
-
 // NewServiceDiscoveryApiFromConfig creates a new AWS Cloud Map API connection manager from an AWS client config.
 func NewServiceDiscoveryApiFromConfig(cfg *aws.Config) ServiceDiscoveryApi {
 	return &serviceDiscoveryApi{
@@ -65,8 +60,8 @@ func NewServiceDiscoveryApiFromConfig(cfg *aws.Config) ServiceDiscoveryApi {
 	}
 }
 
-func (sdApi *serviceDiscoveryApi) ListNamespaces(ctx context.Context) ([]*Resource, error) {
-	namespaces := make([]*Resource, 0)
+func (sdApi *serviceDiscoveryApi) ListNamespaces(ctx context.Context) ([]*model.Resource, error) {
+	namespaces := make([]*model.Resource, 0)
 	pages := sd.NewListNamespacesPaginator(sdApi.awsFacade, &sd.ListNamespacesInput{})
 
 	for pages.HasMorePages() {
@@ -76,7 +71,7 @@ func (sdApi *serviceDiscoveryApi) ListNamespaces(ctx context.Context) ([]*Resour
 		}
 
 		for _, ns := range output.Namespaces {
-			namespaces = append(namespaces, &Resource{
+			namespaces = append(namespaces, &model.Resource{
 				Id:   aws.ToString(ns.Id),
 				Name: aws.ToString(ns.Name),
 			})
@@ -86,8 +81,8 @@ func (sdApi *serviceDiscoveryApi) ListNamespaces(ctx context.Context) ([]*Resour
 	return namespaces, nil
 }
 
-func (sdApi *serviceDiscoveryApi) ListServices(ctx context.Context, nsId string) ([]*Resource, error) {
-	svcs := make([]*Resource, 0)
+func (sdApi *serviceDiscoveryApi) ListServices(ctx context.Context, nsId string) ([]*model.Resource, error) {
+	svcs := make([]*model.Resource, 0)
 
 	filter := types.ServiceFilter{
 		Name:   types.ServiceFilterNameNamespaceId,
@@ -103,7 +98,7 @@ func (sdApi *serviceDiscoveryApi) ListServices(ctx context.Context, nsId string)
 		}
 
 		for _, svc := range output.Services {
-			svcs = append(svcs, &Resource{
+			svcs = append(svcs, &model.Resource{
 				Id:   aws.ToString(svc.Id),
 				Name: aws.ToString(svc.Name),
 			})
@@ -226,12 +221,12 @@ func (sdApi *serviceDiscoveryApi) DeregisterInstance(ctx context.Context, svcId
 }
 
 func (sdApi *serviceDiscoveryApi) PollCreateNamespace(ctx context.Context, opId string) (nsId string, err error) {
-	return nsId, wait.Poll(defaultOperationPollInterval, defaultOperationPollTimeout, func() (done bool, pollErr error) {
+	err = wait.Poll(defaultOperationPollInterval, defaultOperationPollTimeout, func() (done bool, err error) {
 		sdApi.log.Info("polling operation", "opId", opId)
-		op, opErr := sdApi.GetOperation(ctx, opId)
+		op, err := sdApi.GetOperation(ctx, opId)
 
-		if opErr != nil {
-			return true, opErr
+		if err != nil {
+			return true, err
 		}
 
 		if op.Status == types.OperationStatusFail {
@@ -246,4 +241,10 @@ func (sdApi *serviceDiscoveryApi) PollCreateNamespace(ctx context.Context, opId
 
 		return false, nil
 	})
+
+	if err == wait.ErrWaitTimeout {
+		err = errors.New(operationPollTimoutErrorMessage)
+	}
+
+	return nsId, err
 }

pkg/cloudmap/operation_poller.go

@@ -2,7 +2,7 @@ package cloudmap
 
 import (
 	"context"
-	"fmt"
+	"errors"
 	"github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go-v2/service/servicediscovery/types"
 	"github.com/go-logr/logr"
@@ -18,6 +18,8 @@ const (
 
 	// Time until we stop polling the operation
 	defaultOperationPollTimeout = 5 * time.Minute
+
+	operationPollTimoutErrorMessage = "timed out while polling operations"
 )
 
 // OperationPoller polls a list operations for a terminal status.
@@ -27,19 +29,21 @@ type OperationPoller interface {
 }
 
 type operationPoller struct {
-	log   logr.Logger
-	sdApi ServiceDiscoveryApi
-	opIds []string
+	log     logr.Logger
+	sdApi   ServiceDiscoveryApi
+	timeout time.Duration
 
+	opIds  []string
 	svcId  string
 	opType types.OperationType
 	start  int64
 }
 
 func newOperationPoller(sdApi ServiceDiscoveryApi, svcId string, opIds []string, startTime int64) operationPoller {
 	return operationPoller{
-		log:   ctrl.Log.WithName("cloudmap"),
-		sdApi: sdApi,
+		log:     ctrl.Log.WithName("cloudmap"),
+		sdApi:   sdApi,
+		timeout: defaultOperationPollTimeout,
 
 		opIds: opIds,
 		svcId: svcId,
@@ -67,7 +71,7 @@ func (opPoller *operationPoller) Poll(ctx context.Context) (err error) {
 		return nil
 	}
 
-	return wait.Poll(defaultOperationPollInterval, defaultOperationPollTimeout, func() (done bool, err error) {
+	err = wait.Poll(defaultOperationPollInterval, opPoller.timeout, func() (done bool, err error) {
 		opPoller.log.Info("polling operations", "operations", opPoller.opIds)
 
 		sdOps, err := opPoller.sdApi.ListOperations(ctx, opPoller.buildFilters())
@@ -92,14 +96,20 @@ func (opPoller *operationPoller) Poll(ctx context.Context) (err error) {
 
 		if len(failedOps) != 0 {
 			for _, failedOp := range failedOps {
-				opPoller.log.Info("Operation failed", "failedOp", failedOp, "reason", opPoller.getFailedOpReason(ctx, failedOp))
+				opPoller.log.Info("operation failed", "failedOp", failedOp, "reason", opPoller.getFailedOpReason(ctx, failedOp))
 			}
-			return true, fmt.Errorf("operation failure")
+			return true, errors.New("operation failure")
 		}
 
 		opPoller.log.Info("operations completed successfully")
 		return true, nil
 	})
+
+	if err == wait.ErrWaitTimeout {
+		return errors.New(operationPollTimoutErrorMessage)
+	}
+
+	return err
 }
 
 func (opPoller *operationPoller) buildFilters() []types.OperationFilter {