Feature/467 poc emr submitter (#493)

* Refactor submitJob out in a separate service

* Refactor SparkYarnSinkConfig to SparkConfig

* Add SparkEmrClusterService

* Add custom validation for spark config

* Small refactorings

* Add support for local aws profile

* Fixes

* Add config validation

* Add test

* Fixes after merge

Author: kevinwallimann

pom.xml

@@ -121,6 +121,7 @@
         <snakeyaml.version>1.26</snakeyaml.version>
         <commons-validator.version>1.7</commons-validator.version>
         <hibernate.validator.version>6.2.0.Final</hibernate.validator.version>
+        <aws.java.sdk.bom.version>1.12.29</aws.java.sdk.bom.version>
 
         <!-- Deployment -->
         <gpg.plugin.version>1.6</gpg.plugin.version>
@@ -133,6 +134,18 @@
         <skip.postgres.tests>true</skip.postgres.tests>
     </properties>
 
+    <dependencyManagement>
+        <dependencies>
+            <dependency>
+                <groupId>com.amazonaws</groupId>
+                <artifactId>aws-java-sdk-bom</artifactId>
+                <version>${aws.java.sdk.bom.version}</version>
+                <type>pom</type>
+                <scope>import</scope>
+            </dependency>
+        </dependencies>
+    </dependencyManagement>
+
     <dependencies>
         <dependency>
             <groupId>com.fasterxml.jackson.module</groupId>
@@ -250,6 +263,10 @@
             <artifactId>hibernate-validator</artifactId>
             <version>${hibernate.validator.version}</version>
         </dependency>
+        <dependency>
+            <groupId>com.amazonaws</groupId>
+            <artifactId>aws-java-sdk-emr</artifactId>
+        </dependency>
 
         <!-- For Liquibase yaml -->
         <dependency>

src/main/resources/application.properties

@@ -88,7 +88,9 @@ kafkaSource.properties.sasl.jaas.config=
 shellExecutor.executablesFolder=/
 
 #Spark yarn sink properties. Properties used to deploy and run Spark job in Yarn.
+spark.submitApi=yarn
 sparkYarnSink.hadoopResourceManagerUrlBase=http://localhost:8088
+sparkYarnSink.userUsedToKillJob=
 sparkYarnSink.hadoopConfDir=/opt/hadoop
 sparkYarnSink.sparkHome=/opt/spark
 sparkYarnSink.master=yarn
@@ -106,7 +108,12 @@ sparkYarnSink.additionalConfs.spark.yarn.principal=
 sparkYarnSink.additionalConfs.spark.shuffle.service.enabled=true
 sparkYarnSink.additionalConfs.spark.dynamicAllocation.enabled=true
 sparkYarnSink.executablesFolder=/
-sparkYarnSink.userUsedToKillJob=
+
+spark.emr.clusterId=
+spark.emr.filesToDeploy=
+spark.emr.additionalConfs=
+spark.emr.awsProfile=
+spark.emr.region=
 
 #Postgresql properties for connection to trigger metastore
 db.driver=net.bull.javamelody.JdbcDriver

src/main/scala/za/co/absa/hyperdrive/trigger/api/rest/controllers/AppInfoController.scala

@@ -16,16 +16,16 @@
 package za.co.absa.hyperdrive.trigger.api.rest.controllers
 
 import org.springframework.web.bind.annotation._
-import za.co.absa.hyperdrive.trigger.configuration.application.{GeneralConfig, SparkYarnSinkConfig}
+import za.co.absa.hyperdrive.trigger.configuration.application.{GeneralConfig, SparkConfig}
 import za.co.absa.hyperdrive.trigger.models.AppInfo
 
 import javax.inject.Inject
 
 @RestController 
-class AppInfoController @Inject()(sparkYarnSinkConfig: SparkYarnSinkConfig, generalConfig: GeneralConfig) {
+class AppInfoController @Inject()(sparkConfig: SparkConfig, generalConfig: GeneralConfig) {
   val environment: String = generalConfig.environment
   val version: String = generalConfig.version
-  val resourceManagerUrl: String = sparkYarnSinkConfig.hadoopResourceManagerUrlBase
+  val resourceManagerUrl: String = sparkConfig.hadoopResourceManagerUrlBase
 
   @GetMapping(path = Array("/app/info"))
   def appInfo(): AppInfo = {

src/main/scala/za/co/absa/hyperdrive/trigger/api/rest/health/YarnConnectionHealthIndicator.scala

@@ -18,18 +18,18 @@ package za.co.absa.hyperdrive.trigger.api.rest.health
 
 import org.springframework.boot.actuate.health.{Health, HealthIndicator}
 import org.springframework.stereotype.Component
-import za.co.absa.hyperdrive.trigger.configuration.application.{HealthConfig, SparkYarnSinkConfig}
+import za.co.absa.hyperdrive.trigger.configuration.application.{HealthConfig, SparkConfig}
 
 import java.net.{HttpURLConnection, MalformedURLException, URL}
 import javax.inject.Inject
 import scala.util.{Failure, Success, Try}
 
 @Component
-class YarnConnectionHealthIndicator @Inject()(sparkYarnSinkConfig: SparkYarnSinkConfig, healthConfig: HealthConfig) extends HealthIndicator {
+class YarnConnectionHealthIndicator @Inject()(sparkConfig: SparkConfig, healthConfig: HealthConfig) extends HealthIndicator {
   val successCode = 200
 
   override protected def health(): Health = {
-    val yarnBaseUrl = sparkYarnSinkConfig.hadoopResourceManagerUrlBase.stripSuffix("/")
+    val yarnBaseUrl = sparkConfig.hadoopResourceManagerUrlBase.stripSuffix("/")
     val yarnTestEndpoint = healthConfig.yarnConnectionTestEndpoint.stripPrefix("/")
 
     Try(new URL(s"$yarnBaseUrl/$yarnTestEndpoint")).flatMap(url =>

src/main/scala/za/co/absa/hyperdrive/trigger/api/rest/services/JobInstanceService.scala

@@ -21,7 +21,7 @@ import za.co.absa.hyperdrive.trigger.api.rest.utils.WSClientProvider
 import za.co.absa.hyperdrive.trigger.models.JobInstance
 import za.co.absa.hyperdrive.trigger.persistance.JobInstanceRepository
 import play.api.libs.ws.JsonBodyWritables.writeableOf_JsValue
-import za.co.absa.hyperdrive.trigger.configuration.application.SparkYarnSinkConfig
+import za.co.absa.hyperdrive.trigger.configuration.application.SparkConfig
 
 import scala.concurrent.{ExecutionContext, Future}
 import scala.util.Try
@@ -34,15 +34,15 @@ trait JobInstanceService {
 
 @Service
 class JobInstanceServiceImpl(override val jobInstanceRepository: JobInstanceRepository,
-                             sparkYarnSinkConfig: SparkYarnSinkConfig) extends JobInstanceService {
+                             sparkConfig: SparkConfig) extends JobInstanceService {
 
   override def getJobInstances(dagInstanceId: Long)(implicit ec: ExecutionContext): Future[Seq[JobInstance]] = {
     jobInstanceRepository.getJobInstances(dagInstanceId)
   }
 
   override def killJob(applicationId: String)(implicit ec: ExecutionContext): Future[Boolean] = {
     val url: String =
-      s"${sparkYarnSinkConfig.hadoopResourceManagerUrlBase}/ws/v1/cluster/apps/$applicationId/state?user.name=${sparkYarnSinkConfig.userUsedToKillJob}"
+      s"${sparkConfig.hadoopResourceManagerUrlBase}/ws/v1/cluster/apps/$applicationId/state?user.name=${sparkConfig.userUsedToKillJob}"
     val data = Json.obj(
       "state" -> "KILLED"
     )

src/main/scala/za/co/absa/hyperdrive/trigger/api/rest/services/JobTemplateResolutionService.scala

@@ -17,7 +17,7 @@ package za.co.absa.hyperdrive.trigger.api.rest.services
 
 import org.springframework.stereotype.Service
 import za.co.absa.hyperdrive.trigger.configuration.application.JobDefinitionConfig.{KeysToMerge, MergedValuesSeparator}
-import za.co.absa.hyperdrive.trigger.configuration.application.{ShellExecutorConfig, SparkYarnSinkConfig}
+import za.co.absa.hyperdrive.trigger.configuration.application.{ShellExecutorConfig, SparkConfig}
 import za.co.absa.hyperdrive.trigger.models._
 
 import java.nio.file.Paths
@@ -28,7 +28,7 @@ trait JobTemplateResolutionService {
 }
 
 @Service
-class JobTemplateResolutionServiceImpl(sparkYarnSinkConfig: SparkYarnSinkConfig, shellExecutorConfig: ShellExecutorConfig) extends JobTemplateResolutionService {
+class JobTemplateResolutionServiceImpl(sparkConfig: SparkConfig, shellExecutorConfig: ShellExecutorConfig) extends JobTemplateResolutionService {
   def resolveDagDefinitionJoined(dagDefinitionJoined: DagDefinitionJoined, jobTemplates: Seq[JobTemplate]): Seq[ResolvedJobDefinition] = {
     val jobTemplatesLookup = jobTemplates.map(t => t.id -> t).toMap
     dagDefinitionJoined.jobDefinitions.map(jd => {
@@ -63,22 +63,22 @@ class JobTemplateResolutionServiceImpl(sparkYarnSinkConfig: SparkYarnSinkConfig,
 
   private def mergeSparkAndHyperdriveParameters(definitionParams: HyperdriveDefinitionParameters, templateParams: SparkTemplateParameters): SparkInstanceParameters = {
     SparkInstanceParameters(
-      jobJar = Paths.get(sparkYarnSinkConfig.executablesFolder, templateParams.jobJar.getOrElse("")).toString,
+      jobJar = Paths.get(sparkConfig.yarn.executablesFolder, templateParams.jobJar.getOrElse("")).toString,
       mainClass = templateParams.mainClass.getOrElse(""),
       appArguments = mergeLists(definitionParams.appArguments, templateParams.appArguments),
-      additionalJars = mergeLists(definitionParams.additionalJars, templateParams.additionalJars).map(jar => Paths.get(sparkYarnSinkConfig.executablesFolder, jar).toString),
-      additionalFiles = mergeLists(definitionParams.additionalFiles, templateParams.additionalFiles).map(file => Paths.get(sparkYarnSinkConfig.executablesFolder, file).toString),
+      additionalJars = mergeLists(definitionParams.additionalJars, templateParams.additionalJars).map(jar => Paths.get(sparkConfig.yarn.executablesFolder, jar).toString),
+      additionalFiles = mergeLists(definitionParams.additionalFiles, templateParams.additionalFiles).map(file => Paths.get(sparkConfig.yarn.executablesFolder, file).toString),
       additionalSparkConfig = mergeMaps(definitionParams.additionalSparkConfig, templateParams.additionalSparkConfig, mergeSortedMapEntries)
     )
   }
 
   private def mergeSparkParameters(definitionParams: SparkDefinitionParameters, templateParams: SparkTemplateParameters): SparkInstanceParameters = {
     SparkInstanceParameters(
-      jobJar = Paths.get(sparkYarnSinkConfig.executablesFolder, mergeOptionStrings(definitionParams.jobJar, templateParams.jobJar)).toString,
+      jobJar = Paths.get(sparkConfig.yarn.executablesFolder, mergeOptionStrings(definitionParams.jobJar, templateParams.jobJar)).toString,
       mainClass = mergeOptionStrings(definitionParams.mainClass, templateParams.mainClass),
       appArguments = mergeLists(definitionParams.appArguments, templateParams.appArguments),
-      additionalJars = mergeLists(definitionParams.additionalJars, templateParams.additionalJars).map(jar => Paths.get(sparkYarnSinkConfig.executablesFolder, jar).toString),
-      additionalFiles = mergeLists(definitionParams.additionalFiles, templateParams.additionalFiles).map(file => Paths.get(sparkYarnSinkConfig.executablesFolder, file).toString),
+      additionalJars = mergeLists(definitionParams.additionalJars, templateParams.additionalJars).map(jar => Paths.get(sparkConfig.yarn.executablesFolder, jar).toString),
+      additionalFiles = mergeLists(definitionParams.additionalFiles, templateParams.additionalFiles).map(file => Paths.get(sparkConfig.yarn.executablesFolder, file).toString),
       additionalSparkConfig = mergeMaps(definitionParams.additionalSparkConfig, templateParams.additionalSparkConfig, mergeSortedMapEntries)
     )
   }

src/main/scala/za/co/absa/hyperdrive/trigger/configuration/application/AwsRegion.java

@@ -0,0 +1,37 @@
+
+/*
+ * Copyright 2018 ABSA Group Limited
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package za.co.absa.hyperdrive.trigger.configuration.application;
+
+import javax.validation.Constraint;
+import javax.validation.Payload;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+@Target({ ElementType.METHOD, ElementType.FIELD })
+@Retention(RetentionPolicy.RUNTIME)
+@Constraint(validatedBy = AwsRegionValidator.class)
+public @interface AwsRegion {
+
+    String message() default "{awsRegion.constraint}";
+
+    Class<?>[] groups() default {};
+
+    Class<? extends Payload>[] payload() default {};
+}
+

src/main/scala/za/co/absa/hyperdrive/trigger/scheduler/executors/Executor.scala renamed to src/main/scala/za/co/absa/hyperdrive/trigger/configuration/application/AwsRegionValidator.scala

@@ -1,3 +1,4 @@
+
 /*
  * Copyright 2018 ABSA Group Limited
  *
@@ -13,13 +14,14 @@
  * limitations under the License.
  */
 
-package za.co.absa.hyperdrive.trigger.scheduler.executors
-
-import za.co.absa.hyperdrive.trigger.models.{JobInstance, JobInstanceParameters}
+package za.co.absa.hyperdrive.trigger.configuration.application
 
-import scala.concurrent.{ExecutionContext, Future}
+import com.amazonaws.regions.Regions
+import javax.validation.{ConstraintValidator, ConstraintValidatorContext}
+import scala.util.Try
 
-trait Executor[T <: JobInstanceParameters] {
-  def execute(jobInstance: JobInstance, jobParameters: T, updateJob: JobInstance => Future[Unit])
-             (implicit executionContext: ExecutionContext, executorConfig: ExecutorConfig): Future[Unit]
+class AwsRegionValidator extends ConstraintValidator[AwsRegion, String]{
+  override def isValid(awsRegion: String, constraintValidatorContext: ConstraintValidatorContext): Boolean = {
+    awsRegion == null || awsRegion.isEmpty || Try(Regions.fromName(awsRegion)).isSuccess
+  }
 }

src/main/scala/za/co/absa/hyperdrive/trigger/configuration/application/SparkConfig.scala

@@ -0,0 +1,110 @@
+
+/*
+ * Copyright 2018 ABSA Group Limited
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package za.co.absa.hyperdrive.trigger.configuration.application
+
+import org.springframework.boot.context.properties.bind.{DefaultValue, Name}
+import org.springframework.boot.context.properties.{ConfigurationProperties, ConstructorBinding}
+import org.springframework.validation.annotation.Validated
+import org.springframework.validation.{Errors, Validator}
+import za.co.absa.hyperdrive.trigger.configuration.application.SparkConfig.{toNonEmptyOption, transformAdditionalConfsProperty, transformFilesProperty}
+
+import java.util.Properties
+import javax.validation.constraints.{NotBlank, NotNull}
+import scala.annotation.meta.field
+
+@ConfigurationProperties
+@ConstructorBinding
+@Validated
+class SparkConfig (
+  @DefaultValue(Array("yarn"))
+  @Name("spark.submitApi")
+  val submitApi: String,
+  @Name("sparkYarnSink")
+  val yarn: SparkYarnSinkConfig,
+  @Name("spark.emr")
+  val emr: SparkEmrSinkConfig,
+  @(NotBlank @field)
+  @Name("sparkYarnSink.hadoopResourceManagerUrlBase")
+  val hadoopResourceManagerUrlBase: String,
+  @DefaultValue(Array("Unknown"))
+  @Name("sparkYarnSink.userUsedToKillJob")
+  val userUsedToKillJob: String
+) {
+  if (submitApi == "yarn" && yarn == null) {
+    throw new RuntimeException("If spark.submitApi is yarn, sparkYarnSink arguments are required")
+  } else if (submitApi == "emr" && emr == null) {
+    throw new RuntimeException("If spark.submitApi is emr, spark.emr arguments are required")
+  } else if (submitApi != "yarn" && submitApi != "emr") {
+    throw new RuntimeException("spark.submitApi has to be either 'yarn' or 'emr'")
+  }
+}
+
+object SparkConfig {
+  def transformFilesProperty(filesToDeployInternal: String): Seq[String] = Option(filesToDeployInternal)
+    .map(_.split(",").toSeq)
+    .getOrElse(Seq())
+    .filter(_.nonEmpty)
+
+  def transformAdditionalConfsProperty(additionalConfsInternal: Properties): Map[String, String] = {
+    import scala.collection.JavaConverters._
+    Option(additionalConfsInternal)
+      .map(_.asScala.toMap).getOrElse(Map())
+  }
+
+  def toNonEmptyOption(string: String): Option[String] = {
+    Option(string).collect { case x if x.trim.nonEmpty => x}
+  }
+}
+
+class SparkYarnSinkConfig (
+  @NotNull
+  val submitTimeout: Int,
+  @(NotBlank @field)
+  val hadoopConfDir: String,
+  @(NotBlank @field)
+  val master: String,
+  @(NotBlank @field)
+  val sparkHome: String,
+  @Name("filesToDeploy")
+  filesToDeployInternal: String,
+  @Name("additionalConfs")
+  additionalConfsInternal: Properties,
+  @NotNull
+  val executablesFolder: String
+) {
+  val filesToDeploy: Seq[String] = transformFilesProperty(filesToDeployInternal)
+  val additionalConfs: Map[String, String] = transformAdditionalConfsProperty(additionalConfsInternal)
+}
+
+class SparkEmrSinkConfig (
+  @NotNull
+  val clusterId: String,
+  @Name("awsProfile")
+  awsProfileInternal: String,
+  @Name("region")
+  @(AwsRegion @field)(message = "Not a valid aws region")
+  val regionInternal: String,
+  @Name("filesToDeploy")
+  filesToDeployInternal: String,
+  @Name("additionalConfs")
+  additionalConfsInternal: Properties
+) {
+  val awsProfile: Option[String] = toNonEmptyOption(awsProfileInternal)
+  val region: Option[String] = toNonEmptyOption(regionInternal)
+  val filesToDeploy: Seq[String] = transformFilesProperty(filesToDeployInternal)
+  val additionalConfs: Map[String, String] = transformAdditionalConfsProperty(additionalConfsInternal)
+}