Feature/385 symlinks for files (#391)

* #385 symlinks for files

Author: jozefbakus

src/main/resources/application.properties

@@ -49,7 +49,8 @@ scheduler.executors.thread.pool.size=30
 scheduler.jobs.parallel.number=100
 scheduler.heart.beat=5000
 scheduler.lag.threshold=20000
-scheduler.sensors.changedSensorsMinimalChunkQuerySize=100
+scheduler.sensors.changedSensorsChunkQuerySize=100
+scheduler.executors.executablesFolder=/
 
 #Kafka sensor properties.
 kafkaSource.group.id=hyper_drive_${appUniqueId}

src/main/scala/za/co/absa/hyperdrive/trigger/persistance/SensorRepository.scala

@@ -53,7 +53,7 @@ class SensorRepositoryImpl extends SensorRepository {
 
   override def getChangedSensors(originalSensors: Seq[Sensor])(implicit ec: ExecutionContext): Future[Seq[Sensor]] = {
     Future.sequence(
-      originalSensors.grouped(SensorsConfig.getChangedSensorsMinimalChunkQuerySize).toSeq.map(group => getChangedSensorsInternal(group))
+      originalSensors.grouped(SensorsConfig.getChangedSensorsChunkQuerySize).toSeq.map(group => getChangedSensorsInternal(group))
     ).map(_.flatten)
   }
 

src/main/scala/za/co/absa/hyperdrive/trigger/scheduler/executors/shell/ShellParameters.scala

@@ -15,14 +15,17 @@
 
 package za.co.absa.hyperdrive.trigger.scheduler.executors.shell
 
+import java.nio.file.Paths
+
 import za.co.absa.hyperdrive.trigger.models.JobParameters
+import za.co.absa.hyperdrive.trigger.scheduler.utilities.ExecutorsConfig
 
 case class ShellParameters(
   scriptLocation: String
 )
 
 object ShellParameters {
   def apply(jobParameters: JobParameters): ShellParameters = new ShellParameters(
-    scriptLocation = jobParameters.variables("scriptLocation")
+    scriptLocation = Paths.get(ExecutorsConfig.getExecutablesFolder, jobParameters.variables("scriptLocation")).toString
   )
-}
+}

src/main/scala/za/co/absa/hyperdrive/trigger/scheduler/executors/spark/SparkParameters.scala

@@ -15,7 +15,10 @@
 
 package za.co.absa.hyperdrive.trigger.scheduler.executors.spark
 
+import java.nio.file.Paths
+
 import za.co.absa.hyperdrive.trigger.models.JobParameters
+import za.co.absa.hyperdrive.trigger.scheduler.utilities.ExecutorsConfig
 
 import scala.util.Try
 
@@ -35,12 +38,12 @@ case class SparkParameters(
 object SparkParameters {
   def apply(jobParameters: JobParameters): SparkParameters = {
     SparkParameters(
-      jobJar = jobParameters.variables("jobJar"),
+      jobJar = Paths.get(ExecutorsConfig.getExecutablesFolder, jobParameters.variables("jobJar")).toString,
       mainClass = jobParameters.variables("mainClass"),
       deploymentMode = jobParameters.variables("deploymentMode"),
       appArguments = Try(jobParameters.maps("appArguments")).getOrElse(List.empty[String]),
-      additionalJars = Try(jobParameters.maps("additionalJars")).getOrElse(List.empty[String]),
-      additionalFiles = Try(jobParameters.maps("additionalFiles")).getOrElse(List.empty[String]),
+      additionalJars = Try(jobParameters.maps("additionalJars")).getOrElse(List.empty[String]).map(jar => Paths.get(ExecutorsConfig.getExecutablesFolder, jar).toString),
+      additionalFiles = Try(jobParameters.maps("additionalFiles")).getOrElse(List.empty[String]).map(file => Paths.get(ExecutorsConfig.getExecutablesFolder, file).toString),
       additionalSparkConfig = Try(jobParameters.keyValuePairs("additionalSparkConfig")).getOrElse(Map.empty[String, String])
     )
   }

src/main/scala/za/co/absa/hyperdrive/trigger/scheduler/utilities/Configs.scala

@@ -69,8 +69,8 @@ object KafkaConfig {
 object SensorsConfig {
   val getThreadPoolSize: Int =
     Configs.conf.getInt("scheduler.sensors.thread.pool.size")
-  val getChangedSensorsMinimalChunkQuerySize: Int =
-    Configs.conf.getInt("scheduler.sensors.changedSensorsMinimalChunkQuerySize")
+  val getChangedSensorsChunkQuerySize: Int =
+    Configs.conf.getInt("scheduler.sensors.changedSensorsChunkQuerySize")
 }
 
 object SchedulerConfig {
@@ -85,6 +85,8 @@ object SchedulerConfig {
 object ExecutorsConfig {
   val getThreadPoolSize: Int =
     Configs.conf.getInt("scheduler.executors.thread.pool.size")
+  val getExecutablesFolder: String =
+    Configs.conf.getString("scheduler.executors.executablesFolder")
 }
 
 object SparkExecutorConfig {

src/test/resources/application.properties

@@ -20,7 +20,8 @@ auth.inmemory.password=password
 
 appUniqueId=9c282190-4078-4380-8960-ce52f43b94fg
 
-scheduler.sensors.changedSensorsMinimalChunkQuerySize=100
+scheduler.executors.executablesFolder=src/test/resources/
+scheduler.sensors.changedSensorsChunkQuerySize=100
 
 db.connectionPool=disabled
 db.url=jdbc:h2:mem:hyperdriver;DATABASE_TO_UPPER=false

src/test/scala/za/co/absa/hyperdrive/trigger/scheduler/executors/shell/ShellExecutorTest.scala

@@ -35,7 +35,7 @@ class ShellExecutorTest extends FlatSpec with Matchers with BeforeAndAfterAll wi
 
   private val updateJobStub: JobInstance => Future[Unit] = mock[JobInstance => Future[Unit]]
 
-  private val testScriptLocation = "src/test/resources/testShellScript.sh"
+  private val testScriptLocation = "testShellScript.sh"
   private val testJobInstance = JobInstance(
     jobName = "jobName",
     jobType = Shell,

ui/src/app/services/workflow/workflow.service.ts

@@ -333,7 +333,7 @@ export class WorkflowService {
   private static getShellFormParts(): FormPart[] {
     return [
       FormPartFactory.create(
-        'Script location',
+        'Script',
         'jobParameters.variables.scriptLocation',
         'string-field',
         PartValidationFactory.create(true, undefined, 1),