Function version provisioned concurrency (#116)

Issue #, if available:

Description of changes:

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

Author: Vandita2020

apis/v1alpha1/ack-generate-metadata.yaml

@@ -1,13 +1,13 @@
 ack_generate_info:
-  build_date: "2024-02-05T21:49:22Z"
+  build_date: "2024-02-05T22:05:18Z"
   build_hash: 5b4565ec2712d29988b8123aeeed6a4af57467bf
   go_version: go1.21.6
   version: v0.29.2-4-g5b4565e
-api_directory_checksum: ade63fbeb51f31fc6deec256bb90966c67e73825
+api_directory_checksum: c67645b15db39980ba51ff6303c34c5aafc55a9e
 api_version: v1alpha1
 aws_sdk_go_version: v1.44.181
 generator_config_info:
-  file_checksum: ad8dfe3985ccc54cbe75614e8bf1381b9212bfee
+  file_checksum: d71df9c47727f5905a2be2ac8d5cfceff3c7bbec
   original_file_name: generator.yaml
 last_modification:
   reason: API generation

apis/v1alpha1/generator.yaml

@@ -230,6 +230,10 @@ resources:
         from:
           operation: PutFunctionEventInvokeConfig
           path: .
+      ProvisionedConcurrencyConfig:
+        from:
+          operation: PutProvisionedConcurrencyConfig
+          path: .
     tags:
       ignore: true
     update_operation:

apis/v1alpha1/version.go

@@ -117,6 +117,16 @@ spec:
                         type: string
                     type: object
                 type: object
+              provisionedConcurrencyConfig:
+                properties:
+                  functionName:
+                    type: string
+                  provisionedConcurrentExecutions:
+                    format: int64
+                    type: integer
+                  qualifier:
+                    type: string
+                type: object
               revisionID:
                 description: |-
                   Only update the function if the revision ID matches the ID that's specified.

apis/v1alpha1/zz_generated.deepcopy.go

@@ -230,6 +230,10 @@ resources:
         from:
           operation: PutFunctionEventInvokeConfig
           path: .
+      ProvisionedConcurrencyConfig:
+        from:
+          operation: PutProvisionedConcurrencyConfig
+          path: .
     tags:
       ignore: true
     update_operation:

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

@@ -117,6 +117,16 @@ spec:
                         type: string
                     type: object
                 type: object
+              provisionedConcurrencyConfig:
+                properties:
+                  functionName:
+                    type: string
+                  provisionedConcurrentExecutions:
+                    format: int64
+                    type: integer
+                  qualifier:
+                    type: string
+                type: object
               revisionID:
                 description: |-
                   Only update the function if the revision ID matches the ID that's specified.

generator.yaml

@@ -71,6 +71,13 @@ func (rm *resourceManager) customUpdateVersion(
 		}
 	}
 
+	if delta.DifferentAt("Spec.ProvisionedConcurrencyConfig") {
+		err = rm.updateProvisionedConcurrency(ctx, desired)
+		if err != nil {
+			return nil, err
+		}
+	}
+
 	readOneLatest, err := rm.ReadOne(ctx, desired)
 	if err != nil {
 		return nil, err
@@ -193,6 +200,87 @@ func (rm *resourceManager) setFunctionEventInvokeConfig(
 	return nil
 }
 
+// updateProvisionedConcurrency calls `PutProvisionedConcurrencyConfig` to update the fields
+// or `DeleteProvisionedConcurrencyConfig` if users removes the fields
+func (rm *resourceManager) updateProvisionedConcurrency(
+	ctx context.Context,
+	desired *resource,
+) error {
+	var err error
+	rlog := ackrtlog.FromContext(ctx)
+	exit := rlog.Trace("rm.updateProvisionedConcurrency")
+	defer exit(err)
+
+	if desired.ko.Status.Version == nil {
+		return nil
+	}
+
+	// Check if the user deleted the 'ProvisionedConcurrency' configuration
+	// If yes, delete ProvisionedConcurrencyConfig
+	if desired.ko.Spec.ProvisionedConcurrencyConfig == nil || desired.ko.Spec.ProvisionedConcurrencyConfig.ProvisionedConcurrentExecutions == nil {
+		input_delete := &svcsdk.DeleteProvisionedConcurrencyConfigInput{
+			FunctionName: aws.String(*desired.ko.Spec.FunctionName),
+			Qualifier:    aws.String(*desired.ko.Status.Version),
+		}
+		_, err = rm.sdkapi.DeleteProvisionedConcurrencyConfigWithContext(ctx, input_delete)
+		rm.metrics.RecordAPICall("DELETE", "DeleteProvisionedConcurrency", err)
+		if err != nil {
+			return err
+		}
+		return nil
+	}
+
+	dspec := desired.ko.Spec
+	input := &svcsdk.PutProvisionedConcurrencyConfigInput{
+		FunctionName:                    aws.String(*desired.ko.Spec.FunctionName),
+		Qualifier:                       aws.String(*desired.ko.Status.Version),
+		ProvisionedConcurrentExecutions: aws.Int64(*dspec.ProvisionedConcurrencyConfig.ProvisionedConcurrentExecutions),
+	}
+
+	_, err = rm.sdkapi.PutProvisionedConcurrencyConfigWithContext(ctx, input)
+	rm.metrics.RecordAPICall("UPDATE", "UpdateProvisionedConcurrency", err)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+// setProvisionedConcurrencyConfig sets the Provisioned Concurrency
+// for the Function's Version
+func (rm *resourceManager) setProvisionedConcurrencyConfig(
+	ctx context.Context,
+	ko *svcapitypes.Version,
+) (err error) {
+	rlog := ackrtlog.FromContext(ctx)
+	exit := rlog.Trace("rm.setProvisionedConcurrencyConfig")
+	defer exit(err)
+
+	var getProvisionedConcurrencyConfigOutput *svcsdk.GetProvisionedConcurrencyConfigOutput
+	getProvisionedConcurrencyConfigOutput, err = rm.sdkapi.GetProvisionedConcurrencyConfigWithContext(
+		ctx,
+		&svcsdk.GetProvisionedConcurrencyConfigInput{
+			FunctionName: ko.Spec.FunctionName,
+			Qualifier:    ko.Status.Version,
+		},
+	)
+	rm.metrics.RecordAPICall("GET", "GetProvisionedConcurrencyConfig", err)
+
+	if err != nil {
+		if awserr, ok := ackerr.AWSError(err); ok && (awserr.Code() == "ProvisionedConcurrencyConfigNotFoundException" || awserr.Code() == "ResourceNotFoundException") {
+			ko.Spec.ProvisionedConcurrencyConfig = nil
+		} else {
+			return err
+		}
+	} else {
+		// creating ProvisionedConcurrency object to store the values returned from `Get` call
+		cloudProvisionedConcurrency := &svcapitypes.PutProvisionedConcurrencyConfigInput{}
+		cloudProvisionedConcurrency.ProvisionedConcurrentExecutions = getProvisionedConcurrencyConfigOutput.RequestedProvisionedConcurrentExecutions
+		ko.Spec.ProvisionedConcurrencyConfig = cloudProvisionedConcurrency
+	}
+
+	return nil
+}
+
 // setResourceAdditionalFields will describe the fields that are not return by the
 // getFunctionConfiguration API call
 func (rm *resourceManager) setResourceAdditionalFields(
@@ -209,5 +297,11 @@ func (rm *resourceManager) setResourceAdditionalFields(
 		return err
 	}
 
+	// To set Provisioned Concurrency for the function's version
+	err = rm.setProvisionedConcurrencyConfig(ctx, ko)
+	if err != nil {
+		return err
+	}
+
 	return nil
 }

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

@@ -3,4 +3,10 @@ if ko.Spec.FunctionEventInvokeConfig != nil {
    if err != nil{
       return nil, err
    }
+}
+if ko.Spec.ProvisionedConcurrencyConfig != nil {
+   err = rm.updateProvisionedConcurrency(ctx,desired)
+   if err != nil{
+      return nil, err
+   }
 }

pkg/resource/version/delta.go

@@ -0,0 +1,11 @@
+apiVersion: lambda.services.k8s.aws/v1alpha1
+kind: Version
+metadata:
+  name: $VERSION_NAME
+  annotations:
+    services.k8s.aws/region: $AWS_REGION
+spec:
+  functionName: $FUNCTION_NAME
+  provisionedConcurrencyConfig:
+    provisionedConcurrentExecutions: $PROVISIONED_CONCURRENT_EXECUTIONS
+  description: version created by ACK lambda-controller e2e tests

pkg/resource/version/hooks.go

@@ -91,7 +91,7 @@ def test_smoke(self, lambda_client, lambda_function):
         replacements["AWS_REGION"] = get_region()
         replacements["FUNCTION_NAME"] = lambda_function_name
         replacements["VERSION_NAME"] = resource_name
-        
+
         # Load Lambda CR
         resource_data = load_lambda_resource(
             "version",
@@ -280,7 +280,7 @@ def test_smoke_ref(self, lambda_client, lambda_function):
 
         # Check alias doesn't exist
         assert not lambda_validator.version_exists(function_resource_name, version_number)
-    
+
     def test_function_event_invoke_config(self, lambda_client, lambda_function):
         (_, function_resource) = lambda_function
         lambda_function_name = function_resource["spec"]["name"]
@@ -325,7 +325,7 @@ def test_function_event_invoke_config(self, lambda_client, lambda_function):
 
         version_number = cr['status']['version']
 
-         # Check version exists
+        # Check version exists
         assert lambda_validator.version_exists(lambda_function_name, version_number)
 
         # Update cr
@@ -359,4 +359,80 @@ def test_function_event_invoke_config(self, lambda_client, lambda_function):
         time.sleep(DELETE_WAIT_AFTER_SECONDS)
 
         # Check version doesn't exist
-        assert not lambda_validator.version_exists(lambda_function_name, version_number)
+        assert not lambda_validator.version_exists(lambda_function_name, version_number)
+    
+    def test_provisioned_concurrency_config(self, lambda_client, lambda_function):
+        (_, function_resource) = lambda_function
+        lambda_function_name = function_resource["spec"]["name"]
+
+        resource_name = random_suffix_name("lambda-version", 24)
+
+        resources = get_bootstrap_resources()
+        logging.debug(resources)
+
+        replacements = REPLACEMENT_VALUES.copy()
+        replacements["AWS_REGION"] = get_region()
+        replacements["FUNCTION_NAME"] = lambda_function_name
+        replacements["VERSION_NAME"] = resource_name
+        replacements["PROVISIONED_CONCURRENT_EXECUTIONS"] = "1"
+
+        # Load version CR
+        resource_data = load_lambda_resource(
+            "version_provisioned_concurrency",
+            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)
+
+        cr = k8s.wait_resource_consumed_by_controller(ref)
+
+        lambda_validator = LambdaValidator(lambda_client)
+
+        version_number = cr['status']['version']
+
+        # Check version exists
+        assert lambda_validator.version_exists(lambda_function_name, version_number)
+        
+        # Update provisioned_concurrency
+        cr["spec"]["provisionedConcurrencyConfig"]["provisionedConcurrentExecutions"] = 2
+
+        # Patch k8s resource
+        k8s.patch_custom_resource(ref, cr)
+        time.sleep(UPDATE_WAIT_AFTER_SECONDS)
+
+        #Check provisioned_concurrency_config update fields
+        provisioned_concurrency_config = lambda_validator.get_provisioned_concurrency_config(lambda_function_name, version_number)
+        assert provisioned_concurrency_config["RequestedProvisionedConcurrentExecutions"] == 2
+
+        # Delete provisioned_concurrency from version
+        cr = k8s.wait_resource_consumed_by_controller(ref)
+        cr["spec"]["provisionedConcurrencyConfig"] = None
+
+        # Patch k8s resource     
+        k8s.patch_custom_resource(ref, cr)
+        time.sleep(UPDATE_WAIT_AFTER_SECONDS)
+
+        #Check provisioned_concurrency_config is deleted
+        assert not lambda_validator.get_provisioned_concurrency_config(lambda_function_name, version_number)
+
+        # Delete k8s resource
+        _, deleted = k8s.delete_custom_resource(ref)
+        assert deleted
+
+        time.sleep(DELETE_WAIT_AFTER_SECONDS)
+
+        # Check version doesn't exist
+        assert not lambda_validator.version_exists(lambda_function_name, version_number)
+