src/toil_lib/spark.py
def start(self, job):
        """
        Start spark and hdfs worker containers

        :param job: The underlying job.
        """

        # start spark and our datanode
        self.sparkContainerID = dockerCheckOutput(job=job,
                                                  defer=STOP,
                                                  workDir=os.getcwd(),
                                                  tool="quay.io/ucsc_cgl/apache-spark-worker:1.5.2",
                                                  dockerParameters=["--net=host",
                                                                    "-d",
                                                                    "-v", "/mnt/ephemeral/:/ephemeral/:rw",
                                                                    "-e",
                                                                    "\"SPARK_MASTER_IP=" + self.masterIP + ":" + _SPARK_MASTER_PORT + "\"",
                                                                    "-e", "SPARK_LOCAL_DIRS=/ephemeral/spark/local",
                                                                    "-e", "SPARK_WORKER_DIR=/ephemeral/spark/work"],
                                                  parameters=[self.masterIP + ":" + _SPARK_MASTER_PORT])[:-1]
        self.__start_datanode(job)
        
        # fake do/while to check if HDFS is up
        hdfs_down = True
        retries = 0
        while hdfs_down and (retries < 5):

            _log.info("Sleeping 30 seconds before checking HDFS startup.")
            time.sleep(30)
            clusterID = ""
            try:
                clusterID = subprocess.check_output(["docker",
                                                     "exec",
                                                     self.hdfsContainerID,
                                                     "grep",
                                                     "clusterID",
                                                     "-R",
                                                     "/opt/apache-hadoop/logs"])
            except:
                # grep returns a non-zero exit code if the pattern is not found
                # we expect to not find the pattern, so a non-zero code is OK
                pass

            if "Incompatible" in clusterID:
                _log.warning("Hadoop Datanode failed to start with: %s", clusterID)
                _log.warning("Retrying container startup, retry #%d.", retries)
                retries += 1

                _log.warning("Removing ephemeral hdfs directory.")
                subprocess.check_call(["docker",
                                       "exec",
                                       self.hdfsContainerID,
                                       "rm",
                                       "-rf",
                                       "/ephemeral/hdfs"])

                _log.warning("Killing container %s.", self.hdfsContainerID)
                subprocess.check_call(["docker",
                                       "kill",
                                       self.hdfsContainerID])

                # todo: this is copied code. clean up!
                _log.info("Restarting datanode.")
                self.__start_datanode(job)

            else:
                _log.info("HDFS datanode started up OK!")
                hdfs_down = False

        if retries >= 5:
            raise RuntimeError("Failed %d times trying to start HDFS datanode." % retries)

        return
def start(self, job):
        """
        Start spark and hdfs worker containers

        :param job: The underlying job.
        """

        # start spark and our datanode
        self.sparkContainerID = dockerCheckOutput(job=job,
                                                  defer=STOP,
                                                  workDir=os.getcwd(),
                                                  tool="quay.io/ucsc_cgl/apache-spark-worker:1.5.2",
                                                  dockerParameters=["--net=host",
                                                                    "-d",
                                                                    "-v", "/mnt/ephemeral/:/ephemeral/:rw",
                                                                    "-e",
                                                                    "\"SPARK_MASTER_IP=" + self.masterIP + ":" + _SPARK_MASTER_PORT + "\"",
                                                                    "-e", "SPARK_LOCAL_DIRS=/ephemeral/spark/local",
                                                                    "-e", "SPARK_WORKER_DIR=/ephemeral/spark/work"],
                                                  parameters=[self.masterIP + ":" + _SPARK_MASTER_PORT])[:-1]
        self.__start_datanode(job)
        
        # fake do/while to check if HDFS is up
        hdfs_down = True
        retries = 0
        while hdfs_down and (retries < 5):

            _log.info("Sleeping 30 seconds before checking HDFS startup.")
            time.sleep(30)
            clusterID = ""
            try:
                clusterID = subprocess.check_output(["docker",
                                                     "exec",
                                                     self.hdfsContainerID,
                                                     "grep",
                                                     "clusterID",
                                                     "-R",
                                                     "/opt/apache-hadoop/logs"])
            except:
                # grep returns a non-zero exit code if the pattern is not found
                # we expect to not find the pattern, so a non-zero code is OK
                pass

            if "Incompatible" in clusterID:
                _log.warning("Hadoop Datanode failed to start with: %s", clusterID)
                _log.warning("Retrying container startup, retry #%d.", retries)
                retries += 1

                _log.warning("Removing ephemeral hdfs directory.")
                subprocess.check_call(["docker",
                                       "exec",
                                       self.hdfsContainerID,
                                       "rm",
                                       "-rf",
                                       "/ephemeral/hdfs"])

                _log.warning("Killing container %s.", self.hdfsContainerID)
                subprocess.check_call(["docker",
                                       "kill",
                                       self.hdfsContainerID])

                # todo: this is copied code. clean up!
                _log.info("Restarting datanode.")
                self.__start_datanode(job)

            else:
                _log.info("HDFS datanode started up OK!")
                hdfs_down = False

        if retries >= 5:
            raise RuntimeError("Failed %d times trying to start HDFS datanode." % retries)

        return
Start spark and hdfs worker containers

        :param job: The underlying job.