Support JobRun ConfigurationOverrides field (#21)

dalbhanj · web-flow · commit b94345e20c4f · 2022-10-26T08:45:39.000-07:00
Issue #, if available: aws-controllers-k8s/community#1445 Description of changes: - Support JobRun `ConfigurationOverrides` field By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
diff --git a/apis/v1alpha1/ack-generate-metadata.yaml b/apis/v1alpha1/ack-generate-metadata.yaml
@@ -1,13 +1,13 @@
 ack_generate_info:
-  build_date: "2022-09-12T20:46:01Z"
+  build_date: "2022-10-26T14:38:17Z"
   build_hash: 2944c8772f216656d84ee02d392eaca501274c1e
   go_version: go1.18.1
-  version: v0.20.1
-api_directory_checksum: 2322dd513cbf25a6b2e4b99236b5f3ee0fb8c22e
+  version: v0.20.1-dirty
+api_directory_checksum: bf57d462582340d7b4409d8f75f13a5f6552a1f6
 api_version: v1alpha1
 aws_sdk_go_version: v1.44.93
 generator_config_info:
-  file_checksum: 8452039ec1ac012add845247a98a2f640647b7b0
+  file_checksum: a0f3b99e009427dd814f4d54ea99932c78bc3502
   original_file_name: generator.yaml
 last_modification:
   reason: API generation
diff --git a/apis/v1alpha1/generator.yaml b/apis/v1alpha1/generator.yaml
@@ -45,6 +45,20 @@ resources:
         is_immutable: true
       JobDriver:
         is_immutable: true
+      # The default ConfigurationOverrides that is part of JobRun API's does not work well with ACK. By design a CRD must be describable using an OpenAPIv3 schema. ConfigurationOverrides has a property within ApplicationConfiguration called 'configurations' which is recursive and is the root of the issue. Hence, we are creating custom JSON blob to accomodate ConfigurationOverrides and we have to manually handle how to unmarshall this field so that customers can define all/any properties available
+      configurationOverrides:
+        type: "string"
+        is_immutable: true
+        is_required: False
+        compare: 
+          is_ignored: true
+    hooks:
+      delta_pre_compare:
+        code: customPreCompare(delta, a, b)
+      sdk_create_post_build_request:
+        template_path: hooks/configurationOverrides/sdk_create_post_build_request.go.tpl
+      sdk_read_one_pre_set_output:
+        template_path: hooks/configurationOverrides/sdk_read_one_pre_set_output.go.tpl
     exceptions:
       terminal_codes:
       - ValidationException
diff --git a/apis/v1alpha1/job_run.go b/apis/v1alpha1/job_run.go
diff --git a/apis/v1alpha1/zz_generated.deepcopy.go b/apis/v1alpha1/zz_generated.deepcopy.go
diff --git a/config/crd/bases/emrcontainers.services.k8s.aws_jobruns.yaml b/config/crd/bases/emrcontainers.services.k8s.aws_jobruns.yaml
@@ -38,6 +38,8 @@ spec:
               jar, PySpark script, or SparkSQL query, that you submit to Amazon EMR
               on EKS."
             properties:
+              configurationOverrides:
+                type: string
               executionRoleARN:
                 description: The execution role ARN for the job run.
                 type: string
diff --git a/generator.yaml b/generator.yaml
@@ -45,6 +45,20 @@ resources:
         is_immutable: true
       JobDriver:
         is_immutable: true
+      # The default ConfigurationOverrides that is part of JobRun API's does not work well with ACK. By design a CRD must be describable using an OpenAPIv3 schema. ConfigurationOverrides has a property within ApplicationConfiguration called 'configurations' which is recursive and is the root of the issue. Hence, we are creating custom JSON blob to accomodate ConfigurationOverrides and we have to manually handle how to unmarshall this field so that customers can define all/any properties available
+      configurationOverrides:
+        type: "string"
+        is_immutable: true
+        is_required: False
+        compare: 
+          is_ignored: true
+    hooks:
+      delta_pre_compare:
+        code: customPreCompare(delta, a, b)
+      sdk_create_post_build_request:
+        template_path: hooks/configurationOverrides/sdk_create_post_build_request.go.tpl
+      sdk_read_one_pre_set_output:
+        template_path: hooks/configurationOverrides/sdk_read_one_pre_set_output.go.tpl
     exceptions:
       terminal_codes:
       - ValidationException
diff --git a/go.mod b/go.mod
@@ -5,6 +5,7 @@ go 1.17
 require (
 	github.com/aws-controllers-k8s/runtime v0.20.1
 	github.com/aws/aws-sdk-go v1.44.93
+	github.com/ghodss/yaml v1.0.0
 	github.com/go-logr/logr v1.2.0
 	github.com/spf13/pflag v1.0.5
 	k8s.io/api v0.23.0
diff --git a/go.sum b/go.sum
@@ -140,6 +140,7 @@ github.com/fsnotify/fsnotify v1.5.1 h1:mZcQUHVQUQWoPXXtuf9yuEXKudkV2sx1E06UadKWp
 github.com/fsnotify/fsnotify v1.5.1/go.mod h1:T3375wBYaZdLLcVNkcVbzGHY7f1l/uK5T5Ai1i3InKU=
 github.com/getkin/kin-openapi v0.76.0/go.mod h1:660oXbgy5JFMKreazJaQTw7o+X00qeSyhcnluiMv+Xg=
 github.com/getsentry/raven-go v0.2.0/go.mod h1:KungGk8q33+aIAZUIVWZDr2OfAEBsO49PX4NzFV5kcQ=
+github.com/ghodss/yaml v1.0.0 h1:wQHKEahhL6wmXdzwWG11gIVCkOv05bNOh+Rxn0yngAk=
 github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
 github.com/go-gl/glfw v0.0.0-20190409004039-e6da0acd62b1/go.mod h1:vR7hzQXu2zJy9AVAgeJqvqgH9Q5CA+iKCZ2gyEVpxRU=
 github.com/go-gl/glfw/v3.3/glfw v0.0.0-20191125211704-12ad95a8df72/go.mod h1:tQ2UAYgL5IevRw8kRxooKSPJfGvJ9fJQFa0TUsXzTg8=
diff --git a/helm/crds/emrcontainers.services.k8s.aws_jobruns.yaml b/helm/crds/emrcontainers.services.k8s.aws_jobruns.yaml
@@ -38,6 +38,8 @@ spec:
               jar, PySpark script, or SparkSQL query, that you submit to Amazon EMR
               on EKS."
             properties:
+              configurationOverrides:
+                type: string
               executionRoleARN:
                 description: The execution role ARN for the job run.
                 type: string
diff --git a/pkg/resource/job_run/delta.go b/pkg/resource/job_run/delta.go
diff --git a/pkg/resource/job_run/hooks.go b/pkg/resource/job_run/hooks.go
@@ -0,0 +1,129 @@
+package job_run
+
+import (
+	"reflect"
+
+	ackcompare "github.com/aws-controllers-k8s/runtime/pkg/compare"
+	"github.com/aws/aws-sdk-go/aws"
+	svcsdk "github.com/aws/aws-sdk-go/service/emrcontainers"
+	"github.com/ghodss/yaml"
+)
+
+func configurationOverridesToString(cfg *svcsdk.ConfigurationOverrides) (*string, error) {
+	configBytes, err := yaml.Marshal(cfg)
+	if err != nil {
+		return nil, err
+	}
+	configStr := string(configBytes)
+	return &configStr, nil
+}
+
+func stringToConfigurationOverrides(cfg *string) (*svcsdk.ConfigurationOverrides, error) {
+	if cfg == nil {
+		cfg = aws.String("")
+	}
+
+	var config svcsdk.ConfigurationOverrides
+	err := yaml.Unmarshal([]byte(*cfg), &config)
+	if err != nil {
+		return nil, err
+	}
+	return &config, nil
+}
+
+func customPreCompare(
+	delta *ackcompare.Delta,
+	a *resource,
+	b *resource,
+) {
+	aConfig, err := stringToConfigurationOverrides(a.ko.Spec.ConfigurationOverrides)
+	if err != nil {
+		panic(err)
+	}
+	bConfig, err := stringToConfigurationOverrides(b.ko.Spec.ConfigurationOverrides)
+	if err != nil {
+		panic(err)
+	}
+
+	// background:
+	// API Always return a non empty configuration
+	// Users can give empty configuration, and API still might return something
+	// Users can give empty fields and API still might return something
+
+	// If we have a nil configuration and API returns a non nil configuration - something is wrong
+	if ackcompare.HasNilDifference(aConfig.MonitoringConfiguration, bConfig.MonitoringConfiguration) {
+		delta.Add("Spec.ConfigurationOverrides", aConfig.MonitoringConfiguration, bConfig.MonitoringConfiguration)
+	} else if aConfig.MonitoringConfiguration != nil && bConfig.MonitoringConfiguration != nil {
+		if ackcompare.HasNilDifference(aConfig.MonitoringConfiguration.PersistentAppUI, bConfig.MonitoringConfiguration.PersistentAppUI) {
+			if aConfig.MonitoringConfiguration.PersistentAppUI == nil && *bConfig.MonitoringConfiguration.PersistentAppUI == "ENABLED" {
+				// We do not consider this as a difference because the API defaults PersistentAppUI to "ENABLED"
+			} else {
+				delta.Add("Spec.ConfigurationOverrides.PersistentAppUI", aConfig.MonitoringConfiguration.PersistentAppUI, bConfig.MonitoringConfiguration.PersistentAppUI)
+			}
+		} else if aConfig.MonitoringConfiguration.PersistentAppUI != nil && bConfig.MonitoringConfiguration.PersistentAppUI != nil {
+			if *aConfig.MonitoringConfiguration.PersistentAppUI != *bConfig.MonitoringConfiguration.PersistentAppUI {
+				delta.Add("Spec.ConfigurationOverrides.PersistentAppUI", aConfig.MonitoringConfiguration.PersistentAppUI, bConfig.MonitoringConfiguration.PersistentAppUI)
+			}
+		}
+		if ackcompare.HasNilDifference(
+			aConfig.MonitoringConfiguration.CloudWatchMonitoringConfiguration,
+			bConfig.MonitoringConfiguration.CloudWatchMonitoringConfiguration,
+		) {
+			delta.Add("Spec.ConfigurationOverrides.CloudWatchMonitoringConfiguration", aConfig.MonitoringConfiguration.CloudWatchMonitoringConfiguration, bConfig.MonitoringConfiguration.CloudWatchMonitoringConfiguration)
+		} else if aConfig.MonitoringConfiguration.CloudWatchMonitoringConfiguration != nil &&
+			bConfig.MonitoringConfiguration.CloudWatchMonitoringConfiguration != nil {
+
+			if ackcompare.HasNilDifference(
+				aConfig.MonitoringConfiguration.CloudWatchMonitoringConfiguration.LogGroupName,
+				bConfig.MonitoringConfiguration.CloudWatchMonitoringConfiguration.LogGroupName,
+			) {
+				delta.Add("Spec.ConfigurationOverrides.CloudWatchMonitoringConfiguration.LogGroupName", aConfig.MonitoringConfiguration.CloudWatchMonitoringConfiguration.LogGroupName, bConfig.MonitoringConfiguration.CloudWatchMonitoringConfiguration.LogGroupName)
+			} else if *aConfig.MonitoringConfiguration.CloudWatchMonitoringConfiguration.LogGroupName !=
+				*bConfig.MonitoringConfiguration.CloudWatchMonitoringConfiguration.LogGroupName {
+				delta.Add("Spec.ConfigurationOverrides.CloudWatchMonitoringConfiguration.LogGroupName", aConfig.MonitoringConfiguration.CloudWatchMonitoringConfiguration.LogGroupName, bConfig.MonitoringConfiguration.CloudWatchMonitoringConfiguration.LogGroupName)
+			}
+			if ackcompare.HasNilDifference(
+				aConfig.MonitoringConfiguration.CloudWatchMonitoringConfiguration.LogStreamNamePrefix,
+				bConfig.MonitoringConfiguration.CloudWatchMonitoringConfiguration.LogStreamNamePrefix,
+			) {
+				delta.Add("Spec.ConfigurationOverrides.CloudWatchMonitoringConfiguration.LogStreamNamePrefix", aConfig.MonitoringConfiguration.CloudWatchMonitoringConfiguration.LogStreamNamePrefix, bConfig.MonitoringConfiguration.CloudWatchMonitoringConfiguration.LogStreamNamePrefix)
+			} else if *aConfig.MonitoringConfiguration.CloudWatchMonitoringConfiguration.LogStreamNamePrefix !=
+				*bConfig.MonitoringConfiguration.CloudWatchMonitoringConfiguration.LogStreamNamePrefix {
+				delta.Add("Spec.ConfigurationOverrides.CloudWatchMonitoringConfiguration.LogStreamNamePrefix", aConfig.MonitoringConfiguration.CloudWatchMonitoringConfiguration.LogStreamNamePrefix, bConfig.MonitoringConfiguration.CloudWatchMonitoringConfiguration.LogStreamNamePrefix)
+			}
+		}
+
+		//
+
+		if ackcompare.HasNilDifference(
+			aConfig.MonitoringConfiguration.S3MonitoringConfiguration,
+			bConfig.MonitoringConfiguration.S3MonitoringConfiguration,
+		) {
+			delta.Add("Spec.ConfigurationOverrides.S3MonitoringConfiguration", aConfig.MonitoringConfiguration.S3MonitoringConfiguration, bConfig.MonitoringConfiguration.S3MonitoringConfiguration)
+		} else if aConfig.MonitoringConfiguration.S3MonitoringConfiguration != nil &&
+			bConfig.MonitoringConfiguration.S3MonitoringConfiguration != nil {
+			if ackcompare.HasNilDifference(
+				aConfig.MonitoringConfiguration.S3MonitoringConfiguration.LogUri,
+				bConfig.MonitoringConfiguration.S3MonitoringConfiguration.LogUri,
+			) {
+				delta.Add("Spec.ConfigurationOverrides.S3MonitoringConfiguration.LogUri", aConfig.MonitoringConfiguration.S3MonitoringConfiguration.LogUri, bConfig.MonitoringConfiguration.S3MonitoringConfiguration.LogUri)
+			} else if *aConfig.MonitoringConfiguration.S3MonitoringConfiguration.LogUri !=
+				*bConfig.MonitoringConfiguration.S3MonitoringConfiguration.LogUri {
+				delta.Add("Spec.ConfigurationOverrides.S3MonitoringConfiguration.LogUri", aConfig.MonitoringConfiguration.S3MonitoringConfiguration.LogUri, bConfig.MonitoringConfiguration.S3MonitoringConfiguration.LogUri)
+			}
+		}
+
+	}
+
+	// If two arrays have different sizes then automatically they are different
+	if len(aConfig.ApplicationConfiguration) != len(bConfig.ApplicationConfiguration) {
+		delta.Add("Spec.ApplicationConfiguration", aConfig.ApplicationConfiguration, bConfig.ApplicationConfiguration)
+	} else if len(aConfig.ApplicationConfiguration) > 0 {
+		// at this stage we know that they have the same size they contain at least one element
+		// We assume that the EMRContainer API doesn't mess with the order of the provided Application
+		// Configuration (To verify).
+		if !reflect.DeepEqual(aConfig.ApplicationConfiguration, bConfig.ApplicationConfiguration) {
+			delta.Add("Spec.ApplicationConfiguration", aConfig.ApplicationConfiguration, bConfig.ApplicationConfiguration)
+		}
+	}
+}
diff --git a/pkg/resource/job_run/sdk.go b/pkg/resource/job_run/sdk.go
diff --git a/templates/hooks/configurationOverrides/sdk_create_post_build_request.go.tpl b/templates/hooks/configurationOverrides/sdk_create_post_build_request.go.tpl
@@ -0,0 +1,7 @@
+    // Unmarshall ConfigurationOverrides structure
+    if desired.ko.Spec.ConfigurationOverrides != nil {
+        input.ConfigurationOverrides, err = stringToConfigurationOverrides(desired.ko.Spec.ConfigurationOverrides)
+        if err != nil {
+          return nil, err
+        }
+    }
diff --git a/templates/hooks/configurationOverrides/sdk_read_one_pre_set_output.go.tpl b/templates/hooks/configurationOverrides/sdk_read_one_pre_set_output.go.tpl
@@ -0,0 +1,7 @@
+		// DescribeJobRun should output ConfigurationOverrides and show all available configuration
+		if resp.JobRun.ConfigurationOverrides != nil {
+			ko.Spec.ConfigurationOverrides, err = configurationOverridesToString(resp.JobRun.ConfigurationOverrides)
+			if err != nil {
+				return nil, err
+			}
+		}
diff --git a/test/e2e/bootstrap_resources.py b/test/e2e/bootstrap_resources.py
@@ -16,6 +16,8 @@
 """
 
 from dataclasses import dataclass
+
+from acktest.bootstrapping.s3 import Bucket
 from acktest.bootstrapping import Resources
 from acktest.bootstrapping.iam import Role
 from e2e import bootstrap_directory
@@ -24,6 +26,7 @@
 @dataclass
 class BootstrapResources(Resources):
     JobExecutionRole: Role
+    EMREKSS3BucketName: Bucket
     HostCluster_VC: EMREnabledEKSCluster
     HostCluster_JR: EMREnabledEKSCluster
 
diff --git a/test/e2e/resources/job_run.yaml b/test/e2e/resources/job_run.yaml
@@ -4,19 +4,19 @@ metadata:
   name: $JOBRUN_NAME
 spec:
   name: $JOBRUN_NAME
-  virtualClusterID: $VIRTUALCLUSTER_ID
+  virtualClusterRef:
+    from:
+      name: $VIRTUALCLUSTER_NAME
   executionRoleARN: $JOB_EXECUTION_ROLE
   releaseLabel: $EMR_RELEASE_LABEL
   jobDriver:
     sparkSubmitJobDriver:
       entryPoint: "local:///usr/lib/spark/examples/src/main/python/pi.py"
       entryPointArguments:
       sparkSubmitParameters: "--conf spark.executor.instances=2 --conf spark.executor.memory=1G --conf spark.executor.cores=1 --conf spark.driver.cores=1"
-  # configurationOverrides:
-  #   applicationConfiguration:
-  #   monitoringConfiguration:
-  #     cloudWatchMonitoringConfiguration:
-  #       logGroupName:
-  #       logStreamNamePrefix:
-  #     s3MonitoringConfiguration:
-  #       logUri:
+  configurationOverrides: |
+    ApplicationConfiguration: null
+    MonitoringConfiguration:
+      PersistentAppUI: ENABLED
+      S3MonitoringConfiguration:
+        LogUri: s3://$EMREKSS3BucketName/logs
diff --git a/test/e2e/service_bootstrap.py b/test/e2e/service_bootstrap.py
@@ -19,6 +19,7 @@
 
 from acktest.bootstrapping import Resources, BootstrapFailureException
 from acktest.bootstrapping.iam import Role, UserPolicies
+from acktest.bootstrapping.s3 import Bucket
 from e2e import bootstrap_directory
 from e2e.bootstrap_resources import BootstrapResources
 from e2e.bootstrappable.emr_eks_cluster import EMREnabledEKSCluster
@@ -63,6 +64,7 @@ def service_bootstrap() -> Resources:
         JobExecutionRole=Role("ack-emrcontainers-job-execution-role", "ec2.amazonaws.com",
             user_policies=UserPolicies("ack-emrcontainers-job-execution-policy", [job_execution_policy])
         ),
+        EMREKSS3BucketName=Bucket("ack-emr-eks-logs"),
         HostCluster_VC=EMREnabledEKSCluster("ack-emr-eks", "emr-ns"),
         HostCluster_JR=EMREnabledEKSCluster("ack-emr-eks", "emr-ns")
     )
diff --git a/test/e2e/tests/test_jobrun.py b/test/e2e/tests/test_jobrun.py
@@ -71,12 +71,14 @@ def jobrun():
     emr_release_label = "emr-6.3.0-latest"
     eks_clustername = get_bootstrap_resources().HostCluster_JR.cluster.name
     job_execution_role = get_bootstrap_resources().JobExecutionRole.arn
+    emr_eks_s3_logs_name = get_bootstrap_resources().EMREKSS3BucketName.name
 
     replacements = REPLACEMENT_VALUES.copy()
     replacements["JOBRUN_NAME"] = job_run_name
-    replacements["VIRTUALCLUSTER_ID"] = virtual_cluster_id
+    replacements["VIRTUALCLUSTER_NAME"] = virtual_cluster_name
     replacements["EMR_RELEASE_LABEL"] = emr_release_label
     replacements["JOB_EXECUTION_ROLE"] = job_execution_role
+    replacements["EMREKSS3BucketName"] = emr_eks_s3_logs_name
 
     resource_data = load_resource(
         "job_run",
