[WIP] Armada integration

assembly/pom.xml

@@ -237,6 +237,16 @@
         </dependency>
       </dependencies>
     </profile>
+    <profile>
+      <id>armada</id>
+      <dependencies>
+        <dependency>
+          <groupId>io.armadaproject.spark</groupId>
+          <artifactId>spark-armada_${scala.binary.version}</artifactId>
+          <version>${project.version}</version>
+        </dependency>
+      </dependencies>
+    </profile>
     <profile>
       <id>hive</id>
       <dependencies>

core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala

@@ -260,9 +260,10 @@ private[spark] class SparkSubmit extends Logging {
           case "yarn" => YARN
           case m if m.startsWith("spark") => STANDALONE
           case m if m.startsWith("k8s") => KUBERNETES
+          case m if m.startsWith("armada") => ARMADA
           case m if m.startsWith("local") => LOCAL
           case _ =>
-            error("Master must either be yarn or start with spark, k8s, or local")
+            error("Master must either be yarn or start with spark, k8s, armada, or local")
             -1
         }
       case None => LOCAL // default master or remote mode.
@@ -296,6 +297,14 @@ private[spark] class SparkSubmit extends Logging {
       }
     }
 
+    if (clusterManager == ARMADA) {
+      if (!Utils.classIsLoadable(ARMADA_CLUSTER_SUBMIT_CLASS) && !Utils.isTesting) {
+        error(
+          s"Could not load ARMADA classes. \"${ARMADA_CLUSTER_SUBMIT_CLASS}\"" +
+          "This copy of Spark may not have been compiled with ARMADA support.")
+      }
+    }
+
     // Fail fast, the following modes are not supported or applicable
     (clusterManager, deployMode) match {
       case (STANDALONE, CLUSTER) if args.isPython =>
@@ -329,6 +338,8 @@ private[spark] class SparkSubmit extends Logging {
     val isKubernetesClient = clusterManager == KUBERNETES && deployMode == CLIENT
     val isKubernetesClusterModeDriver = isKubernetesClient &&
       sparkConf.getBoolean("spark.kubernetes.submitInDriver", false)
+    val isArmadaCluster = clusterManager == ARMADA && deployMode == CLUSTER
+    // TODO: Support armada & client?
     val isCustomClasspathInClusterModeDisallowed =
       !sparkConf.get(ALLOW_CUSTOM_CLASSPATH_BY_PROXY_USER_IN_CLUSTER_MODE) &&
       args.proxyUser != null &&
@@ -416,6 +427,7 @@ private[spark] class SparkSubmit extends Logging {
         downloadFileList(_, targetDir, sparkConf, hadoopConf)
       }.orNull
 
+      // TODO: May have to do the same/similar for Armada
       if (isKubernetesClusterModeDriver) {
         // SPARK-33748: this mimics the behaviour of Yarn cluster mode. If the driver is running
         // in cluster mode, the archives should be available in the driver's current working
@@ -670,6 +682,7 @@ private[spark] class SparkSubmit extends Logging {
         confKey = KEYTAB.key),
       OptionAssigner(args.pyFiles, ALL_CLUSTER_MGRS, CLUSTER, confKey = SUBMIT_PYTHON_FILES.key),
 
+      // TODO: Add Armada where appropriate.
       // Propagate attributes for dependency resolution at the driver side
       OptionAssigner(args.packages, STANDALONE | KUBERNETES,
         CLUSTER, confKey = JAR_PACKAGES.key),
@@ -864,6 +877,11 @@ private[spark] class SparkSubmit extends Logging {
       }
     }
 
+    if (isArmadaCluster) {
+      childMainClass = ARMADA_CLUSTER_SUBMIT_CLASS
+      // TODO: Setup childArgs
+    }
+
     // Load any properties specified through --conf and the default properties file
     for ((k, v) <- args.sparkProperties) {
       sparkConf.setIfMissing(k, v)
@@ -1071,7 +1089,8 @@ object SparkSubmit extends CommandLineUtils with Logging {
   private val STANDALONE = 2
   private val LOCAL = 8
   private val KUBERNETES = 16
-  private val ALL_CLUSTER_MGRS = YARN | STANDALONE | LOCAL | KUBERNETES
+  private val ARMADA = 32
+  private val ALL_CLUSTER_MGRS = YARN | STANDALONE | LOCAL | KUBERNETES | ARMADA
 
   // Deploy modes
   private val CLIENT = 1
@@ -1095,6 +1114,8 @@ object SparkSubmit extends CommandLineUtils with Logging {
   private[deploy] val STANDALONE_CLUSTER_SUBMIT_CLASS = classOf[ClientApp].getName()
   private[deploy] val KUBERNETES_CLUSTER_SUBMIT_CLASS =
     "org.apache.spark.deploy.k8s.submit.KubernetesClientApplication"
+  private[deploy] val ARMADA_CLUSTER_SUBMIT_CLASS =
+    "org.apache.spark.deploy.armada.submit.ArmadaClientApplication"
 
   override def main(args: Array[String]): Unit = {
     Option(System.getenv("SPARK_PREFER_IPV6"))

core/src/main/scala/org/apache/spark/deploy/SparkSubmitArguments.scala

@@ -515,7 +515,8 @@ private[deploy] class SparkSubmitArguments(args: Seq[String], env: Map[String, S
       s"""
         |Options:
         |  --master MASTER_URL         spark://host:port, yarn,
-        |                              k8s://https://host:port, or local (Default: local[*]).
+        |                              k8s://https://host:port, armada://host:port,
+        |                              or local (Default: local[*]).
         |  --deploy-mode DEPLOY_MODE   Whether to launch the driver program locally ("client") or
         |                              on one of the worker machines inside the cluster ("cluster")
         |                              (Default: client).

core/src/test/scala/org/apache/spark/status/AppStatusListenerSuite.scala

@@ -279,12 +279,10 @@ abstract class AppStatusListenerSuite extends SparkFunSuite with BeforeAndAfter
     executorStageSummaryWrappers.foreach { exec =>
       // only the first executor is expected to be excluded
       val expectedExcludedFlag = exec.executorId == execIds.head
-      assert(exec.info.isBlacklistedForStage === expectedExcludedFlag)
       assert(exec.info.isExcludedForStage === expectedExcludedFlag)
     }
 
     check[ExecutorSummaryWrapper](execIds.head) { exec =>
-      assert(exec.info.blacklistedInStages === Set(stages.head.stageId))
       assert(exec.info.excludedInStages === Set(stages.head.stageId))
 
     }
@@ -306,7 +304,6 @@ abstract class AppStatusListenerSuite extends SparkFunSuite with BeforeAndAfter
     assert(executorStageSummaryWrappersForNode.nonEmpty)
     executorStageSummaryWrappersForNode.foreach { exec =>
       // both executor is expected to be excluded
-      assert(exec.info.isBlacklistedForStage)
       assert(exec.info.isExcludedForStage)
 
     }
@@ -467,7 +464,6 @@ abstract class AppStatusListenerSuite extends SparkFunSuite with BeforeAndAfter
     }
 
     check[ExecutorSummaryWrapper](execIds.head) { exec =>
-      assert(exec.info.blacklistedInStages === Set())
       assert(exec.info.excludedInStages === Set())
     }
 
@@ -495,7 +491,6 @@ abstract class AppStatusListenerSuite extends SparkFunSuite with BeforeAndAfter
       stageAttemptId = stages.last.attemptNumber()))
 
     check[ExecutorSummaryWrapper](execIds.head) { exec =>
-      assert(exec.info.blacklistedInStages === Set(stages.last.stageId))
       assert(exec.info.excludedInStages === Set(stages.last.stageId))
     }
 
@@ -652,29 +647,25 @@ abstract class AppStatusListenerSuite extends SparkFunSuite with BeforeAndAfter
     time += 1
     listener.onExecutorExcluded(SparkListenerExecutorExcluded(time, "1", 42))
     check[ExecutorSummaryWrapper]("1") { exec =>
-      assert(exec.info.isBlacklisted)
       assert(exec.info.isExcluded)
     }
 
     time += 1
     listener.onExecutorUnexcluded(SparkListenerExecutorUnexcluded(time, "1"))
     check[ExecutorSummaryWrapper]("1") { exec =>
-      assert(!exec.info.isBlacklisted)
       assert(!exec.info.isExcluded)
     }
 
     // Exclude a node.
     time += 1
     listener.onNodeExcluded(SparkListenerNodeExcluded(time, "1.example.com", 2))
     check[ExecutorSummaryWrapper]("1") { exec =>
-      assert(exec.info.isBlacklisted)
       assert(exec.info.isExcluded)
     }
 
     time += 1
     listener.onNodeUnexcluded(SparkListenerNodeUnexcluded(time, "1.example.com"))
     check[ExecutorSummaryWrapper]("1") { exec =>
-      assert(!exec.info.isBlacklisted)
       assert(!exec.info.isExcluded)
     }
 

pom.xml

@@ -3409,6 +3409,14 @@
       </modules>
     </profile>
 
+    <profile>
+      <id>armada</id>
+      <modules>
+        <module>resource-managers/armada/core</module>
+      </modules>
+    </profile>
+
+
     <profile>
       <id>hive-thriftserver</id>
       <modules>

resource-managers/armada/core/README.md

@@ -0,0 +1,30 @@
+Armada Cluster Manager for Spark
+---
+
+Build
+---
+From the spark repository root: 
+```bash
+$ JAVA_HOME=/usr/lib/jvm/java/ ./build/sbt -Parmada
+```
+You may have to adjust `JAVA_HOME` to suit your environment. `-Parmada` tells
+sbt to enable the armada project.
+
+Once at the sbt console:
+```
+sbt:spark-parent> project armada
+```
+Switches to the aramda project.
+
+Now with feeling:
+```
+sbt:spark-armada> package
+```
+Should build what's available underneath `spark/resource-managers/armada`.
+
+Test
+---
+```
+sbt:spark-armada> test
+```
+Runs all tests!