ability to specify dataproc version in BBC

src/main/resources/templates/create-cluster.ssp

@@ -12,7 +12,8 @@
     cluster_name='${cluster.name}'.lower(),
     num_workers=${ evalObject.evaluateVariable(cluster.cluster.workers,localVariables)},
     zone='${evalObject.evaluateVariable(programConfiguration.zone,localVariables )}',
-    #if(cluster.cluster.subNetWork.nonEmpty)subnetwork_uri='${evalObject.evaluateVariable(cluster.cluster.subNetWork.get,localVariables)}',
+    #if(cluster.cluster.version.nonEmpty)image_version='${evalObject.evaluateVariable(cluster.cluster.version.get,localVariables)}',
+    #end#if(cluster.cluster.subNetWork.nonEmpty)subnetwork_uri='${evalObject.evaluateVariable(cluster.cluster.subNetWork.get,localVariables)}',
     #end#if(cluster.cluster.initialScript.nonEmpty)init_actions_uris=['${evalObject.evaluateVariable(cluster.cluster.initialScript.get,localVariables)}'],
     #end#if(cluster.cluster.initialActionTimeout.nonEmpty)init_action_timeout='${evalObject.evaluateVariable(cluster.cluster.initialActionTimeout.get,localVariables)}',
     #end#if(hasTags)tags=[${cluster.cluster.tags.get.map(f=> "'"+evalObject.evaluateVariable(f,localVariables )+"'").mkString(",")}],

src/main/scala/com/recipegrace/bbc/grmr/BBCGrammar.scala

@@ -51,6 +51,7 @@ trait BBCGrammar  extends BaseGrammar{
       ClusterConfigWorkers( NumberExpression( f._2.toInt))
     })) |
       (IMAGE ~ EQUAL ~ _expression) ^^ (f => ClusterConfigImage(f._2)) |
+      (DATAPROCVERSION ~ EQUAL ~ _expression) ^^ (f => ClusterConfigDataProcVersion(f._2)) |
       TAGS ~ EQUAL ~ repsep(_expression, COMMA) ^^ (f => {
         val stripped = f._2.map(f => f).toArray
         ClusterConfigTags(stripped)

src/main/scala/com/recipegrace/bbc/grmr/BBCStructures.scala

@@ -65,6 +65,7 @@ object BBCStructures {
 
   case class ClusterConfigWorkers(workers: Expression) extends ClusterConfig
   case class ClusterConfigImage(image: Expression) extends ClusterConfig
+  case class ClusterConfigDataProcVersion(version:Expression) extends ClusterConfig
   case class ClusterConfigSubNetwork(subNetwork: Expression) extends ClusterConfig
   case class ClusterConfigTags(tags: Array[Expression]) extends ClusterConfig
   case class ClusterConfigProperties(props: List[Expression]) extends ClusterConfig
@@ -73,7 +74,7 @@ object BBCStructures {
   case class ClusterConfigInitialScriptTimeOut(timeOut: Expression) extends ClusterConfig
 
 
-  case class ClusterWrapper(workers:Expression,image:Expression,initialScript:Option[Expression],initialActionTimeout:Option[Expression], subNetWork:Option[Expression], tags:Option[Array[Expression]])
+  case class ClusterWrapper(workers:Expression,image:Expression,initialScript:Option[Expression],initialActionTimeout:Option[Expression], subNetWork:Option[Expression], tags:Option[Array[Expression]], version:Option[Expression])
   case class Cluster(id: Int, variableName: String, clusterConfigs: List[ClusterConfig]) extends Declaration {
     val name = "BB-"+variableName + rand
     lazy val cluster = calculateWrapper()
@@ -94,12 +95,15 @@ object BBCStructures {
         case x: ClusterConfigInitialScript =>INITIALSCRIPT -> x.script
         case x: ClusterConfigSubNetwork =>SUBNETWORK -> x.subNetwork
         case x: ClusterConfigInitialScriptTimeOut =>INITIALSCRIPTTIMEOUT -> x.timeOut
+        case x: ClusterConfigDataProcVersion => DATAPROCVERSION -> x.version
         case _ => (System.currentTimeMillis()+"") -> StringExpression(System.currentTimeMillis()+"" )
 
       }.toMap
     def calculateWrapper() = {
 
-      new ClusterWrapper(configMap(NUMWORKERS), configMap(IMAGE), configMap.get(INITIALSCRIPT),configMap.get(INITIALSCRIPTTIMEOUT), configMap.get(SUBNETWORK), tagMap.get(TAGS))
+      new ClusterWrapper(configMap(NUMWORKERS), configMap(IMAGE),
+        configMap.get(INITIALSCRIPT),configMap.get(INITIALSCRIPTTIMEOUT),
+        configMap.get(SUBNETWORK), tagMap.get(TAGS), configMap.get(DATAPROCVERSION))
     }
 
   }

src/main/scala/com/recipegrace/bbc/grmr/GrammarKeywords.scala

@@ -12,6 +12,7 @@ trait GrammarKeywords {
   val IMAGE = "image"
   val TAGS = "tags"
   val INITIALSCRIPT = "initialScript"
+  val DATAPROCVERSION = "dataprocVersion"
   val VERSION = "version"
   val INITIALSCRIPTTIMEOUT = "initialScriptTimeOut"
   val CONTAINER = "container"

src/test/scala/com/recipegrace/bbc/composer/TemplateTest.scala

@@ -264,6 +264,32 @@ class TemplateTest extends BaseBBCGrammarTest with ExpressionCreator{
 
   }
 
+  test("template test for  create cluster with dataproc version") {
+    val cluster = Cluster(1,"big-bricks",List(ClusterConfigImage(expr("image")),
+      ClusterConfigDataProcVersion(expr("1.3.19-deb9")),ClusterConfigWorkers(expr("2"))))
+    val programConfiguration = ProgramConfiguration(expr("one"),expr("zone"),"name",None)
+    val localVariables = Map():Map[String, Expression]
+    object evalObject extends ExpressionCreator
+    Templates.translate("templates/create-cluster.ssp",Map("name"-> taskName,
+      "localVariables" -> localVariables,
+      "evalObject" -> evalObject,
+      "programConfiguration" -> programConfiguration,
+      "cluster" -> cluster )) shouldBe "\n"+s"""  ${taskName} = DataprocClusterCreateOperatorModified(
+                                               |    task_id='create_cluster_$taskName',
+                                               |    # Give the cluster a unique name by appending the date scheduled.
+                                               |    # See https://airflow.apache.org/code.html#default-variables
+                                               |    cluster_name='${cluster.name}'.lower(),
+                                               |    num_workers=${ evalObject.evaluateVariable(cluster.cluster.workers,localVariables)},
+                                               |    zone='${evalObject.evaluateVariable(programConfiguration.zone,localVariables )}',
+                                               |    image_version='${evalObject.evaluateVariable(cluster.cluster.version.get, localVariables)}',
+                                               |    master_machine_type='${evalObject.evaluateVariable(cluster.cluster.image,localVariables)}',
+                                               |    worker_machine_type='${evalObject.evaluateVariable(cluster.cluster.image,localVariables)}'
+                                               |  )""".stripMargin
+
+
+  }
+
+
 
 
 

src/test/scala/com/recipegrace/bbc/grmr/BBCClusterTest.scala

@@ -81,4 +81,35 @@ class BBCClusterTest extends BaseBBCGrammarTest {
       case ClusterConfigProperties(List(StringExpression(x))) => x shouldBe properties
     }
   }
+
+  test("cluster  with dataproc test") {
+    val clusterName = "datacluster"
+    val workers = 0
+    val image = "sdd-hs-12"
+    val properties = "properties"
+    val dataprocVersion = "DATAPROCVERSION"
+
+    val clusterBlock =
+      s"""
+      cluster ${clusterName} {
+      dataprocVersion = "$dataprocVersion"
+      workers=${workers}
+      image="${image}"
+      properties="${properties}"
+      }
+    """
+
+
+
+
+    val cluster = parseAll(_clusterBody, clusterBlock).get
+
+
+    cluster._2.clusterConfigs.foreach {
+      case ClusterConfigWorkers(NumberExpression(x)) => x shouldBe workers
+      case ClusterConfigImage(StringExpression(x)) => x shouldBe image
+      case ClusterConfigProperties(List(StringExpression(x))) => x shouldBe properties
+      case ClusterConfigDataProcVersion(StringExpression(text)) => text shouldBe dataprocVersion
+    }
+  }
 }