Merge pull request #30 from civitaspo/develop

v0.0.6

CHANGELOG.md

@@ -1,3 +1,8 @@
+0.0.6 (2018-11-13)
+==================
+
+* [Enhancement] Enable to use params as env for `ecs_task.sh`
+
 0.0.5 (2018-11-13)
 ==================
 

README.md

@@ -15,7 +15,7 @@ _export:
     repositories:
       - https://jitpack.io
     dependencies:
-      - pro.civitaspo:digdag-operator-ecs_task:0.0.5
+      - pro.civitaspo:digdag-operator-ecs_task:0.0.6
   ecs_task:
     auth_method: profile
 

build.gradle

@@ -6,7 +6,7 @@ plugins {
 }
 
 group = 'pro.civitaspo'
-version = '0.0.5'
+version = '0.0.6'
 
 def digdagVersion = '0.9.31'
 def scalaSemanticVersion = "2.12.6"
@@ -37,6 +37,7 @@ shadowJar {
     classifier = null
     dependencies {
         exclude(dependency('io.digdag:.*'))
+        exclude(dependency('.*:jackson.*:.*'))
     }
 }
 

example/example.dig

@@ -4,7 +4,7 @@ _export:
       - file://${repos}
       # - https://jitpack.io
     dependencies:
-      - pro.civitaspo:digdag-operator-ecs_task:0.0.5
+      - pro.civitaspo:digdag-operator-ecs_task:0.0.6
   ecs_task:
     auth_method: profile
 

src/main/resources/pro/civitaspo/digdag/plugin/ecs_task/sh/run.sh

@@ -16,6 +16,9 @@ cd workspace
 # Unset e option for returning embulk results to digdag
 set +e
 
+# envs
+export ${ECS_TASK_SH_EXPORT_ENV}
+
 # Run
 ${ECS_TASK_SH_COMMAND} \
         2>> ../stderr.log \

src/main/scala/pro/civitaspo/digdag/plugin/ecs_task/command/AbstractEcsTaskCommandOperator.scala

@@ -8,10 +8,13 @@ import pro.civitaspo.digdag.plugin.ecs_task.util.TryWithResource
 
 import scala.collection.JavaConverters._
 import scala.util.Random
+import scala.util.matching.Regex
 
 abstract class AbstractEcsTaskCommandOperator(operatorName: String, context: OperatorContext, systemConfig: Config, templateEngine: TemplateEngine)
     extends AbstractEcsTaskOperator(operatorName, context, systemConfig, templateEngine) {
 
+  protected val validEnvKeyRegex: Regex = "[a-zA-Z_][a-zA-Z_0-9]*".r
+
   protected val mainScriptName: String
 
   private lazy val tmpStorageConfig: Config = {
@@ -53,7 +56,20 @@ abstract class AbstractEcsTaskCommandOperator(operatorName: String, context: Ope
   protected def collectEnvironments(): Map[String, String] = {
     val vars: PrivilegedVariables = context.getPrivilegedVariables
     vars.getKeys.asScala.foldLeft(Map.empty[String, String]) { (env, key) =>
-      env ++ Map(key -> vars.get(key))
+    if (isValidEnvKey(key)) {
+        env ++ Map(key -> vars.get(key))
+      }
+      else {
+        logger.info(s"$key is invalid env key.")
+        env
+      }
+    }
+  }
+
+  protected def isValidEnvKey(key: String): Boolean = {
+    key match {
+      case validEnvKeyRegex() => true
+      case _ => false
     }
   }
 

src/main/scala/pro/civitaspo/digdag/plugin/ecs_task/package.scala

@@ -2,6 +2,6 @@ package pro.civitaspo.digdag.plugin
 
 package object ecs_task {
 
-  val VERSION: String = "0.0.5"
+  val VERSION: String = "0.0.6"
 
 }

src/main/scala/pro/civitaspo/digdag/plugin/ecs_task/sh/EcsTaskShOperotar.scala

@@ -1,11 +1,14 @@
 package pro.civitaspo.digdag.plugin.ecs_task.sh
+import com.fasterxml.jackson.databind.JsonNode
 import io.digdag.client.config.Config
 import io.digdag.spi.{OperatorContext, TemplateEngine}
 import pro.civitaspo.digdag.plugin.ecs_task.aws.AmazonS3UriWrapper
 import pro.civitaspo.digdag.plugin.ecs_task.command.{AbstractEcsTaskCommandOperator, TmpStorage}
 import pro.civitaspo.digdag.plugin.ecs_task.util.TryWithResource
 
+import scala.collection.JavaConverters._
 import scala.io.Source
+import scala.util.Try
 
 class EcsTaskShOperotar(operatorName: String, context: OperatorContext, systemConfig: Config, templateEngine: TemplateEngine)
     extends AbstractEcsTaskCommandOperator(operatorName, context, systemConfig, templateEngine) {
@@ -25,11 +28,29 @@ class EcsTaskShOperotar(operatorName: String, context: OperatorContext, systemCo
     val dup: Config = params.deepCopy()
     dup.set("ECS_TASK_SH_BUCKET", AmazonS3UriWrapper(tmpStorage.getLocation).getBucket)
     dup.set("ECS_TASK_SH_PREFIX", AmazonS3UriWrapper(tmpStorage.getLocation).getKey)
+    dup.set("ECS_TASK_SH_EXPORT_ENV", convertParamsAsEnv().map { case (k: String, v: String) => s"$k=$v" }.mkString(" "))
     dup.set("ECS_TASK_SH_COMMAND", command)
 
     TryWithResource(classOf[EcsTaskShOperotar].getResourceAsStream(runShResourcePath)) { is =>
       val runShContentTemplate: String = Source.fromInputStream(is).mkString
       templateEngine.template(runShContentTemplate, dup)
     }
   }
+
+  protected def convertParamsAsEnv(params: Config = params): Map[String, String] = {
+    val keys: Seq[String] = params.getKeys.asScala
+    keys.foldLeft(Map.empty[String, String]) { (env, key) =>
+      if (isValidEnvKey(key)) {
+        val jn: JsonNode = params.getInternalObjectNode.get(key)
+        val v: String =
+          if (jn.isTextual) s""""${jn.textValue().replace("\"", "\\\"")}""""
+          else jn.toString
+        env ++ Map(key -> v)
+      }
+      else {
+        logger.info(s"$key is invalid env key.")
+        env
+      }
+    }
+  }
 }