Add reservedConcurrentExecutions field for Function Resource (#12)

The AWS Lambda API support setting up reserved concurrent executions for
a lambda functions. This patch adds a new field to allow users to set the
concurrency configuration for functions.

Description of changes:
- Add `reservedConcurrentExecutions` fields for `Function` Resource
- Add a new e2e test case to specifically test update and create
operations on function resources with a non-nil
`reservedConcurrentExecutions`

By submitting this pull request, I confirm that my contribution is made under
the terms of the Apache 2.0 license.

Author: a-hilaly

apis/v1alpha1/ack-generate-metadata.yaml

@@ -1,13 +1,13 @@
 ack_generate_info:
-  build_date: "2021-11-19T13:19:34Z"
+  build_date: "2021-11-23T23:35:17Z"
   build_hash: dca757058c55b80b4eb20f62f55829981a70f7de
   go_version: go1.16.4
   version: v0.15.2
-api_directory_checksum: 84d93710fb0b89a7ed99b0923611297242a55a99
+api_directory_checksum: a3c5e80eca3fc5591e8a0a2763d048f2ed4a6ddd
 api_version: v1alpha1
 aws_sdk_go_version: v1.40.28
 generator_config_info:
-  file_checksum: 4f4df395ccad50e23ed16cffcf1d4d24a909f9cc
+  file_checksum: 47d63e3b8348d4f33111e6a3cbd6eca2412e9b46
   original_file_name: generator.yaml
 last_modification:
   reason: API generation

apis/v1alpha1/function.go

@@ -12,10 +12,10 @@ resources:
       Name:
         is_primary_key: true
         is_required: true
-      # ReservedConcurrentExecutions:
-      #  from:
-      #    operation: PutFunctionConcurrency
-      #    path: ReservedConcurrentExecutions
+      ReservedConcurrentExecutions:
+        from:
+          operation: PutFunctionConcurrency
+          path: ReservedConcurrentExecutions
       Code:
         compare:
           is_ignored: true

apis/v1alpha1/generator.yaml

@@ -137,6 +137,11 @@ spec:
                 description: Set to true to publish the first version of the function
                   during creation.
                 type: boolean
+              reservedConcurrentExecutions:
+                description: The number of simultaneous executions to reserve for
+                  the function.
+                format: int64
+                type: integer
               role:
                 description: The Amazon Resource Name (ARN) of the function's execution
                   role.

apis/v1alpha1/zz_generated.deepcopy.go

@@ -12,10 +12,10 @@ resources:
       Name:
         is_primary_key: true
         is_required: true
-      # ReservedConcurrentExecutions:
-      #  from:
-      #    operation: PutFunctionConcurrency
-      #    path: ReservedConcurrentExecutions
+      ReservedConcurrentExecutions:
+        from:
+          operation: PutFunctionConcurrency
+          path: ReservedConcurrentExecutions
       Code:
         compare:
           is_ignored: true

config/crd/bases/lambda.services.k8s.aws_functions.yaml

@@ -137,6 +137,11 @@ spec:
                 description: Set to true to publish the first version of the function
                   during creation.
                 type: boolean
+              reservedConcurrentExecutions:
+                description: The number of simultaneous executions to reserve for
+                  the function.
+                format: int64
+                type: integer
               role:
                 description: The Amazon Resource Name (ARN) of the function's execution
                   role.

generator.yaml

@@ -16,6 +16,7 @@ package function
 import (
 	"context"
 
+	svcapitypes "github.com/aws-controllers-k8s/lambda-controller/apis/v1alpha1"
 	ackcompare "github.com/aws-controllers-k8s/runtime/pkg/compare"
 	ackrtlog "github.com/aws-controllers-k8s/runtime/pkg/runtime/log"
 	"github.com/aws/aws-sdk-go/aws"
@@ -63,6 +64,12 @@ func (rm *resourceManager) customUpdateFunction(
 			return nil, err
 		}
 	}
+	if delta.DifferentAt("Spec.ReservedConcurrentExecutions") {
+		err = rm.updateFunctionConcurrency(ctx, desired)
+		if err != nil {
+			return nil, err
+		}
+	}
 
 	readOneLatest, err := rm.ReadOne(ctx, desired)
 	if err != nil {
@@ -346,3 +353,58 @@ func customPreCompare(
 		}
 	}
 }
+
+// updateFunctionConcurrency calls UpdateFunctionConcurrency to update a specific
+// lambda function reserved concurrent executions.
+func (rm *resourceManager) updateFunctionConcurrency(
+	ctx context.Context,
+	desired *resource,
+) error {
+	var err error
+	rlog := ackrtlog.FromContext(ctx)
+	exit := rlog.Trace("rm.updateFunctionConcurrency")
+	defer exit(err)
+
+	dspec := desired.ko.Spec
+	input := &svcsdk.PutFunctionConcurrencyInput{
+		FunctionName: aws.String(*dspec.Name),
+	}
+
+	if desired.ko.Spec.ReservedConcurrentExecutions != nil {
+		input.ReservedConcurrentExecutions = aws.Int64(*desired.ko.Spec.ReservedConcurrentExecutions)
+	} else {
+		input.ReservedConcurrentExecutions = aws.Int64(0)
+	}
+
+	_, err = rm.sdkapi.PutFunctionConcurrencyWithContext(ctx, input)
+	rm.metrics.RecordAPICall("UPDATE", "PutFunctionConcurrency", err)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+// setResourceAdditionalFields will describe the fields that are not return by
+// GetFunctionConcurrency calls
+func (rm *resourceManager) setResourceAdditionalFields(
+	ctx context.Context,
+	ko *svcapitypes.Function,
+) (err error) {
+	rlog := ackrtlog.FromContext(ctx)
+	exit := rlog.Trace("rm.setResourceAdditionalFields")
+	defer exit(err)
+
+	var getFunctionConcurrencyOutput *svcsdk.GetFunctionConcurrencyOutput
+	getFunctionConcurrencyOutput, err = rm.sdkapi.GetFunctionConcurrencyWithContext(
+		ctx,
+		&svcsdk.GetFunctionConcurrencyInput{
+			FunctionName: ko.Spec.Name,
+		},
+	)
+	rm.metrics.RecordAPICall("GET", "GetFunctionConcurrency", err)
+	if err != nil {
+		return err
+	}
+	ko.Spec.ReservedConcurrentExecutions = getFunctionConcurrencyOutput.ReservedConcurrentExecutions
+	return nil
+}

helm/crds/lambda.services.k8s.aws_functions.yaml

@@ -235,4 +235,7 @@
 		ko.Spec.VPCConfig = f30
 	} else {
 		ko.Spec.VPCConfig = nil
+	}
+	if err := rm.setResourceAdditionalFields(ctx, ko); err != nil {
+		return nil, err
 	}

pkg/resource/function/delta.go

@@ -12,4 +12,5 @@ spec:
   role: $LAMBDA_ROLE
   runtime: python3.9
   handler: main
-  description: function created by ACK lambda-controller e2e tests
+  description: function created by ACK lambda-controller e2e tests
+  reservedConcurrentExecutions: $RESERVED_CONCURRENT_EXECUTIONS

pkg/resource/function/hooks.go

@@ -47,6 +47,7 @@ def lambda_function():
         replacements["BUCKET_NAME"] = resources.FunctionsBucketName
         replacements["LAMBDA_ROLE"] = resources.LambdaBasicRoleARN
         replacements["LAMBDA_FILE_NAME"] = resources.LambdaFunctionFileZip
+        replacements["RESERVED_CONCURRENT_EXECUTIONS"] = "0"
         replacements["AWS_REGION"] = get_region()
 
         # Load function CR

pkg/resource/function/sdk.go

@@ -47,6 +47,7 @@ def lambda_function():
         replacements["BUCKET_NAME"] = resources.FunctionsBucketName
         replacements["LAMBDA_ROLE"] = resources.LambdaESMRoleARN
         replacements["LAMBDA_FILE_NAME"] = resources.LambdaFunctionFileZip
+        replacements["RESERVED_CONCURRENT_EXECUTIONS"] = "0"
         replacements["AWS_REGION"] = get_region()
 
         # Load function CR

templates/hooks/function/sdk_read_one_post_set_output.go.tpl

@@ -52,6 +52,17 @@ def get_function(self, lambda_client, function_name: str) -> dict:
             logging.debug(e)
             return None
 
+    def get_function_concurrency(self, lambda_client, function_name: str) -> int:
+        try:
+            resp = lambda_client.get_function_concurrency(
+                FunctionName=function_name
+            )
+            return resp['ReservedConcurrentExecutions']
+
+        except Exception as e:
+            logging.debug(e)
+            return None
+
     def function_exists(self, lambda_client, function_name: str) -> bool:
         return self.get_function(lambda_client, function_name) is not None
 
@@ -66,6 +77,7 @@ def test_smoke(self, lambda_client):
         replacements["BUCKET_NAME"] = resources.FunctionsBucketName
         replacements["LAMBDA_ROLE"] = resources.LambdaBasicRoleARN
         replacements["LAMBDA_FILE_NAME"] = resources.LambdaFunctionFileZip
+        replacements["RESERVED_CONCURRENT_EXECUTIONS"] = "0"
         replacements["AWS_REGION"] = get_region()
 
         # Load Lambda CR
@@ -121,3 +133,64 @@ def test_smoke(self, lambda_client):
         exists = self.function_exists(lambda_client, resource_name)
         assert not exists
 
+    def test_reserved_concurrent_executions(self, lambda_client):
+        resource_name = random_suffix_name("lambda-function", 24)
+
+        resources = get_bootstrap_resources()
+        logging.debug(resources)
+
+        replacements = REPLACEMENT_VALUES.copy()
+        replacements["FUNCTION_NAME"] = resource_name
+        replacements["BUCKET_NAME"] = resources.FunctionsBucketName
+        replacements["LAMBDA_ROLE"] = resources.LambdaBasicRoleARN
+        replacements["LAMBDA_FILE_NAME"] = resources.LambdaFunctionFileZip
+        replacements["RESERVED_CONCURRENT_EXECUTIONS"] = "2"
+        replacements["AWS_REGION"] = get_region()
+
+        # Load Lambda CR
+        resource_data = load_lambda_resource(
+            "function",
+            additional_replacements=replacements,
+        )
+        logging.debug(resource_data)
+
+        # Create k8s resource
+        ref = k8s.CustomResourceReference(
+            CRD_GROUP, CRD_VERSION, RESOURCE_PLURAL,
+            resource_name, namespace="default",
+        )
+        k8s.create_custom_resource(ref, resource_data)
+        cr = k8s.wait_resource_consumed_by_controller(ref)
+
+        assert cr is not None
+        assert k8s.get_resource_exists(ref)
+
+        time.sleep(CREATE_WAIT_AFTER_SECONDS)
+
+        # Check Lambda function exists
+        exists = self.function_exists(lambda_client, resource_name)
+        assert exists
+
+        reservedConcurrentExecutions = self.get_function_concurrency(lambda_client, resource_name)
+        assert reservedConcurrentExecutions == 2
+
+        # Update cr
+        cr["spec"]["reservedConcurrentExecutions"] = 0
+
+        # Patch k8s resource
+        k8s.patch_custom_resource(ref, cr)
+        time.sleep(UPDATE_WAIT_AFTER_SECONDS)
+
+        # Check function updated fields
+        reservedConcurrentExecutions = self.get_function_concurrency(lambda_client, resource_name)
+        assert reservedConcurrentExecutions == 0
+
+        # Delete k8s resource
+        _, deleted = k8s.delete_custom_resource(ref)
+        assert deleted is True
+
+        time.sleep(DELETE_WAIT_AFTER_SECONDS)
+
+        # Check Lambda function doesn't exist
+        exists = self.function_exists(lambda_client, resource_name)
+        assert not exists