Repository: jcrist/skein
Pull request #58 (2018-08-09 22:47:10)

Added support for ViewFs
Prior to this commit the Daemon only accquired a delegation token for
defaultFs which did not work well in the presence of federation. Now the
application spec contains a list the namenodes to acquire the delegation
token for in addition to defaultFs. In theory, all tokens should be
automatically renewed by the RM. However, I did not have the occasion
to test this.

superbobry: A proper way to test this is, of course, to add federation to hadoop-test-cluster, but I'd rather not have any more iterations on YARN/HDFS configuration this week :)

Patch:
diff --git a/continuous_integration/travis/script.sh b/continuous_integration/travis/script.sh
index 12b9b55..f4f2d28 100644
--- a/continuous_integration/travis/script.sh
+++ b/continuous_integration/travis/script.sh
@@ -11,9 +11,9 @@ if [[ "$DOCS" != "true" ]]; then
         htcluster exec -- kinit testuser -kt testuser.keytab
     fi
     # Run py.test inside docker
-    htcluster exec -- $CONDA_ENV/bin/py.test skein/ -vv
+    htcluster exec -- $CONDA_ENV/bin/py.test skein/skein/ -vv
     # linting
-    htcluster exec -- $CONDA_ENV/bin/flake8 skein/
+    htcluster exec -- $CONDA_ENV/bin/flake8 skein/skein/
 else
     # Build docs
     pushd docs
diff --git a/docs/source/distributing-files.rst b/docs/source/distributing-files.rst
index e447547..3e66ded 100644
--- a/docs/source/distributing-files.rst
+++ b/docs/source/distributing-files.rst
@@ -109,49 +109,46 @@ When an application launches, the following process occurs:
    container and any archives are extracted to the destination directories.
    Based on the file ``visibility`` setting, this may be done once per node per
    application, or once per node (with an LRU cache for clearing old files).
-   For more information on this process see `this blogpost from Hortonworks
-   <https://hortonworks.com/blog/resource-localization-in-yarn-deep-dive/>`__.
+   For more information on this process see `this blogpost from Hortonworks`_.
 
 5. When the application completes, the staging directory is deleted.
 
 
-Distribute Python Environments With Conda-Pack
-----------------------------------------------
+Distributing Python Environments
+--------------------------------
 
 When deploying Python applications, one needs to figure out how to distribute
-any library dependencies. If Python and all required libraries are already
-installed on every node (option 1 above), then you can use the local Python and
-avoid this problem completely. Otherwise, one way that users can distribute
-libraries themselves is to use the `conda package manager
-<https://conda.io/docs/>`__ to create a Python environment, and `conda-pack
-<https://conda.github.io/conda-pack/>`__ to package that environment for
-distribution with YARN (or other services).
-
-``conda-pack`` is a tool for taking a conda environment and creating an archive
-of it in a way that (most) absolute paths in any libraries or scripts are
-altered to be relocatable. This archive then can be distributed with your
-application, and will be automatically extracted during `YARN resource
-localization
-<https://hortonworks.com/blog/resource-localization-in-yarn-deep-dive/>`__.
-
-
-Packaging the Environment
-~~~~~~~~~~~~~~~~~~~~~~~~~
-
-Here we create an example environment using ``conda``, and then use
-``conda-pack`` to package the environment into a ``tar.gz`` file named
+any library dependencies. If Python and the required libraries are already
+installed on every node (option 1 above), you can use the local Python and
+avoid this problem completely. If they aren't, then one needs to package the
+environment to distribute with the application. This is typically handled using
+
+- conda-pack_ for conda_ environments
+- venv-pack_  for virtual environments (both venv_ and virtualenv_ supported)
+
+Both are tools for taking an environment and creating an archive of it in a way
+that (most) absolute paths in any libraries or scripts are altered to be
+relocatable. This archive then can be distributed with your application, and
+will be automatically extracted during `YARN resource localization`_
+
+
+Packaging a Conda Environment with Conda-Pack
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Here we create a conda environment using conda_, and then use
+conda-pack_ to package the environment into a ``tar.gz`` file named
 ``environment.tar.gz``. This is what will be distributed with our application.
 
-.. code-block:: console
+.. code-block:: bash
 
-    # Create a new conda environment (output trimmed for brevity)
+    # Create a new conda environment
     $ conda create -n example
     ...
 
     # Activate the environment
     $ conda activate example
 
-    # Install the needed packages (output trimmed for brevity)
+    # Install the needed packages
     $ conda install conda-pack skein numpy scikit-learn numba -c conda-forge
     ...
 
@@ -162,12 +159,47 @@ Here we create an example environment using ``conda``, and then use
     [########################################] | 100% Completed | 24.2s
 
 
+Packaging a Virtual Environment with Venv-Pack
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Here we create a virtual environment, and then use venv-pack_ to package it
+into a ``tar.gz`` file named ``environment.tar.gz``. This is what will be
+distributed with our application. The virtual environment can be created using
+either venv_ or virtualenv_.
+
+Note that the python linked to in the virtual environment must exist and be
+accessible on every node in the YARN cluster. If the environment was created
+with a different Python, you can change the link path using the
+``--python-prefix`` flag. For more information see the `venv-pack
+documentation`_.
+
+.. code-block:: bash
+
+    # Create a virtual environment
+    $ python -m venv example            # Using venv
+    $ python -m virtualenv example      # Or using virtualenv
+    ...
+
+    # Activate the environment
+    $ source example/bin/activate
+
+    # Install the needed packages
+    $ pip install venv-pack skein numpy scikit-learn numba
+    ...
+
+    # Package the environment into environment.tar.gz
+    $ venv-pack -o environment.tar.gz
+    Collecting packages...
+    Packing environment at '/home/jcrist/environments/example' to 'environment.tar.gz'
+    [########################################] | 100% Completed |  12.4s
+
+
 Using the Packaged Environment
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
 To use the packaged environment in a service specification, you need to include
 the archive in ``files``, and activate the environment in the ``commands``
-list.
+list. This looks the same for environments packaged using either tool.
 
 .. code-block:: yaml
 
@@ -185,3 +217,14 @@ list.
           # Run commands inside the environment. All executables or imported
           # python libraries will be from within the packaged environment.
           - my-cool-application
+
+
+.. _conda-pack: https://conda.github.io/conda-pack/
+.. _conda: http://conda.io/
+.. _venv:
+.. _virtual environments: https://docs.python.org/3/library/venv.html
+.. _virtualenv: https://virtualenv.pypa.io/en/stable/
+.. _venv-pack documentation:
+.. _venv-pack: https://jcrist.github.io/venv-pack/
+.. _this blogpost from Hortonworks:
+.. _YARN resource localization: https://hortonworks.com/blog/resource-localization-in-yarn-deep-dive/
diff --git a/examples/echo_server/client.py b/examples/echo_server/client.py
new file mode 100644
index 0000000..2fa8d8b
--- /dev/null
+++ b/examples/echo_server/client.py
@@ -0,0 +1,41 @@
+import asyncio
+import skein
+import sys
+
+
+async def tcp_echo_client(message, loop, host, port):
+    """Generic python tcp echo client"""
+    print("Connecting to server at %s:%d" % (host, port))
+    reader, writer = await asyncio.open_connection(host, port, loop=loop)
+
+    writer.write(message.encode())
+    print('Sent: %r' % message)
+
+    data = await reader.read(100)
+    print('Received: %r' % data.decode())
+
+    writer.close()
+
+
+async def echo_all(app, message):
+    """Send and recieve a message from all running echo servers"""
+    # Loop through all registered server addresses
+    for address in app.kv.get_prefix('address.').values():
+        # Parse the host and port from the stored address
+        host, port = address.decode().split(':')
+        port = int(port)
+
+        # Send the message to the echo server
+        await tcp_echo_client(message, loop, host, port)
+
+
+# Get the application id from the command-line args
+app_id = sys.argv[1]
+
+# Connect to the application
+app = skein.Client().connect(app_id)
+
+# Send message to every running echo server
+loop = asyncio.get_event_loop()
+loop.run_until_complete(echo_all(app, 'Hello World!'))
+loop.close()
diff --git a/examples/echo_server/server.py b/examples/echo_server/server.py
new file mode 100644
index 0000000..e22b769
--- /dev/null
+++ b/examples/echo_server/server.py
@@ -0,0 +1,49 @@
+import asyncio
+import socket
+import skein
+
+
+async def handle_echo(reader, writer):
+    """Generic python tcp echo server"""
+    data = await reader.read(100)
+    message = data.decode()
+    print("Received %r" % message)
+
+    writer.write(data)
+    await writer.drain()
+    print("Sent: %r" % message)
+
+    writer.close()
+
+
+# Setup the server with a dynamically chosen port
+loop = asyncio.get_event_loop()
+coro = asyncio.start_server(handle_echo, '0.0.0.0', 0, loop=loop)
+server = loop.run_until_complete(coro)
+
+# Determine the dynamically chosen address
+host = socket.gethostbyname(socket.gethostname())
+port = server.sockets[0].getsockname()[1]
+address = '%s:%d' % (host, port)
+
+# Form a unique key to store the address using the current container id
+key = 'address.%s' % skein.properties.container_id
+
+# Connect to the application master
+app = skein.ApplicationClient.from_current()
+
+# The key-value store only accepts bytes as values
+value = address.encode()
+
+# Store the server address in the key-value store, assigning the current
+# container as the owner of the key. This ensures that the key is deleted if
+# the container exits.
+app.kv.put(key, value, owner=skein.properties.container_id)
+
+# Serve requests until shutdown
+loop.run_forever()
+
+# Cleanup
+server.close()
+loop.run_until_complete(server.wait_closed())
+loop.close()
diff --git a/examples/echo_server/spec.yaml b/examples/echo_server/spec.yaml
new file mode 100644
index 0000000..2eaf6e8
--- /dev/null
+++ b/examples/echo_server/spec.yaml
@@ -0,0 +1,19 @@
+name: echoserver
+
+services:
+    server:
+        resources:
+            vcores: 1
+            memory: 256
+        files:
+            # A packaged conda environment to be distributed with the
+            # application. During YARN resource localization this will be
+            # automatically unpacked into the directory ``environment``.
+            environment: environment.tar.gz
+            # The server implementation.
+            server.py: server.py
+        commands:
+            # Activate the conda environment
+            - source environment/bin/activate
+            # Start the server
+            - python server.py
diff --git a/java/pom.xml b/java/pom.xml
index bb1720b..d5befc2 100644
--- a/java/pom.xml
+++ b/java/pom.xml
@@ -171,6 +171,13 @@
       <scope>provided</scope>
     </dependency>
 
+    <dependency>
+      <groupId>org.apache.hadoop</groupId>
+      <artifactId>hadoop-mapreduce-client-core</artifactId>
+      <version>${hadoopVersion}</version>
+      <scope>provided</scope>
+    </dependency>
+
     <dependency>
       <groupId>org.apache.hadoop</groupId>
       <artifactId>hadoop-yarn-client</artifactId>
diff --git a/java/src/main/java/com/anaconda/skein/Daemon.java b/java/src/main/java/com/anaconda/skein/Daemon.java
index eda8d6d..4857688 100644
--- a/java/src/main/java/com/anaconda/skein/Daemon.java
+++ b/java/src/main/java/com/anaconda/skein/Daemon.java
@@ -1,5 +1,7 @@
 package com.anaconda.skein;
 
+import com.google.common.collect.ObjectArrays;
+
 import io.grpc.Server;
 import io.grpc.Status;
 import io.grpc.StatusRuntimeException;
@@ -17,10 +19,11 @@ import org.apache.hadoop.fs.FileUtil;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.io.DataOutputBuffer;
+import org.apache.hadoop.mapreduce.security.TokenCache;
 import org.apache.hadoop.security.Credentials;
 import org.apache.hadoop.security.UserGroupInformation;
-import org.apache.hadoop.yarn.api.ApplicationConstants.Environment;
 import org.apache.hadoop.yarn.api.ApplicationConstants;
+import org.apache.hadoop.yarn.api.ApplicationConstants.Environment;
 import org.apache.hadoop.yarn.api.records.ApplicationId;
 import org.apache.hadoop.yarn.api.records.ApplicationReport;
 import org.apache.hadoop.yarn.api.records.ApplicationSubmissionContext;
@@ -263,14 +266,14 @@ public class Daemon {
     // Add security tokens as needed
     ByteBuffer fsTokens = null;
     if (UserGroupInformation.isSecurityEnabled()) {
-      Credentials credentials = new Credentials();
-      String tokenRenewer = conf.get(YarnConfiguration.RM_PRINCIPAL);
-      if (tokenRenewer == null || tokenRenewer.length() == 0) {
-        throw new IOException("Can't determine Yarn ResourceManager Kerberos "
-                              + "principal for the RM to use as renewer");
-      }
+      Credentials credentials = UserGroupInformation.getLoginUser().getCredentials();
+      TokenCache.obtainTokensForNamenodes(
+              credentials,
+              ObjectArrays.concat(
+                      new Path(defaultFs.getUri()),
+                      spec.getFileSystems().toArray(new Path[0])),
+              conf);
 
-      defaultFs.addDelegationTokens(tokenRenewer, credentials);
       DataOutputBuffer dob = new DataOutputBuffer();
       credentials.writeTokenStorageToStream(dob);
       fsTokens = ByteBuffer.wrap(dob.getData(), 0, dob.getLength());
diff --git a/java/src/main/java/com/anaconda/skein/Model.java b/java/src/main/java/com/anaconda/skein/Model.java
index d529ed9..f358719 100644
--- a/java/src/main/java/com/anaconda/skein/Model.java
+++ b/java/src/main/java/com/anaconda/skein/Model.java
@@ -1,5 +1,6 @@
 package com.anaconda.skein;
 
+import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.yarn.api.records.ContainerId;
 import org.apache.hadoop.yarn.api.records.LocalResource;
 import org.apache.hadoop.yarn.api.records.NodeId;
@@ -106,16 +107,19 @@ public class Model {
     private String queue;
     private int maxAttempts;
     private Set<String> tags;
+    private List<Path> fileSystems;
     private Map<String, Service> services;
 
     public ApplicationSpec() {}
 
     public ApplicationSpec(String name, String queue, int maxAttempts,
-                           Set<String> tags, Map<String, Service> services) {
+                           Set<String> tags, List<Path> fileSystems,
+                           Map<String, Service> services) {
       this.name = name;
       this.queue = queue;
       this.maxAttempts = maxAttempts;
       this.tags = tags;
+      this.fileSystems = fileSystems;
       this.services = services;
     }
 
@@ -125,6 +129,7 @@ public class Model {
               + "queue: " + queue + ", "
               + "maxAttempts: " + maxAttempts + ", "
               + "tags: " + tags + ", "
+              + "fileSystems" + fileSystems + ", "
               + "services: " + services + ">");
     }
 
@@ -140,6 +145,11 @@ public class Model {
     public void setTags(Set<String> tags) { this.tags = tags; }
     public Set<String> getTags() { return this.tags; }
 
+    public void setFileSystems(List<Path> fileSystems) {
+      this.fileSystems = fileSystems;
+    }
+    public List<Path> getFileSystems() { return this.fileSystems; }
+
     public void setServices(Map<String, Service> services) { this.services = services; }
     public Map<String, Service> getServices() { return services; }
 
@@ -148,6 +158,7 @@ public class Model {
       throwIfNull(queue, "queue");
       throwIfLessThan(maxAttempts, 1, "maxAttempts");
       throwIfNull(tags, "tags");
+      throwIfNull(fileSystems, "fileSystems");
       throwIfNull(services, "services");
       if (services.size() == 0) {
         throw new IllegalArgumentException("There must be at least one service");
diff --git a/java/src/main/java/com/anaconda/skein/MsgUtils.java b/java/src/main/java/com/anaconda/skein/MsgUtils.java
index fd7b69c..b1263ab 100644
--- a/java/src/main/java/com/anaconda/skein/MsgUtils.java
+++ b/java/src/main/java/com/anaconda/skein/MsgUtils.java
@@ -1,5 +1,9 @@
 package com.anaconda.skein;
 
+import com.google.common.base.Functions;
+import com.google.common.collect.Lists;
+
+import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.yarn.api.records.ApplicationReport;
 import org.apache.hadoop.yarn.api.records.ApplicationResourceUsageReport;
 import org.apache.hadoop.yarn.api.records.ContainerId;
@@ -299,7 +303,8 @@ public class MsgUtils {
         .setName(spec.getName())
         .setQueue(spec.getQueue())
         .setMaxAttempts(spec.getMaxAttempts())
-        .addAllTags(spec.getTags());
+        .addAllTags(spec.getTags())
+        .addAllFileSystems(Lists.transform(spec.getFileSystems(), Functions.toStringFunction()));
 
     for (Map.Entry<String, Model.Service> entry : spec.getServices().entrySet()) {
       builder.putServices(entry.getKey(), writeService(entry.getValue()));
@@ -312,10 +317,16 @@ public class MsgUtils {
     for (Map.Entry<String, Msg.Service> entry : spec.getServicesMap().entrySet()) {
       services.put(entry.getKey(), readService(entry.getValue()));
     }
+
+    final List<Path> fileSystems = new ArrayList<Path>();
+    for (int i = 0; i < spec.getFileSystemsCount(); i++) {
+      fileSystems.add(new Path(spec.getFileSystems(i)));
+    }
     return new Model.ApplicationSpec(spec.getName(),
                                      spec.getQueue(),
                                      spec.getMaxAttempts(),
                                      new HashSet<String>(spec.getTagsList()),
+                                     fileSystems,
                                      services);
   }
 
diff --git a/java/src/main/proto/skein.proto b/java/src/main/proto/skein.proto
index fa9e158..30f40c8 100644
--- a/java/src/main/proto/skein.proto
+++ b/java/src/main/proto/skein.proto
@@ -87,7 +87,8 @@ message ApplicationSpec {
   string queue = 2;
   int32 max_attempts = 3;
   repeated string tags = 4;
-  map<string, Service> services = 5;
+  repeated string file_systems = 5;
+  map<string, Service> services = 6;
 }
 
 
diff --git a/setup.py b/setup.py
index 378252e..9946e76 100644
--- a/setup.py
+++ b/setup.py
@@ -65,6 +65,8 @@ class build_proto(Command):
 class build_java(Command):
     description = "build java artifacts"
 
+    user_options = []
+
     def initialize_options(self):
         pass
 
diff --git a/skein/model.py b/skein/model.py
index 3f9bb4f..f1721e4 100644
--- a/skein/model.py
+++ b/skein/model.py
@@ -472,20 +472,25 @@ class ApplicationSpec(Specification):
         The queue to submit to. Defaults to the default queue.
     tags : set, optional
         A set of strings to use as tags for this application.
+    file_systems : list, optional
+        A list of Hadoop file systems to acquire delegation tokens for.
+        A token is always acuired for the ``defaultFS``.
     max_attempts : int, optional
         The maximum number of submission attempts before marking the
         application as failed. Note that this only considers failures of the
         application master during startup. Default is 1.
     """
-    __slots__ = ('services', 'name', 'queue', 'tags', 'max_attempts')
+    __slots__ = ('services', 'name', 'queue', 'tags', 'file_systems',
+                 'max_attempts')
     _protobuf_cls = _proto.ApplicationSpec
 
-    def __init__(self, services=required, name='skein', queue='default', tags=None,
-                 max_attempts=1):
+    def __init__(self, services=required, name='skein', queue='default',
+                 tags=None, file_systems=None, max_attempts=1):
         self._assign_required('services', services)
         self.name = name
         self.queue = queue
         self.tags = set() if tags is None else set(tags)
+        self.file_systems = [] if file_systems is None else file_systems
         self.max_attempts = max_attempts
         self._validate()
 
@@ -497,6 +502,7 @@ class ApplicationSpec(Specification):
         self._check_is_type('name', string)
         self._check_is_type('queue', string)
         self._check_is_set_of('tags', string)
+        self._check_is_list_of('file_systems', string)
         self._check_is_bounded_int('max_attempts', min=1)
         self._check_is_dict_of('services', string, Service)
         if not self.services:
@@ -536,6 +542,7 @@ class ApplicationSpec(Specification):
         return cls(name=obj.name,
                    queue=obj.queue,
                    tags=set(obj.tags),
+                   file_systems=list(obj.file_systems),
                    max_attempts=min(1, obj.max_attempts),
                    services=services)
 


Test patch:
diff --git a/skein/test/test_core.py b/skein/test/test_core.py
index c9e6b10..230a281 100644
--- a/skein/test/test_core.py
+++ b/skein/test/test_core.py
@@ -259,3 +259,16 @@ def test_container_environment(client, has_kerberos_enabled):
     else:
         assert "LOGIN_ID=[yarn]" in logs
         assert "HADOOP_USER_NAME" in logs
+
+
+def test_file_systems(client):
+    commands = ['hdfs dfs -touchz /user/testuser/test_file_systems']
+    service = skein.Service(resources=skein.Resources(memory=124, vcores=1),
+                            commands=commands)
+    spec = skein.ApplicationSpec(name="test_file_systems",
+                                 queue="default",
+                                 services={'service': service},
+                                 file_systems=["hdfs://master.example.com:9000"])
+
+    with run_application(client, spec=spec) as app:
+        wait_for_success(client, app.id)
diff --git a/skein/test/test_model.py b/skein/test/test_model.py
index a27630a..f00246f 100644
--- a/skein/test/test_model.py
+++ b/skein/test/test_model.py
@@ -43,6 +43,8 @@ max_attempts: 2
 tags:
     - tag1
     - tag2
+file_systems:
+    - hdfs://preprod
 
 services:
     service_1:
@@ -283,6 +285,7 @@ def test_application_spec_from_yaml():
     assert spec.name == 'test'
     assert spec.queue == 'default'
     assert spec.tags == {'tag1', 'tag2'}
+    assert spec.file_systems == ['hdfs://preprod']
     assert spec.max_attempts == 2
     assert isinstance(spec.services, dict)
     assert isinstance(spec.services['service_1'], Service)