Add support for SSH sock proxying to scheduler

heron/config/src/yaml/conf/aurora/scheduler.yaml

@@ -8,4 +8,25 @@ heron.class.launcher:                        com.twitter.heron.scheduler.aurora.
 heron.directory.sandbox.java.home:           /usr/lib/jvm/java-1.8.0-openjdk-amd64/
 
 # Invoke the IScheduler as a library directly
-heron.scheduler.is.service:                  False
+heron.scheduler.is.service:                  False
+
+####################################################################
+# Following are config for tunneling
+####################################################################
+# Whether we should attempt to tunnel if there is no direct access to a remote host (e.g. TMaster)
+heron.scheduler.is.tunnel.needed:                          False
+
+# The connection timeout in ms when testing if we can connect to remote host
+heron.scheduler.tunnel.connection.timeout.ms:              1000
+
+# The number of retries when testing direct access to remote host
+heron.scheduler.tunnel.connection.retry.count:             2
+
+# The interval in ms between two retries when testing direct access to remote host
+heron.scheduler.tunnel.retry.interval.ms:                  1000
+
+# The number of retries when verifying connectivity via tunnel
+heron.scheduler.tunnel.verify.count:                       10
+
+# SSH tunnel host
+heron.scheduler.tunnel.host:                               "my.tunnel.host"

heron/config/src/yaml/conf/aurora/statemgr.yaml

@@ -10,3 +10,23 @@ heron.statemgr.root.path:                       /vagrant/.herondata/repository/s
 # create the sub directories, if needed
 heron.statemgr.localfs.is.initialize.file.tree: True
 
+####################################################################
+# Following are config for tunneling
+####################################################################
+# Whether we should attempt to tunnel if there is no direct access to a remote host (e.g. Zookeeper)
+heron.statemgr.is.tunnel.needed:                          False
+
+# The connection timeout in ms when testing if we can connect to remote host
+heron.statemgr.tunnel.connection.timeout.ms:              1000
+
+# The number of retries when testing direct access to remote host
+heron.statemgr.tunnel.connection.retry.count:             2
+
+# The interval in ms between two retries when testing direct access to remote host
+heron.statemgr.tunnel.retry.interval.ms:                  1000
+
+# The number of retries when verifying connectivity via tunnel
+heron.statemgr.tunnel.verify.count:                       10
+
+# SSH tunnel host
+heron.statemgr.tunnel.host:                               "my.tunnel.host"

heron/scheduler-core/src/java/com/twitter/heron/scheduler/RuntimeManagerRunner.java

@@ -35,6 +35,7 @@
 import com.twitter.heron.spi.packing.PackingPlanProtoDeserializer;
 import com.twitter.heron.spi.packing.PackingPlanProtoSerializer;
 import com.twitter.heron.spi.statemgr.SchedulerStateManagerAdaptor;
+import com.twitter.heron.spi.utils.NetworkUtils;
 import com.twitter.heron.spi.utils.ReflectionUtils;
 import com.twitter.heron.spi.utils.Runtime;
 import com.twitter.heron.spi.utils.TMasterUtils;
@@ -91,18 +92,22 @@ public Boolean call() {
    * Handler to activate a topology
    */
   private boolean activateTopologyHandler(String topologyName) {
+    NetworkUtils.TunnelConfig tunnelConfig =
+        NetworkUtils.TunnelConfig.build(config, NetworkUtils.HeronSystem.SCHEDULER);
     return TMasterUtils.transitionTopologyState(topologyName,
         TMasterUtils.TMasterCommand.ACTIVATE, Runtime.schedulerStateManagerAdaptor(runtime),
-        TopologyAPI.TopologyState.PAUSED, TopologyAPI.TopologyState.RUNNING);
+        TopologyAPI.TopologyState.PAUSED, TopologyAPI.TopologyState.RUNNING, tunnelConfig);
   }
 
   /**
    * Handler to deactivate a topology
    */
   private boolean deactivateTopologyHandler(String topologyName) {
+    NetworkUtils.TunnelConfig tunnelConfig =
+        NetworkUtils.TunnelConfig.build(config, NetworkUtils.HeronSystem.SCHEDULER);
     return TMasterUtils.transitionTopologyState(topologyName,
         TMasterUtils.TMasterCommand.DEACTIVATE, Runtime.schedulerStateManagerAdaptor(runtime),
-        TopologyAPI.TopologyState.RUNNING, TopologyAPI.TopologyState.PAUSED);
+        TopologyAPI.TopologyState.RUNNING, TopologyAPI.TopologyState.PAUSED, tunnelConfig);
   }
 
   /**

heron/schedulers/src/java/com/twitter/heron/scheduler/local/LocalScheduler.java

@@ -58,7 +58,8 @@ public class LocalScheduler implements IScheduler, IScalable {
   public void initialize(Config mConfig, Config mRuntime) {
     this.config = mConfig;
     this.runtime = mRuntime;
-    this.updateTopologyManager = new UpdateTopologyManager(runtime, Optional.<IScalable>of(this));
+    this.updateTopologyManager =
+        new UpdateTopologyManager(runtime, Optional.<IScalable>of(this));
   }
 
   @Override
@@ -75,7 +76,7 @@ public void close() {
    */
   @VisibleForTesting
   protected Process startExecutorProcess(int container) {
-    return ShellUtils.runASyncProcess(true,
+    return ShellUtils.runASyncProcess(
         getExecutorCommand(container),
         new File(LocalContext.workingDirectory(config)),
         Integer.toString(container));

heron/schedulers/tests/java/com/twitter/heron/scheduler/yarn/HeronExecutorTaskTest.java

@@ -101,7 +101,8 @@ public void setsEnvironmentForExecutor() throws Exception {
         Mockito.eq(testCmd),
         Mockito.any(File.class),
         Mockito.eq(env),
-        Mockito.any(String.class));
+        Mockito.any(String.class),
+        Mockito.any(Boolean.class));
     spyTask.call(null);
     Mockito.verify(mockProcess).waitFor();
   }

heron/spi/src/java/BUILD

@@ -140,6 +140,7 @@ java_library(
         "**/spi/utils/NetworkUtils.java",
     ]),
     deps = [
+        ":common-spi-java",
         ":shell-utils-java",
         "//heron/common/src/java:basics-java",
     ],