PR #244, continued

spark/src/cgcloud/spark/__init__.py

@@ -1,5 +1,5 @@
 def roles( ):
-    from cgcloud.spark.spark_box import SparkBox, SparkSlave, SparkMaster
+    from cgcloud.spark.spark_box import SparkBox, Spark2Box, SparkSlave, SparkMaster
     return sorted( locals( ).values( ), key=lambda cls: cls.__name__ )
 
 

spark/src/cgcloud/spark/spark_box.py

@@ -25,10 +25,6 @@
 persistent_dir = '/mnt/persistent'
 var_dir = '/var/lib/sparkbox'
 hdfs_replication = 1
-hadoop_version = '2.6.0'
-spark_version = '1.6.2'
-# The major version of Hadoop that the Spark binaries were built against 
-spark_hadoop_version = '2.6'
 
 Service = namedtuple( 'Service', [
     'init_name',
@@ -82,6 +78,11 @@ class SparkBox( ApacheSoftwareBox,
     typically seconds after the reservation has been submitted, the slaves can be started up.
     """
 
+    hadoop_version = '2.6.0'
+    spark_version = '1.6.2'
+    # The major version of Hadoop that the Spark binaries were built against
+    spark_hadoop_version = '2.6'
+
     @classmethod
     def get_role_options( cls ):
         return super( SparkBox, cls ).get_role_options( ) + [
@@ -107,7 +108,11 @@ def _populate_security_group( self, group_id ):
             dict( ip_protocol='tcp', from_port=0, to_port=65535,
                   src_security_group_group_id=group_id ),
             dict( ip_protocol='udp', from_port=0, to_port=65535,
-                  src_security_group_group_id=group_id ) ]
+                  src_security_group_group_id=group_id ),
+            dict( ip_protocol='tcp', from_port=8080, to_port=8081,
+                  cidr_ip='0.0.0.0/0' ),
+            dict( ip_protocol='tcp', from_port=4040, to_port=4045,
+                  cidr_ip='0.0.0.0/0' )]
 
     def _get_iam_ec2_role( self ):
         iam_role_name, policies = super( SparkBox, self )._get_iam_ec2_role( )
@@ -189,7 +194,7 @@ def __ec2_keypair_name( self, ctx ):
     @fabric_task
     def __install_hadoop( self ):
         # Download and extract Hadoop
-        path = fmt( 'hadoop/common/hadoop-{hadoop_version}/hadoop-{hadoop_version}.tar.gz' )
+        path = fmt( 'hadoop/common/hadoop-{self.hadoop_version}/hadoop-{self.hadoop_version}.tar.gz' )
         self._install_apache_package( path, install_dir )
 
         # Add environment variables to hadoop_env.sh
@@ -258,7 +263,7 @@ def __to_hadoop_xml_config( properties ):
     @fabric_task
     def __install_spark( self ):
         # Download and extract Spark
-        path = fmt( 'spark/spark-{spark_version}/spark-{spark_version}-bin-hadoop{spark_hadoop_version}.tgz' )
+        path = fmt( 'spark/spark-{self.spark_version}/spark-{self.spark_version}-bin-hadoop{self.spark_hadoop_version}.tgz' )
         self._install_apache_package( path, install_dir )
 
         spark_dir = var_dir + "/spark"
@@ -271,8 +276,9 @@ def __install_spark( self ):
             SPARK_WORKER_DIR=self._lazy_mkdir( spark_dir, "work" ),
             SPARK_LOCAL_DIRS=self._lazy_mkdir( spark_dir, "local" ),
             JAVA_HOME='/usr/lib/jvm/java-8-oracle',
-            SPARK_MASTER_IP='spark-master',
-            HADOOP_CONF_DIR=fmt( "{install_dir}/hadoop/etc/hadoop" ) )
+            SPARK_MASTER_HOST='spark-master',
+            HADOOP_CONF_DIR=fmt( "{install_dir}/hadoop/etc/hadoop" ),
+            SPARK_PUBLIC_DNS="$(curl -s http://169.254.169.254/latest/meta-data/public-hostname)" )
         with remote_open( spark_env_sh_path, use_sudo=True ) as spark_env_sh:
             spark_env_sh.write( '\n' )
             for name, value in spark_env.iteritems( ):
@@ -430,6 +436,18 @@ def __setup_path( self ):
                             # Spark's and Hadoop's sbin
                             f.write( fmt( 'PATH="$PATH:{install_dir}/{package}/bin"\n' ) )
 
+class Spark2Box( SparkBox ):
+    """
+    A node in a Spark v2.x cluster; used only to create an image for master and worker boxes
+    """
+
+    hadoop_version = '2.7.3'
+    spark_version = '2.1.0'
+    # The major version of Hadoop that the Spark binaries were built against
+    spark_hadoop_version = '2.7'
+
+    def __init__( self, ctx ):
+        super( Spark2Box, self ).__init__( ctx )
 
 class SparkMaster( SparkBox, ClusterLeader ):
     """

spark/src/cgcloud/spark/test/test_spark.py

@@ -4,39 +4,49 @@
 import time
 import logging
 import unittest
+from abc import ABCMeta, abstractmethod
 
 from cgcloud.core.test import CoreTestCase
-from cgcloud.spark.spark_box import install_dir, SparkBox, SparkMaster, SparkSlave
+from cgcloud.spark.spark_box import install_dir, SparkBox, Spark2Box, SparkMaster, SparkSlave
 
 log = logging.getLogger( __name__ )
 
 master = SparkMaster.role( )
 slave = SparkSlave.role( )
-node = SparkBox.role( )
 
 num_slaves = 2
 
 
-class SparkClusterTests( CoreTestCase ):
+class BaseSparkClusterTests( CoreTestCase ):
     """
     Covers the creation of a Spark cluster from scratch and running a simple Spark job on it.
     Also covers persistant HDFS between two cluster incarnations.
     """
+    __metaclass__ = ABCMeta
+
+    node = NotImplemented
     cleanup = True
     create_image = True
 
+    @classmethod
+    @abstractmethod
+    def initNode( cls ):
+        raise NotImplementedError("Abstract method")
+
     @classmethod
     def setUpClass( cls ):
+        cls.initNode
         os.environ[ 'CGCLOUD_PLUGINS' ] = 'cgcloud.spark'
-        super( SparkClusterTests, cls ).setUpClass( )
+        super( BaseSparkClusterTests, cls ).setUpClass( )
         if cls.create_image:
-            cls._cgcloud( 'create', node, '-IT' )
+            cls._cgcloud( 'create', cls.node, '-IT' )
 
     @classmethod
     def tearDownClass( cls ):
         if cls.cleanup and cls.create_image:
-            cls._cgcloud( 'delete-image', node )
-        super( SparkClusterTests, cls ).tearDownClass( )
+            cls._cgcloud( 'delete-image', cls.node )
+        super( BaseSparkClusterTests, cls ).tearDownClass( )
+
 
     def test_wordcount( self ):
         self._create_cluster( )
@@ -131,3 +141,24 @@ def word_count( ):
 
     def _delete_volumes( self ):
         pass
+
+class SparkClusterTests( BaseSparkClusterTests ):
+    """
+    Covers the creation of a Spark v1.x cluster from scratch and running a simple Spark job on it.
+    Also covers persistant HDFS between two cluster incarnations.
+    """
+
+    @classmethod
+    def initNode( cls ):
+        cls.node = SparkBox.role( )
+
+
+class Spark2ClusterTests( BaseSparkClusterTests ):
+    """
+    Covers the creation of a Spark v2.x cluster from scratch and running a simple Spark job on it.
+    Also covers persistant HDFS between two cluster incarnations.
+    """
+
+    @classmethod
+    def initNode( cls ):
+        cls.node = Spark2Box.role( )