Cluster Remote Attestation Fix (#146)

The existing code only had RA working when run locally. This PR adds a sleep for 5 seconds to make sure that all executors are spun up successfully before attestation begins.

Closes #147

Author: octaviansima

README.md

@@ -62,6 +62,8 @@ After downloading the Opaque codebase, build and test it as follows.
 
 Next, run Apache Spark SQL queries with Opaque as follows, assuming [Spark 3.0](https://www.apache.org/dyn/closer.lua/spark/spark-3.0.1/spark-3.0.1-bin-hadoop2.7.tgz) (`wget http://apache.mirrors.pair.com/spark/spark-3.0.1/spark-3.0.1-bin-hadoop2.7.tgz`) is already installed:
 
+\* Opaque needs Spark's `'spark.executor.instances'` property to be set. This can be done in a custom config file, the default config file found at `/opt/spark/conf/spark-defaults.conf`, or as a `spark-submit` or `spark-shell` argument: `--conf 'spark.executor.instances=<value>`.
+
 1. Package Opaque into a JAR:
 
     ```sh

src/main/scala/edu/berkeley/cs/rise/opaque/RA.scala

@@ -22,29 +22,40 @@ import org.apache.spark.internal.Logging
 
 import edu.berkeley.cs.rise.opaque.execution.SP
 
-// Helper to handle remote attestation
-// 
+// Performs remote attestation for all executors
+// that have not been attested yet
 
 object RA extends Logging {
   def initRA(sc: SparkContext): Unit = {
 
-    val rdd = sc.makeRDD(Seq.fill(sc.defaultParallelism) { () })
+    // All executors need to be initialized before attestation can occur
+    var numExecutors = 1
+    if (!sc.isLocal) {
+      numExecutors = sc.getConf.getInt("spark.executor.instances", -1)
+      while (!sc.isLocal && sc.getExecutorMemoryStatus.size < numExecutors) {}
+    }
+
+    val rdd = sc.parallelize(Seq.fill(numExecutors) {()}, numExecutors)
     val intelCert = Utils.findResource("AttestationReportSigningCACert.pem")
     val sp = new SP()
 
     sp.Init(Utils.sharedKey, intelCert)
 
-    val msg1s = rdd.mapPartitionsWithIndex { (i, _) =>
+    // Runs on executors
+    val msg1s = rdd.mapPartitions { (_) =>
       val (enclave, eid) = Utils.initEnclave()
       val msg1 = enclave.GenerateReport(eid)
       Iterator((eid, msg1))
     }.collect.toMap
 
+    // Runs on driver
     val msg2s = msg1s.map{case (eid, msg1) => (eid, sp.ProcessEnclaveReport(msg1))}
 
-    val attestationResults = rdd.mapPartitionsWithIndex { (_, _) =>
+    // Runs on executors
+    val attestationResults = rdd.mapPartitions { (_) =>
       val (enclave, eid) = Utils.initEnclave()
-      enclave.FinishAttestation(eid, msg2s(eid))
+      val msg2 = msg2s(eid)
+      enclave.FinishAttestation(eid, msg2)
       Iterator((eid, true))
     }.collect.toMap