helix-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From kisho...@apache.org
Subject [05/15] Adding Helix-task-framework and Yarn integration modules
Date Fri, 20 Sep 2013 18:30:14 GMT
http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/e38aa54b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/impl/yarn/YarnContainerStatusProvider.java
----------------------------------------------------------------------
diff --git a/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/impl/yarn/YarnContainerStatusProvider.java b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/impl/yarn/YarnContainerStatusProvider.java
new file mode 100644
index 0000000..f7e3076
--- /dev/null
+++ b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/impl/yarn/YarnContainerStatusProvider.java
@@ -0,0 +1,52 @@
+package org.apache.helix.metamanager.impl.yarn;
+
+import org.apache.helix.metamanager.ContainerStatusProvider;
+import org.apache.helix.metamanager.impl.yarn.ContainerMetadata.ContainerState;
+import org.apache.helix.metamanager.impl.yarn.MetadataService.MetadataServiceException;
+
+public class YarnContainerStatusProvider implements ContainerStatusProvider {
+
+	final String metadataAddress;
+	
+	ZookeeperMetadataService metaService;
+	
+	public YarnContainerStatusProvider(String metadataAddress) {
+		this.metadataAddress = metadataAddress;
+		this.metaService = new ZookeeperMetadataService(metadataAddress);
+	}
+
+	@Override
+	public boolean exists(String id) {
+		return metaService.exists(id);
+	}
+
+	@Override
+	public boolean isActive(String id) {
+		try {
+			return metaService.read(id).state == ContainerState.ACTIVE;
+		} catch (MetadataServiceException e) {
+			return false;
+		}
+	}
+
+	@Override
+	public boolean isFailed(String id) {
+		try {
+			return metaService.read(id).state == ContainerState.FAILED;
+		} catch (Exception e) {
+			return false;
+		}
+	}
+
+	public void startService() {
+		metaService = new ZookeeperMetadataService(metadataAddress);
+		metaService.startService();
+	}
+	
+	public void stopService() {
+		if(metaService != null) {
+			metaService.stopService();
+			metaService = null;
+		}
+	}
+}

http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/e38aa54b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/impl/yarn/YarnDataProvider.java
----------------------------------------------------------------------
diff --git a/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/impl/yarn/YarnDataProvider.java b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/impl/yarn/YarnDataProvider.java
new file mode 100644
index 0000000..8bd80b5
--- /dev/null
+++ b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/impl/yarn/YarnDataProvider.java
@@ -0,0 +1,73 @@
+package org.apache.helix.metamanager.impl.yarn;
+
+import java.util.Collection;
+
+/**
+ * Abstraction for a (remote) repository of yarn container meta data. Meta data
+ * is read and updated by {@link YarnContainerProvider}
+ * {@link YarnMasterProcess}, {@link YarnContainerProcess}.<br/>
+ * <b>NOTE:</b> Each operation is assumed to be atomic.
+ * 
+ */
+interface YarnDataProvider {
+
+    /**
+     * Checks for existence of meta data about container insatnce
+     * 
+     * @param id
+     *            unique container id
+     * @return true, if meta data exists
+     */
+    public boolean exists(String id);
+
+    /**
+     * Create meta data entry. Check for non-existence of meta data for given
+     * container id and create node.
+     * 
+     * @param data
+     *            container meta data with unique id
+     * @throws Exception
+     *             if meta data entry already exist
+     */
+    public void create(YarnContainerData data) throws Exception;
+
+    /**
+     * Read meta data for given container id.
+     * 
+     * @param id
+     *            unique container id
+     * @return yarn container data
+     * @throws Exception
+     *             if meta data entry for given id does not exist
+     */
+    public YarnContainerData read(String id) throws Exception;
+
+    /**
+     * Read all meta data stored for this domain space of yarn providers and
+     * containers.
+     * 
+     * @return collection of meta data entries, empty if none
+     * @throws Exception
+     */
+    public Collection<YarnContainerData> readAll() throws Exception;
+
+    /**
+     * Write meta data entry.
+     * 
+     * @param data
+     *            yarn container meta data
+     * @throws Exception
+     *             if meta data entry for given id does not exist
+     */
+    public void update(YarnContainerData data) throws Exception;
+
+    /**
+     * Delete meta data entry. Frees up unique id to be reused. May throw an
+     * exception on non-existence or be idempotent.
+     * 
+     * @param id
+     *            unique container id
+     * @throws Exception
+     */
+    public void delete(String id) throws Exception;
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/e38aa54b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/impl/yarn/YarnMasterProcess.java
----------------------------------------------------------------------
diff --git a/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/impl/yarn/YarnMasterProcess.java b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/impl/yarn/YarnMasterProcess.java
new file mode 100644
index 0000000..d4447ee
--- /dev/null
+++ b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/impl/yarn/YarnMasterProcess.java
@@ -0,0 +1,144 @@
+package org.apache.helix.metamanager.impl.yarn;
+
+import java.net.InetSocketAddress;
+import java.util.Map;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.net.NetUtils;
+import org.apache.hadoop.yarn.api.AMRMProtocol;
+import org.apache.hadoop.yarn.api.ApplicationConstants;
+import org.apache.hadoop.yarn.api.protocolrecords.FinishApplicationMasterRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.RegisterApplicationMasterRequest;
+import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
+import org.apache.hadoop.yarn.api.records.ContainerId;
+import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.ipc.YarnRPC;
+import org.apache.hadoop.yarn.util.ConverterUtils;
+import org.apache.hadoop.yarn.util.Records;
+import org.apache.helix.metamanager.provider.ProviderProcess;
+import org.apache.log4j.Logger;
+
+/**
+ * Host process for {@link YarnContainerProviderProcess}. Hasts application
+ * master in YARN and provider participant to Helix meta cluster. (Program entry
+ * point)
+ * 
+ */
+class YarnMasterProcess {
+
+    static final Logger log = Logger.getLogger(YarnMasterProcess.class);
+
+    public static void main(String[] args) throws Exception {
+        log.trace("BEGIN YarnMaster.main()");
+
+        final ApplicationAttemptId appAttemptId = getApplicationAttemptId();
+        log.info(String.format("Got application attempt id '%s'", appAttemptId.toString()));
+
+        log.debug("Reading master properties");
+        YarnMasterProperties properties = YarnUtils.createMasterProperties(YarnUtils.getPropertiesFromPath(YarnUtils.YARN_MASTER_PROPERTIES));
+
+        if (!properties.isValid())
+            throw new IllegalArgumentException(String.format("master properties not valid: %s", properties.toString()));
+
+        log.debug("Connecting to resource manager");
+        Configuration conf = new YarnConfiguration();
+        conf.set(YarnConfiguration.RM_ADDRESS, properties.getResourceManager());
+        conf.set(YarnConfiguration.RM_SCHEDULER_ADDRESS, properties.getScheduler());
+        conf.set(FileSystem.FS_DEFAULT_NAME_KEY, properties.getHdfs());
+
+        final AMRMProtocol resourceManager = getResourceManager(conf);
+
+        // register the AM with the RM
+        log.debug("Registering application master");
+        RegisterApplicationMasterRequest appMasterRequest = Records.newRecord(RegisterApplicationMasterRequest.class);
+        appMasterRequest.setApplicationAttemptId(appAttemptId);
+        appMasterRequest.setHost("");
+        appMasterRequest.setRpcPort(0);
+        appMasterRequest.setTrackingUrl("");
+
+        resourceManager.registerApplicationMaster(appMasterRequest);
+
+        log.debug("Starting yarndata service");
+        final ZookeeperYarnDataProvider yarnDataService = new ZookeeperYarnDataProvider(properties.getYarnData());
+        yarnDataService.start();
+
+        log.debug("Starting yarn master service");
+        final YarnMasterService service = new YarnMasterService();
+        service.configure(properties);
+        service.setAttemptId(appAttemptId);
+        service.setYarnDataProvider(yarnDataService);
+        service.setProtocol(resourceManager);
+        service.setYarnConfiguration(conf);
+        service.start();
+
+        log.debug("Starting provider");
+        final YarnContainerProvider provider = new YarnContainerProvider();
+        provider.configure(properties);
+        provider.start();
+
+        log.debug("Starting provider process");
+        final ProviderProcess process = new ProviderProcess();
+        process.configure(properties);
+        process.setConteinerProvider(provider);
+        process.start();
+
+        log.debug("Installing shutdown hooks");
+        Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
+            @Override
+            public void run() {
+                log.debug("Stopping provider process");
+                process.stop();
+
+                log.debug("Stopping provider");
+                try { provider.stop(); } catch (Exception ignore) {}
+
+                log.debug("Stopping yarn master service");
+                service.stop();
+
+                log.debug("Stopping yarndata service");
+                yarnDataService.stop();
+
+                // finish application
+                log.debug("Sending finish request");
+                FinishApplicationMasterRequest finishReq = Records.newRecord(FinishApplicationMasterRequest.class);
+
+                finishReq.setAppAttemptId(getApplicationAttemptId());
+                finishReq.setFinishApplicationStatus(FinalApplicationStatus.SUCCEEDED);
+	    	    
+	    	    try { resourceManager.finishApplicationMaster(finishReq); } catch(Exception ignore) {}
+	    	}
+	    }));
+	    
+		log.trace("END YarnMaster.main()");
+	}
+	
+    static AMRMProtocol getResourceManager(Configuration conf) {
+        // Connect to the Scheduler of the ResourceManager.
+        YarnConfiguration yarnConf = new YarnConfiguration(conf);
+        YarnRPC rpc = YarnRPC.create(yarnConf);
+        InetSocketAddress rmAddress = NetUtils.createSocketAddr(yarnConf.get(YarnConfiguration.RM_SCHEDULER_ADDRESS,
+                YarnConfiguration.DEFAULT_RM_SCHEDULER_ADDRESS));
+        log.info("Connecting to ResourceManager at " + rmAddress);
+        AMRMProtocol resourceManager = (AMRMProtocol) rpc.getProxy(AMRMProtocol.class, rmAddress, conf);
+        return resourceManager;
+    }
+
+    static ApplicationAttemptId getApplicationAttemptId() {
+        ContainerId containerId = ConverterUtils.toContainerId(getEnv(ApplicationConstants.AM_CONTAINER_ID_ENV));
+        ApplicationAttemptId appAttemptID = containerId.getApplicationAttemptId();
+        return appAttemptID;
+    }
+
+    static String getEnv(String key) {
+        Map<String, String> envs = System.getenv();
+        String clusterName = envs.get(key);
+        if (clusterName == null) {
+            // container id should always be set in the env by the framework
+            throw new IllegalArgumentException(String.format("%s not set in the environment", key));
+        }
+        return clusterName;
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/e38aa54b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/impl/yarn/YarnMasterProperties.java
----------------------------------------------------------------------
diff --git a/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/impl/yarn/YarnMasterProperties.java b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/impl/yarn/YarnMasterProperties.java
new file mode 100644
index 0000000..abeb461
--- /dev/null
+++ b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/impl/yarn/YarnMasterProperties.java
@@ -0,0 +1,13 @@
+package org.apache.helix.metamanager.impl.yarn;
+
+/**
+ * Base configuration for {@link YarnMasterProcess}. 
+ *
+ */
+public class YarnMasterProperties extends YarnContainerProviderProperties {
+	/**
+	 * 
+	 */
+	private static final long serialVersionUID = -2209509980239674160L;
+	
+}

http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/e38aa54b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/impl/yarn/YarnMasterService.java
----------------------------------------------------------------------
diff --git a/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/impl/yarn/YarnMasterService.java b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/impl/yarn/YarnMasterService.java
new file mode 100644
index 0000000..1e7aec3
--- /dev/null
+++ b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/impl/yarn/YarnMasterService.java
@@ -0,0 +1,414 @@
+package org.apache.helix.metamanager.impl.yarn;
+
+import java.io.File;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.commons.io.FileUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.net.NetUtils;
+import org.apache.hadoop.yarn.api.AMRMProtocol;
+import org.apache.hadoop.yarn.api.ApplicationConstants;
+import org.apache.hadoop.yarn.api.ContainerManager;
+import org.apache.hadoop.yarn.api.protocolrecords.AllocateRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.AllocateResponse;
+import org.apache.hadoop.yarn.api.protocolrecords.StartContainerRequest;
+import org.apache.hadoop.yarn.api.records.AMResponse;
+import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
+import org.apache.hadoop.yarn.api.records.Container;
+import org.apache.hadoop.yarn.api.records.ContainerId;
+import org.apache.hadoop.yarn.api.records.ContainerLaunchContext;
+import org.apache.hadoop.yarn.api.records.ContainerStatus;
+import org.apache.hadoop.yarn.api.records.LocalResource;
+import org.apache.hadoop.yarn.api.records.LocalResourceType;
+import org.apache.hadoop.yarn.api.records.NodeId;
+import org.apache.hadoop.yarn.api.records.Priority;
+import org.apache.hadoop.yarn.api.records.Resource;
+import org.apache.hadoop.yarn.api.records.ResourceRequest;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.exceptions.YarnRemoteException;
+import org.apache.hadoop.yarn.ipc.YarnRPC;
+import org.apache.hadoop.yarn.util.Records;
+import org.apache.helix.metamanager.Service;
+import org.apache.helix.metamanager.impl.yarn.YarnContainerData.ContainerState;
+import org.apache.log4j.Logger;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Maps;
+
+/**
+ * Implements YARN application master. Continuously monitors container health in
+ * YARN and yarn meta data updates. Spawns and destroys containers.
+ * 
+ */
+class YarnMasterService implements Service {
+
+    static final Logger                     log                    = Logger.getLogger(YarnMasterService.class);
+
+    static final String                     REQUIRED_TYPE          = "container";
+
+    static final long                       ZOOKEEPER_TIMEOUT      = 5000;
+    static final long                       MASTERSERVICE_INTERVAL = 1000;
+
+    static final String                     CONTAINERS             = "CONTAINERS";
+
+    static final String                     YARN_CONTAINER_COMMAND = "/bin/sh %s 1>%s/stdout 2>%s/stderr";
+
+    YarnMasterProperties                    properties;
+    AMRMProtocol                            protocol;
+    ApplicationAttemptId                    attemptId;
+    Configuration                           yarnConfig;
+    YarnDataProvider                        yarnDataService;
+
+    final Map<ContainerId, Container>       unassignedContainers   = new HashMap<ContainerId, Container>();
+    final Map<ContainerId, Container>       activeContainers       = new HashMap<ContainerId, Container>();
+    final Map<ContainerId, ContainerStatus> completedContainers    = new HashMap<ContainerId, ContainerStatus>();
+    final Map<ContainerId, String>          yarn2meta              = new HashMap<ContainerId, String>();
+
+    ScheduledExecutorService                executor;
+
+    @Override
+    public void configure(Properties properties) throws Exception {
+        YarnMasterProperties yarnProperties = YarnUtils.createMasterProperties(properties);
+        Preconditions.checkArgument(yarnProperties.isValid());
+        this.properties = yarnProperties;
+    }
+
+    public void setProtocol(AMRMProtocol protocol) {
+        this.protocol = protocol;
+    }
+
+    public void setAttemptId(ApplicationAttemptId attemptId) {
+        this.attemptId = attemptId;
+    }
+
+    public void setYarnConfiguration(Configuration yarnConfig) {
+        this.yarnConfig = yarnConfig;
+    }
+
+    public void setYarnDataProvider(YarnDataProvider yarnDataService) {
+        this.yarnDataService = yarnDataService;
+    }
+
+    @Override
+    public void start() {
+        Preconditions.checkNotNull(properties);
+        Preconditions.checkNotNull(protocol);
+        Preconditions.checkNotNull(attemptId);
+        Preconditions.checkNotNull(yarnConfig);
+        Preconditions.checkNotNull(yarnDataService);
+
+        log.debug("starting yarn master service");
+
+        executor = Executors.newSingleThreadScheduledExecutor();
+        executor.scheduleAtFixedRate(new YarnService(), 0, MASTERSERVICE_INTERVAL, TimeUnit.MILLISECONDS);
+    }
+
+    @Override
+    public void stop() {
+        log.debug("stopping yarn master service");
+
+        if (executor != null) {
+            executor.shutdown();
+            while (!executor.isTerminated()) {
+                try {
+                    Thread.sleep(100);
+                } catch (InterruptedException e) {
+                    // ignore
+                }
+            }
+            executor = null;
+        }
+
+        destroyLocalMasterNamespace();
+    }
+
+    Collection<YarnContainerData> readOwnedYarnData() throws Exception {
+        log.debug("reading container data");
+
+        Collection<YarnContainerData> containers = new ArrayList<YarnContainerData>();
+        for (YarnContainerData meta : yarnDataService.readAll()) {
+            if (meta.owner.equals(properties.getName())) {
+                containers.add(meta);
+                log.debug(String.format("found container node '%s' (state=%s, yarnId=%s, owner=%s)", meta.id, meta.state, meta.yarnId, meta.owner));
+            }
+        }
+        return containers;
+    }
+
+    class YarnService implements Runnable {
+        int responseId = 0;
+
+        @Override
+        public void run() {
+            try {
+                log.debug("running yarn service update cycle");
+
+                Collection<YarnContainerData> yarndata = readOwnedYarnData();
+
+                // active meta containers
+                int numMetaActive = countActiveMeta(yarndata);
+
+                // newly acquired meta containers
+                int numMetaAcquire = countAcquireMeta(yarndata);
+
+                // destroyed meta containers
+                List<ContainerId> destroyedReleasedIds = createDestroyedReleaseList(yarndata);
+                int numMetaCompleted = destroyedReleasedIds.size();
+
+                int numMeta = numMetaAcquire + numMetaActive + numMetaCompleted;
+
+                // yarn containers
+                int numYarnUnassigned = unassignedContainers.size();
+                int numYarnActive = activeContainers.size();
+                int numYarnCompleted = completedContainers.size();
+                int numYarn = numYarnUnassigned + numYarnActive + numYarnCompleted;
+
+                int numYarnRequired = numMetaAcquire - numYarnUnassigned;
+
+                // additionally required containers
+                int numRequestAdditional = Math.max(0, numYarnRequired);
+
+                // overstock containers
+                List<ContainerId> unneededReleasedIds = createOverstockReleaseList(numYarnRequired);
+
+                int numReleased = destroyedReleasedIds.size() + unneededReleasedIds.size();
+
+                log.debug(String.format("meta containers (total=%d, acquire=%d, active=%d, completed=%d)", numMeta, numMetaAcquire, numMetaActive, numMetaCompleted));
+                log.debug(String.format("yarn containers (total=%d, unassigned=%d, active=%d, completed=%d)", numYarn, numYarnUnassigned, numYarnActive, numYarnCompleted));
+                log.debug(String.format("requesting %d new containers, releasing %d", numRequestAdditional, numReleased));
+
+                Priority priority = Records.newRecord(Priority.class);
+                priority.setPriority(0);
+
+                Resource resource = Records.newRecord(Resource.class);
+                resource.setMemory(256); // TODO make dynamic
+
+                ResourceRequest resourceRequest = Records.newRecord(ResourceRequest.class);
+                resourceRequest.setHostName("*");
+                resourceRequest.setNumContainers(numRequestAdditional);
+                resourceRequest.setPriority(priority);
+                resourceRequest.setCapability(resource);
+
+                AllocateRequest request = Records.newRecord(AllocateRequest.class);
+                request.setResponseId(responseId);
+                request.setApplicationAttemptId(attemptId);
+                request.addAsk(resourceRequest);
+                request.addAllReleases(destroyedReleasedIds);
+                request.addAllReleases(unneededReleasedIds);
+
+                responseId++;
+
+                AllocateResponse allocateResponse = null;
+                try {
+                    allocateResponse = protocol.allocate(request);
+                } catch (YarnRemoteException e) {
+                    // ignore
+                    log.error("Error allocating containers", e);
+                    return;
+                }
+
+                AMResponse response = allocateResponse.getAMResponse();
+
+                // remove unassigned container about to be freed
+                for (ContainerId id : unneededReleasedIds) {
+                    log.info(String.format("Unassigned container '%s' about to be freed, removing", id));
+                    unassignedContainers.remove(id);
+                }
+
+                // newly added containers
+                for (Container container : response.getAllocatedContainers()) {
+                    unassignedContainers.put(container.getId(), container);
+                }
+
+                log.info(String.format("%d new containers available, %d required", unassignedContainers.size(), numMetaAcquire));
+
+                Iterator<Container> itYarn = unassignedContainers.values().iterator();
+                Iterator<YarnContainerData> itMeta = yarndata.iterator();
+                while (itYarn.hasNext() && itMeta.hasNext()) {
+                    YarnContainerData meta = itMeta.next();
+
+                    if (meta.yarnId >= 0)
+                        continue;
+
+                    Container containerYarn = itYarn.next();
+
+                    log.debug(String.format("assigning yarn container '%s' to container node '%s'", containerYarn.getId(), meta.id));
+
+                    String command = String.format(YARN_CONTAINER_COMMAND, YarnUtils.YARN_CONTAINER_PATH, ApplicationConstants.LOG_DIR_EXPANSION_VAR,
+                            ApplicationConstants.LOG_DIR_EXPANSION_VAR);
+
+                    log.debug(String.format("Running container command \"%s\"", command));
+
+                    // configuration
+                    YarnContainerProcessProperties containerProp = meta.getProperties();
+                    containerProp.setProperty(YarnContainerProcessProperties.ADDRESS, properties.getAddress());
+                    containerProp.setProperty(YarnContainerProcessProperties.CLUSTER, properties.getCluster());
+                    containerProp.setProperty(YarnContainerProcessProperties.YARNDATA, properties.getYarnData());
+                    containerProp.setProperty(YarnContainerProcessProperties.NAME, meta.id);
+
+                    File propertiesFile = YarnUtils.writePropertiesToTemp(containerProp);
+
+                    // HDFS
+                    final String namespace = attemptId.getApplicationId().toString() + "/" + meta.id;
+                    final Path containerArchive = YarnUtils.copyToHdfs(YarnUtils.YARN_CONTAINER_STAGING, YarnUtils.YARN_CONTAINER_STAGING, namespace, yarnConfig);
+                    final Path containerProperties = YarnUtils.copyToHdfs(propertiesFile.getCanonicalPath(), YarnUtils.YARN_CONTAINER_PROPERTIES, namespace, yarnConfig);
+
+                    // local resources
+                    Map<String, LocalResource> localResources = new HashMap<String, LocalResource>();
+                    localResources.put(YarnUtils.YARN_CONTAINER_DESTINATION,
+                            YarnUtils.createHdfsResource(containerArchive, LocalResourceType.ARCHIVE, yarnConfig));
+                    localResources.put(YarnUtils.YARN_CONTAINER_PROPERTIES,
+                            YarnUtils.createHdfsResource(containerProperties, LocalResourceType.FILE, yarnConfig));
+
+                    ContainerLaunchContext context = Records.newRecord(ContainerLaunchContext.class);
+                    context.setContainerId(containerYarn.getId());
+                    context.setResource(containerYarn.getResource());
+                    context.setEnvironment(Maps.<String, String> newHashMap());
+                    context.setCommands(Collections.singletonList(command));
+                    context.setLocalResources(localResources);
+                    context.setUser(properties.getUser());
+
+                    log.debug(String.format("container '%s' executing command '%s'", meta.id, command));
+
+                    StartContainerRequest startReq = Records.newRecord(StartContainerRequest.class);
+                    startReq.setContainerLaunchContext(context);
+
+                    try {
+                        getContainerManager(containerYarn).startContainer(startReq);
+
+                    } catch (YarnRemoteException e) {
+                        log.error(String.format("Error starting container '%s'", meta.id), e);
+                        return;
+                    }
+
+                    log.debug(String.format("container '%s' started, updating container node", meta.id));
+
+                    meta.setProperties(containerProp);
+                    meta.setState(ContainerState.CONNECTING);
+                    meta.setYarnId(containerYarn.getId().getId());
+                    yarnDataService.update(meta);
+
+                    yarn2meta.put(containerYarn.getId(), meta.id);
+
+                    log.debug(String.format("removing '%s' from unassigned yarn containers and adding to active list", containerYarn.getId()));
+
+                    itYarn.remove();
+                    activeContainers.put(containerYarn.getId(), containerYarn);
+
+                    // cleanup
+                    propertiesFile.deleteOnExit();
+
+                }
+
+                for (ContainerStatus status : response.getCompletedContainersStatuses()) {
+                    ContainerId id = status.getContainerId();
+
+                    log.info(String.format("Container '%s' completed", id));
+
+                    if (unassignedContainers.containsKey(id)) {
+                        log.info(String.format("Unassigned container '%s' terminated, removing", id));
+                        unassignedContainers.remove(id);
+                    }
+
+                    if (activeContainers.containsKey(id)) {
+                        log.info(String.format("Active container '%s' terminated, removing", id));
+                        activeContainers.remove(id);
+
+                        String metaId = yarn2meta.get(id);
+                        YarnContainerData meta = yarnDataService.read(metaId);
+
+                        log.debug(String.format("container '%s' finalized, updating container node", meta.id));
+
+                        yarnDataService.update(meta.setState(ContainerState.FINALIZE));
+                    }
+
+                    completedContainers.put(id, status);
+                }
+
+                log.debug("yarn service update cycle complete");
+
+            } catch (Exception e) {
+                log.error("Error while executing yarn update cycle", e);
+            }
+        }
+
+        private List<ContainerId> createOverstockReleaseList(int numYarnRequired) {
+            List<ContainerId> unneededReleasedIds = new ArrayList<ContainerId>();
+            Iterator<Container> itUnassigned = unassignedContainers.values().iterator();
+            if (numYarnRequired < 0) {
+                for (int i = 0; i < -numYarnRequired && itUnassigned.hasNext(); i++) {
+                    Container container = itUnassigned.next();
+                    unneededReleasedIds.add(container.getId());
+                    log.debug(String.format("Container '%s' no longer required, removing", container.getId()));
+                    itUnassigned.remove();
+                }
+            }
+            return unneededReleasedIds;
+        }
+
+        private List<ContainerId> createDestroyedReleaseList(Collection<YarnContainerData> yarndata) {
+            List<ContainerId> releasedIds = new ArrayList<ContainerId>();
+            for (YarnContainerData meta : yarndata) {
+                if (meta.state == ContainerState.HALTED) {
+                    ContainerId containerId = Records.newRecord(ContainerId.class);
+                    containerId.setApplicationAttemptId(attemptId);
+                    containerId.setId(meta.yarnId);
+                    releasedIds.add(containerId);
+                    log.debug(String.format("releasing container '%s'", containerId));
+                }
+            }
+            return releasedIds;
+        }
+
+        private int countAcquireMeta(Collection<YarnContainerData> yarndata) {
+            int numMetaAcquire = 0;
+            for (YarnContainerData meta : yarndata) {
+                if (meta.state == ContainerState.ACQUIRE) {
+                    numMetaAcquire++;
+                }
+            }
+            return numMetaAcquire;
+        }
+
+        private int countActiveMeta(Collection<YarnContainerData> yarndata) {
+            int numMetaActive = 0;
+            for (YarnContainerData meta : yarndata) {
+                if (meta.state != ContainerState.ACQUIRE && meta.state != ContainerState.HALTED && meta.state != ContainerState.FINALIZE) {
+                    numMetaActive++;
+                }
+            }
+            return numMetaActive;
+        }
+    }
+
+    private ContainerManager getContainerManager(Container container) {
+        YarnConfiguration yarnConf = new YarnConfiguration(yarnConfig);
+        YarnRPC rpc = YarnRPC.create(yarnConf);
+        NodeId nodeId = container.getNodeId();
+        String containerIpPort = String.format("%s:%d", nodeId.getHost(), nodeId.getPort());
+        log.info("Connecting to ContainerManager at: " + containerIpPort);
+        InetSocketAddress addr = NetUtils.createSocketAddr(containerIpPort);
+        ContainerManager cm = (ContainerManager) rpc.getProxy(ContainerManager.class, addr, yarnConfig);
+        return cm;
+    }
+
+    public static void destroyLocalMasterNamespace() {
+        log.info("cleaning up master directory");
+        FileUtils.deleteQuietly(new File(YarnUtils.YARN_MASTER_DESTINATION));
+        FileUtils.deleteQuietly(new File(YarnUtils.YARN_MASTER_PROPERTIES));
+        FileUtils.deleteQuietly(new File(YarnUtils.YARN_CONTAINER_STAGING));
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/e38aa54b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/impl/yarn/YarnStatusProvider.java
----------------------------------------------------------------------
diff --git a/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/impl/yarn/YarnStatusProvider.java b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/impl/yarn/YarnStatusProvider.java
new file mode 100644
index 0000000..b4a13b9
--- /dev/null
+++ b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/impl/yarn/YarnStatusProvider.java
@@ -0,0 +1,67 @@
+package org.apache.helix.metamanager.impl.yarn;
+
+import java.util.Properties;
+
+import org.apache.helix.metamanager.StatusProviderService;
+import org.apache.helix.metamanager.impl.yarn.YarnContainerData.ContainerState;
+import org.apache.log4j.Logger;
+
+/**
+ * StatusProvider for YARN-based containers spawned via
+ * {@link YarnContainerProvider}. Reads {@link YarnDataProvider} meta data.
+ * Runnable and configurable service.
+ * 
+ */
+public class YarnStatusProvider implements StatusProviderService {
+
+    static final Logger       log = Logger.getLogger(YarnStatusProvider.class);
+
+    String                    yarndata;
+
+    ZookeeperYarnDataProvider yarnDataService;
+
+    public YarnStatusProvider() {
+        // left blank
+    }
+
+    public YarnStatusProvider(String yarndata) {
+        this.yarndata = yarndata;
+        this.yarnDataService = new ZookeeperYarnDataProvider(yarndata);
+    }
+
+    @Override
+    public void configure(Properties properties) throws Exception {
+        this.yarndata = properties.getProperty("yarndata");
+        this.yarnDataService = new ZookeeperYarnDataProvider(yarndata);
+    }
+
+    @Override
+    public void start() throws Exception {
+        yarnDataService = new ZookeeperYarnDataProvider(yarndata);
+        yarnDataService.start();
+    }
+
+    @Override
+    public void stop() throws Exception {
+        if (yarnDataService != null) {
+            yarnDataService.stop();
+            yarnDataService = null;
+        }
+    }
+
+    @Override
+    public boolean exists(String id) {
+        return yarnDataService.exists(id);
+    }
+
+    @Override
+    public boolean isHealthy(String id) {
+        try {
+            return yarnDataService.read(id).state == ContainerState.ACTIVE;
+        } catch (Exception e) {
+            log.warn(String.format("Could not get activity data of %s", id));
+            return false;
+        }
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/e38aa54b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/impl/yarn/YarnUtils.java
----------------------------------------------------------------------
diff --git a/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/impl/yarn/YarnUtils.java b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/impl/yarn/YarnUtils.java
new file mode 100644
index 0000000..fe093c8
--- /dev/null
+++ b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/impl/yarn/YarnUtils.java
@@ -0,0 +1,174 @@
+package org.apache.helix.metamanager.impl.yarn;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.io.Writer;
+import java.nio.charset.Charset;
+import java.util.Properties;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileContext;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.UnsupportedFileSystemException;
+import org.apache.hadoop.yarn.api.records.LocalResource;
+import org.apache.hadoop.yarn.api.records.LocalResourceType;
+import org.apache.hadoop.yarn.api.records.LocalResourceVisibility;
+import org.apache.hadoop.yarn.api.records.URL;
+import org.apache.hadoop.yarn.util.ConverterUtils;
+import org.apache.hadoop.yarn.util.Records;
+import org.apache.helix.metamanager.impl.yarn.YarnContainerData.ContainerState;
+import org.apache.log4j.Logger;
+
+import com.google.common.base.Preconditions;
+import com.google.common.io.Files;
+import com.google.gson.Gson;
+import com.google.gson.GsonBuilder;
+import com.google.gson.TypeAdapter;
+import com.google.gson.stream.JsonReader;
+import com.google.gson.stream.JsonToken;
+import com.google.gson.stream.JsonWriter;
+
+/**
+ * Utility for writing property files, transferring data via HDFS and
+ * serializing {@link YarnContainerData} for zookeeper.
+ * 
+ */
+class YarnUtils {
+
+    static final Logger log                         = Logger.getLogger(YarnUtils.class);
+
+    static final String YARN_MASTER_ARCHIVE_PATH    = "target/metamanager-assembly.tar.gz";
+    static final String YARN_MASTER_PATH            = "master/metamanager/bin/yarn-master-process.sh";
+    static final String YARN_MASTER_STAGING         = "master.tar.gz";
+    static final String YARN_MASTER_DESTINATION     = "master";
+    static final String YARN_MASTER_PROPERTIES      = "master.properties";
+    static final String YARN_CONTAINER_ARCHIVE_PATH = "target/metamanager-assembly.tar.gz";
+    static final String YARN_CONTAINER_STAGING      = "container.tar.gz";
+    static final String YARN_CONTAINER_PATH         = "container/metamanager/bin/yarn-container-process.sh";
+    static final String YARN_CONTAINER_DESTINATION  = "container";
+    static final String YARN_CONTAINER_PROPERTIES   = "container.properties";
+
+    static Gson         gson;
+    static {
+        GsonBuilder builder = new GsonBuilder();
+        builder.registerTypeAdapter(ContainerState.class, new ContainerStateAdapter());
+        builder.setPrettyPrinting();
+        gson = builder.create();
+    }
+
+    public static String toJson(YarnContainerData meta) {
+        return gson.toJson(meta);
+    }
+
+    public static YarnContainerData fromJson(String json) {
+        return gson.fromJson(json, YarnContainerData.class);
+    }
+
+    public static Properties getPropertiesFromPath(String path) throws IOException {
+        Properties properties = new Properties();
+        properties.load(new InputStreamReader(new FileInputStream(path)));
+        return properties;
+    }
+
+    public static File writePropertiesToTemp(Properties properties) throws IOException {
+        File tmpFile = File.createTempFile("provider", ".properties");
+        Writer writer = Files.newWriter(tmpFile, Charset.defaultCharset());
+        properties.store(writer, null);
+        writer.flush();
+        writer.close();
+        return tmpFile;
+    }
+
+    public static Path copyToHdfs(String source, String dest, String namespace, Configuration conf) throws IOException {
+        Path sourcePath = makeQualified(source);
+        Path destPath = makeQualified(conf.get(FileSystem.FS_DEFAULT_NAME_KEY) + "/" + namespace + "/" + dest);
+        log.debug(String.format("Copying '%s' to '%s'", sourcePath, destPath));
+
+        FileSystem fs = FileSystem.get(conf);
+        fs.copyFromLocalFile(false, true, sourcePath, destPath);
+        fs.close();
+        return destPath;
+    }
+
+    public static void destroyHdfsNamespace(String namespace, Configuration conf) throws IOException {
+        Path path = makeQualified(conf.get(FileSystem.FS_DEFAULT_NAME_KEY) + "/" + namespace);
+        log.debug(String.format("Deleting '%s'", path));
+
+        FileSystem fs = FileSystem.get(conf);
+        fs.delete(path, true);
+        fs.close();
+    }
+
+    public static LocalResource createHdfsResource(Path path, LocalResourceType type, Configuration conf) throws IOException {
+        FileSystem fs = FileSystem.get(conf);
+
+        URL url = ConverterUtils.getYarnUrlFromPath(path);
+
+        FileStatus status = fs.getFileStatus(path);
+
+        LocalResource resource = Records.newRecord(LocalResource.class);
+        resource.setResource(url);
+        resource.setSize(status.getLen());
+        resource.setTimestamp(status.getModificationTime());
+        resource.setType(type);
+        resource.setVisibility(LocalResourceVisibility.APPLICATION);
+
+        fs.close();
+
+        return resource;
+    }
+
+    static Path makeQualified(String path) throws UnsupportedFileSystemException {
+        return FileContext.getFileContext().makeQualified(new Path(path));
+    }
+
+    static class ContainerStateAdapter extends TypeAdapter<ContainerState> {
+        @Override
+        public ContainerState read(JsonReader reader) throws IOException {
+            if (reader.peek() == JsonToken.NULL) {
+                reader.nextNull();
+                return null;
+            }
+            return ContainerState.valueOf(reader.nextString());
+        }
+
+        @Override
+        public void write(JsonWriter writer, ContainerState value) throws IOException {
+            if (value == null) {
+                writer.nullValue();
+                return;
+            }
+            writer.value(value.name());
+        }
+    }
+
+    static YarnContainerProcessProperties createContainerProcessProperties(Properties properties) {
+        Preconditions.checkNotNull(properties);
+        YarnContainerProcessProperties yarnProp = new YarnContainerProcessProperties();
+        yarnProp.putAll(properties);
+        return yarnProp;
+    }
+
+    static YarnContainerProviderProperties createContainerProviderProperties(Properties properties) {
+        Preconditions.checkNotNull(properties);
+        YarnContainerProviderProperties yarnProp = new YarnContainerProviderProperties();
+        yarnProp.putAll(properties);
+        return yarnProp;
+    }
+
+    static YarnMasterProperties createMasterProperties(Properties properties) {
+        Preconditions.checkNotNull(properties);
+        YarnMasterProperties yarnProp = new YarnMasterProperties();
+        yarnProp.putAll(properties);
+        return yarnProp;
+    }
+
+    private YarnUtils() {
+        // left blank
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/e38aa54b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/impl/yarn/ZookeeperMetadataProvider.java
----------------------------------------------------------------------
diff --git a/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/impl/yarn/ZookeeperMetadataProvider.java b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/impl/yarn/ZookeeperMetadataProvider.java
new file mode 100644
index 0000000..79efd8c
--- /dev/null
+++ b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/impl/yarn/ZookeeperMetadataProvider.java
@@ -0,0 +1,116 @@
+package org.apache.helix.metamanager.impl.yarn;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Properties;
+
+import org.I0Itec.zkclient.ZkClient;
+import org.I0Itec.zkclient.exception.ZkException;
+import org.apache.helix.metamanager.Service;
+import org.apache.log4j.Logger;
+
+public class ZookeeperMetadataProvider implements MetadataProvider, Service {
+
+    static final Logger log                 = Logger.getLogger(ZookeeperMetadataProvider.class);
+
+    static final String CONTAINER_NAMESPACE = "containers";
+
+    static final String BASE_PATH           = "/" + CONTAINER_NAMESPACE;
+
+    static final int    META_TIMEOUT        = 5000;
+    static final long   POLL_INTERVAL       = 100;
+
+    String              metadata;
+
+    ZkClient            client;
+
+    public ZookeeperMetadataProvider() {
+        // left blank
+    }
+
+    public ZookeeperMetadataProvider(String metadataAddress) {
+        this.metadata = metadataAddress;
+    }
+
+    @Override
+    public void configure(Properties properties) throws Exception {
+        this.metadata = properties.getProperty("metadata");
+    }
+
+    @Override
+    public void start() {
+        log.debug(String.format("starting metadata service for '%s'", metadata));
+
+        client = new ZkClient(metadata, META_TIMEOUT, META_TIMEOUT);
+
+        client.createPersistent(BASE_PATH, true);
+    }
+
+    @Override
+    public void stop() {
+        log.debug(String.format("stopping metadata service for '%s'", metadata));
+        if (client != null) {
+            client.close();
+            client = null;
+        }
+    }
+
+    @Override
+    public boolean exists(String id) {
+        return client.exists(makePath(id));
+    }
+
+    @Override
+    public void create(ContainerMetadata meta) throws MetadataException {
+        try {
+            client.createEphemeral(makePath(meta.id), YarnUtils.toJson(meta));
+        } catch (ZkException e) {
+            throw new MetadataException(e);
+        }
+    }
+
+    @Override
+    public ContainerMetadata read(String id) throws MetadataException {
+        try {
+            return YarnUtils.fromJson(client.<String> readData(makePath(id)));
+        } catch (ZkException e) {
+            throw new MetadataException(e);
+        }
+    }
+
+    @Override
+    public Collection<ContainerMetadata> readAll() throws MetadataException {
+        try {
+            Collection<ContainerMetadata> metadata = new ArrayList<ContainerMetadata>();
+            for (String id : client.getChildren(BASE_PATH)) {
+                metadata.add(YarnUtils.fromJson(client.<String> readData(makePath(id))));
+            }
+            return metadata;
+        } catch (ZkException e) {
+            throw new MetadataException(e);
+        }
+    }
+
+    @Override
+    public void update(ContainerMetadata meta) throws MetadataException {
+        try {
+            client.writeData(makePath(meta.id), YarnUtils.toJson(meta));
+        } catch (ZkException e) {
+            throw new MetadataException(e);
+        }
+    }
+
+    @Override
+    public void delete(String id) throws MetadataException {
+        try {
+            client.delete(makePath(id));
+        } catch (ZkException e) {
+            throw new MetadataException(e);
+        }
+    }
+
+    String makePath(String containerId) {
+        return BASE_PATH + "/" + containerId;
+    }
+
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/e38aa54b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/impl/yarn/ZookeeperMetadataService.java
----------------------------------------------------------------------
diff --git a/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/impl/yarn/ZookeeperMetadataService.java b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/impl/yarn/ZookeeperMetadataService.java
new file mode 100644
index 0000000..b0e150a
--- /dev/null
+++ b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/impl/yarn/ZookeeperMetadataService.java
@@ -0,0 +1,102 @@
+package org.apache.helix.metamanager.impl.yarn;
+
+import java.util.ArrayList;
+import java.util.Collection;
+
+import org.I0Itec.zkclient.ZkClient;
+import org.I0Itec.zkclient.exception.ZkException;
+import org.apache.log4j.Logger;
+
+public class ZookeeperMetadataService implements MetadataService {
+
+    static final Logger log                 = Logger.getLogger(ZookeeperMetadataService.class);
+
+    static final String CONTAINER_NAMESPACE = "containers";
+
+    static final String BASE_PATH           = "/" + CONTAINER_NAMESPACE;
+
+    static final long   POLL_INTERVAL       = 100;
+
+    final String        metadataAddress;
+
+    ZkClient            client;
+
+    public ZookeeperMetadataService(String metadataAddress) {
+        this.metadataAddress = metadataAddress;
+    }
+
+    public void startService() {
+        log.debug(String.format("starting metadata service for '%s'", metadataAddress));
+
+        client = new ZkClient(metadataAddress);
+
+        client.createPersistent(BASE_PATH, true);
+    }
+
+    public void stopService() {
+        log.debug(String.format("stopping metadata service for '%s'", metadataAddress));
+        if (client != null) {
+            client.close();
+            client = null;
+        }
+    }
+
+    @Override
+    public boolean exists(String id) {
+        return client.exists(makePath(id));
+    }
+
+    @Override
+    public void create(ContainerMetadata meta) throws MetadataServiceException {
+        try {
+            client.createPersistent(makePath(meta.id), YarnUtils.toJson(meta));
+        } catch (ZkException e) {
+            throw new MetadataServiceException(e);
+        }
+    }
+
+    @Override
+    public ContainerMetadata read(String id) throws MetadataServiceException {
+        try {
+            return YarnUtils.fromJson(client.<String> readData(makePath(id)));
+        } catch (ZkException e) {
+            throw new MetadataServiceException(e);
+        }
+    }
+
+    @Override
+    public Collection<ContainerMetadata> readAll() throws MetadataServiceException {
+        try {
+            Collection<ContainerMetadata> metadata = new ArrayList<ContainerMetadata>();
+            for (String id : client.getChildren(BASE_PATH)) {
+                metadata.add(YarnUtils.fromJson(client.<String> readData(makePath(id))));
+            }
+            return metadata;
+        } catch (ZkException e) {
+            throw new MetadataServiceException(e);
+        }
+    }
+
+    @Override
+    public void update(ContainerMetadata meta) throws MetadataServiceException {
+        try {
+            client.writeData(makePath(meta.id), YarnUtils.toJson(meta));
+        } catch (ZkException e) {
+            throw new MetadataServiceException(e);
+        }
+    }
+
+    @Override
+    public void delete(String id) throws MetadataServiceException {
+        try {
+            client.delete(makePath(id));
+        } catch (ZkException e) {
+            throw new MetadataServiceException(e);
+        }
+    }
+
+    String makePath(String containerId) {
+        return BASE_PATH + "/" + containerId;
+    }
+
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/e38aa54b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/impl/yarn/ZookeeperYarnDataProvider.java
----------------------------------------------------------------------
diff --git a/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/impl/yarn/ZookeeperYarnDataProvider.java b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/impl/yarn/ZookeeperYarnDataProvider.java
new file mode 100644
index 0000000..32af837
--- /dev/null
+++ b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/impl/yarn/ZookeeperYarnDataProvider.java
@@ -0,0 +1,100 @@
+package org.apache.helix.metamanager.impl.yarn;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Properties;
+
+import org.I0Itec.zkclient.ZkClient;
+import org.apache.helix.metamanager.Service;
+import org.apache.log4j.Logger;
+
+/**
+ * Configurable and runnable service for {@link YarnDataProvider} based on
+ * zookeeper.
+ * 
+ */
+public class ZookeeperYarnDataProvider implements YarnDataProvider, Service {
+
+    static final Logger log                 = Logger.getLogger(ZookeeperYarnDataProvider.class);
+
+    static final String CONTAINER_NAMESPACE = "containers";
+
+    static final String BASE_PATH           = "/" + CONTAINER_NAMESPACE;
+
+    static final int    META_TIMEOUT        = 5000;
+    static final long   POLL_INTERVAL       = 100;
+
+    String              yarndata;
+
+    ZkClient            client;
+
+    public ZookeeperYarnDataProvider() {
+        // left blank
+    }
+
+    public ZookeeperYarnDataProvider(String yarndataAddress) {
+        this.yarndata = yarndataAddress;
+    }
+
+    @Override
+    public void configure(Properties properties) throws Exception {
+        this.yarndata = properties.getProperty("yarndata");
+    }
+
+    @Override
+    public void start() {
+        log.debug(String.format("starting yarndata service for '%s'", yarndata));
+
+        client = new ZkClient(yarndata, META_TIMEOUT, META_TIMEOUT);
+
+        client.createPersistent(BASE_PATH, true);
+    }
+
+    @Override
+    public void stop() {
+        log.debug(String.format("stopping yarndata service for '%s'", yarndata));
+        if (client != null) {
+            client.close();
+            client = null;
+        }
+    }
+
+    @Override
+    public boolean exists(String id) {
+        return client.exists(makePath(id));
+    }
+
+    @Override
+    public void create(YarnContainerData meta) throws Exception {
+        client.createEphemeral(makePath(meta.id), YarnUtils.toJson(meta));
+    }
+
+    @Override
+    public YarnContainerData read(String id) throws Exception {
+        return YarnUtils.fromJson(client.<String> readData(makePath(id)));
+    }
+
+    @Override
+    public Collection<YarnContainerData> readAll() throws Exception {
+        Collection<YarnContainerData> yarndata = new ArrayList<YarnContainerData>();
+        for (String id : client.getChildren(BASE_PATH)) {
+            yarndata.add(YarnUtils.fromJson(client.<String> readData(makePath(id))));
+        }
+        return yarndata;
+    }
+
+    @Override
+    public void update(YarnContainerData meta) throws Exception {
+        client.writeData(makePath(meta.id), YarnUtils.toJson(meta));
+    }
+
+    @Override
+    public void delete(String id) throws Exception {
+        client.delete(makePath(id));
+    }
+
+    String makePath(String containerId) {
+        return BASE_PATH + "/" + containerId;
+    }
+
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/e38aa54b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/managed/ContainerProcess.java
----------------------------------------------------------------------
diff --git a/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/managed/ContainerProcess.java b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/managed/ContainerProcess.java
new file mode 100644
index 0000000..11fb75d
--- /dev/null
+++ b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/managed/ContainerProcess.java
@@ -0,0 +1,85 @@
+package org.apache.helix.metamanager.managed;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import org.apache.helix.HelixManager;
+import org.apache.helix.HelixManagerFactory;
+import org.apache.helix.InstanceType;
+import org.apache.log4j.Logger;
+
+public class ContainerProcess
+{
+	static final Logger log = Logger.getLogger(ContainerProcess.class);
+	
+  private String clusterName;
+  private String zkAddress;
+  private String instanceName;
+  private HelixManager participantManager;
+
+ public ContainerProcess(String clusterName, String zkAddress, String instanceName)
+  {
+    this.clusterName = clusterName;
+    this.zkAddress = zkAddress;
+    this.instanceName = instanceName;
+
+  }
+
+  public void start() throws Exception
+  {
+    log.info("STARTING "+ instanceName);
+    participantManager = HelixManagerFactory.getZKHelixManager(clusterName,
+        instanceName, InstanceType.PARTICIPANT, zkAddress);
+    participantManager.getStateMachineEngine().registerStateModelFactory(
+        "MasterSlave", new ManagedFactory());
+    participantManager.connect();
+    log.info("STARTED "+ instanceName);
+
+  }
+
+  public void stop()
+  {
+    if (participantManager != null)
+    {
+      participantManager.disconnect();
+    }
+  }
+
+  public static void main(String[] args) throws Exception
+  {
+    final String zkAddress = args[0];
+    final String clusterName = args[1];
+    final String instanceName = args[2];
+
+    // Give a unique id to each process, most commonly used format hostname_port
+    final ContainerProcess managerProcess = new ContainerProcess(clusterName, zkAddress,
+        instanceName);
+    Runtime.getRuntime().addShutdownHook(new Thread()
+    {
+      @Override
+      public void run()
+      {
+    	  log.info("Shutting down " + instanceName);
+        managerProcess.stop();
+      }
+    });
+    managerProcess.start();
+    Thread.currentThread().join();
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/e38aa54b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/managed/HelixClusterAdmin.java
----------------------------------------------------------------------
diff --git a/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/managed/HelixClusterAdmin.java b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/managed/HelixClusterAdmin.java
new file mode 100644
index 0000000..f33c09c
--- /dev/null
+++ b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/managed/HelixClusterAdmin.java
@@ -0,0 +1,42 @@
+package org.apache.helix.metamanager.managed;
+
+import org.apache.helix.HelixAdmin;
+import org.apache.helix.metamanager.ClusterAdmin;
+import org.apache.helix.model.InstanceConfig;
+import org.apache.log4j.Logger;
+
+public class HelixClusterAdmin implements ClusterAdmin {
+	
+	static final Logger log = Logger.getLogger(HelixClusterAdmin.class);
+
+	final String clusterName;
+	final String resourceName;
+	final int replica;
+	final HelixAdmin admin;
+	
+	public HelixClusterAdmin(String clusterName, String resourceName,
+			int replica, HelixAdmin admin) {
+		this.clusterName = clusterName;
+		this.resourceName = resourceName;
+		this.replica = replica;
+		this.admin = admin;
+	}
+
+	@Override
+	public synchronized void addInstance(String connection) {
+		log.debug(String.format("injecting instance %s in cluster %s", connection, clusterName));
+		admin.addInstance(clusterName, new InstanceConfig(connection));
+	}
+
+	@Override
+	public synchronized void removeInstance(String connection) {
+		log.debug(String.format("removing instance %s from cluster %s", connection, clusterName));
+		admin.dropInstance(clusterName, new InstanceConfig(connection));
+	}
+
+	@Override
+	public void rebalance() {
+		admin.rebalance(clusterName, resourceName, replica);
+	}
+	
+}

http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/e38aa54b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/managed/LocalClusterManager.java
----------------------------------------------------------------------
diff --git a/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/managed/LocalClusterManager.java b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/managed/LocalClusterManager.java
new file mode 100644
index 0000000..2bb64de
--- /dev/null
+++ b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/managed/LocalClusterManager.java
@@ -0,0 +1,42 @@
+package org.apache.helix.metamanager.managed;
+
+import org.apache.helix.HelixAdmin;
+import org.apache.helix.metamanager.ClusterAdmin;
+import org.apache.helix.model.InstanceConfig;
+import org.apache.log4j.Logger;
+
+public class LocalClusterManager implements ClusterAdmin {
+	
+	static final Logger log = Logger.getLogger(LocalClusterManager.class);
+
+	final String clusterName;
+	final String resourceName;
+	final int replica;
+	final HelixAdmin admin;
+	
+	public LocalClusterManager(String clusterName, String resourceName,
+			int replica, HelixAdmin admin) {
+		this.clusterName = clusterName;
+		this.resourceName = resourceName;
+		this.replica = replica;
+		this.admin = admin;
+	}
+
+	@Override
+	public synchronized void addInstance(String connection) {
+		log.debug(String.format("injecting instance %s in cluster %s", connection, clusterName));
+		admin.addInstance(clusterName, new InstanceConfig(connection));
+	}
+
+	@Override
+	public synchronized void removeInstance(String connection) {
+		log.debug(String.format("removing instance %s from cluster %s", connection, clusterName));
+		admin.dropInstance(clusterName, new InstanceConfig(connection));
+	}
+
+	@Override
+	public void rebalance() {
+		admin.rebalance(clusterName, resourceName, replica);
+	}
+	
+}

http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/e38aa54b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/managed/LocalContainerProvider.java
----------------------------------------------------------------------
diff --git a/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/managed/LocalContainerProvider.java b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/managed/LocalContainerProvider.java
new file mode 100644
index 0000000..6c8eec0
--- /dev/null
+++ b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/managed/LocalContainerProvider.java
@@ -0,0 +1,87 @@
+package org.apache.helix.metamanager.managed;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+
+import org.apache.helix.metamanager.ClusterContainerProvider;
+import org.apache.log4j.Logger;
+
+public class LocalContainerProvider implements ClusterContainerProvider {
+
+	static final Logger log = Logger.getLogger(LocalContainerProvider.class);
+	
+	static final String REQUIRED_TYPE = "container";
+	
+	// global view of processes required
+	static final Object staticLock = new Object();
+	static final Map<String, LocalProcess> processes = new HashMap<String, LocalProcess>();
+	
+	int connectionCounter = 0;
+	
+	final String zkAddress;
+	final String clusterName;
+	final String providerName;
+	
+	public LocalContainerProvider(String zkAddress, String clusterName, String providerName) {
+		this.zkAddress = zkAddress;
+		this.clusterName = clusterName;
+		this.providerName = providerName;
+	}
+
+	@Override
+	public void create(String id, String type) throws Exception {
+		synchronized (staticLock) {	
+			if(processes.containsKey(id))
+				throw new IllegalArgumentException(String.format("Process '%s' already exists", id));
+			
+			if(!type.equals(REQUIRED_TYPE))
+				throw new IllegalArgumentException(String.format("Type '%s' not supported", type));
+			
+			log.info(String.format("Running container '%s' (zkAddress='%s', clusterName='%s')", id, zkAddress, clusterName));
+			
+			ManagedProcess process = new ManagedProcess(clusterName, zkAddress, id);
+			process.start();
+		
+			processes.put(id, new LocalProcess(id, providerName, process));
+			
+		}
+	}
+	
+	@Override
+	public void destroy(String id) throws Exception {
+		synchronized (staticLock) {	
+			if(!processes.containsKey(id))
+				throw new IllegalArgumentException(String.format("Process '%s' does not exists", id));
+			
+			log.info(String.format("Destroying container '%s'", id));
+			
+			LocalProcess local = processes.remove(id);
+			
+			local.process.stop();
+		}
+	}
+	
+	@Override
+	public void destroyAll() {
+		synchronized (staticLock) {	
+			log.info("Destroying all processes");
+			for(String id : new HashSet<String>(processes.keySet())) {
+				try { destroy(id); } catch (Exception ignore) {}
+			}
+		}
+	}
+	
+	static class LocalProcess {
+		final String id;
+		final String owner;
+		final ManagedProcess process;
+		
+		public LocalProcess(String id, String owner, ManagedProcess process) {
+			this.id = id;
+			this.owner = owner;
+			this.process = process;
+		}
+	}
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/e38aa54b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/managed/LocalProcessProvider.java
----------------------------------------------------------------------
diff --git a/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/managed/LocalProcessProvider.java b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/managed/LocalProcessProvider.java
new file mode 100644
index 0000000..01e3ab6
--- /dev/null
+++ b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/managed/LocalProcessProvider.java
@@ -0,0 +1,100 @@
+package org.apache.helix.metamanager.managed;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+
+import org.apache.helix.metamanager.ClusterContainerProvider;
+import org.apache.log4j.Logger;
+
+public class LocalProcessProvider implements ClusterContainerProvider {
+
+	static final Logger log = Logger.getLogger(LocalProcessProvider.class);
+	
+	static final String REQUIRED_TYPE = "container";
+	
+	Map<String, ManagedProcess> processes = new HashMap<String, ManagedProcess>();
+	Map<String, String> id2connection = new HashMap<String, String>();
+
+	int connectionCounter = 0;
+	
+	final String zkAddress;
+	final String clusterName;
+	final int basePort;
+	
+	public LocalProcessProvider(String zkAddress, String clusterName, int basePort) {
+		this.zkAddress = zkAddress;
+		this.clusterName = clusterName;
+		this.basePort = basePort;
+	}
+
+	@Override
+	public synchronized String create(String id, String type) throws Exception {
+		if(processes.containsKey(id))
+			throw new IllegalArgumentException(String.format("Process '%s' already exists", id));
+		
+		if(!type.equals(REQUIRED_TYPE))
+			throw new IllegalArgumentException(String.format("Type '%s' not supported", type));
+		
+		String connection = "localhost_" + (basePort + connectionCounter);
+		connectionCounter++;
+		
+		log.info(String.format("Running container '%s' (zkAddress='%s', clusterName='%s', connection='%s')", id, zkAddress, clusterName, connection));
+		
+		ManagedProcess p = new ManagedProcess(clusterName, zkAddress, connection);
+		
+		processes.put(id, p);
+		id2connection.put(id, connection);
+		
+		return connection;
+	}
+	
+	public synchronized void start(String id) throws Exception {
+		if(!processes.containsKey(id))
+			throw new IllegalArgumentException(String.format("Process '%s' does not exists", id));
+		
+		log.info(String.format("Starting container '%s'", id));
+		
+		ManagedProcess p = processes.get(id);
+		
+		p.start();
+	}
+	
+	public synchronized void stop(String id) throws Exception {
+		if(!processes.containsKey(id))
+			throw new IllegalArgumentException(String.format("Process '%s' does not exists", id));
+		
+		log.info(String.format("Stopping container '%s'", id));
+		
+		ManagedProcess p = processes.get(id);
+		
+		p.stop();
+	}
+
+	@Override
+	public synchronized String destroy(String id) throws Exception {
+		if(!processes.containsKey(id))
+			throw new IllegalArgumentException(String.format("Process '%s' does not exists", id));
+		
+		log.info(String.format("Destroying container '%s'", id));
+		
+		String connection = id2connection.get(id);
+
+		processes.remove(id);
+		id2connection.remove(id);
+		
+		return connection;
+	}
+	
+	public synchronized void destroyAll() {
+		log.info("Destroying all processes");
+		for(String id : new HashSet<String>(processes.keySet())) {
+			try {
+				destroy(id);
+			} catch (Exception ignore) {
+				// ignore
+			}
+		}
+	}
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/e38aa54b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/managed/LocalStatusProvider.java
----------------------------------------------------------------------
diff --git a/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/managed/LocalStatusProvider.java b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/managed/LocalStatusProvider.java
new file mode 100644
index 0000000..54f040f
--- /dev/null
+++ b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/managed/LocalStatusProvider.java
@@ -0,0 +1,22 @@
+package org.apache.helix.metamanager.managed;
+
+import org.apache.helix.metamanager.ClusterStatusProvider;
+
+public class LocalStatusProvider implements ClusterStatusProvider {
+
+	int targetContainerCount;
+	
+	public LocalStatusProvider(int targetContainerCount) {
+		this.targetContainerCount = targetContainerCount;
+	}
+
+	@Override
+	public int getTargetContainerCount(String type) {
+		return targetContainerCount;
+	}
+
+	public void setTargetContainerCount(int targetContainerCount) {
+		this.targetContainerCount = targetContainerCount;
+	}
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/e38aa54b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/managed/Managed.java
----------------------------------------------------------------------
diff --git a/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/managed/Managed.java b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/managed/Managed.java
new file mode 100644
index 0000000..1e03103
--- /dev/null
+++ b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/managed/Managed.java
@@ -0,0 +1,64 @@
+package org.apache.helix.metamanager.managed;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import org.apache.helix.NotificationContext;
+import org.apache.helix.model.Message;
+import org.apache.helix.participant.statemachine.StateModel;
+import org.apache.helix.participant.statemachine.StateModelInfo;
+import org.apache.helix.participant.statemachine.Transition;
+import org.apache.log4j.Logger;
+
+@StateModelInfo(initialState = "OFFLINE", states = { "OFFLINE", "SLAVE", "MASTER", "DROPPED" })
+public class Managed extends StateModel {
+	
+	static final Logger log = Logger.getLogger(Managed.class);
+	
+	@Transition(from = "OFFLINE", to = "SLAVE")
+	public void offlineToSlave(Message m, NotificationContext context) {
+		log.trace(String.format("%s transitioning from OFFLINE to SLAVE",
+				context.getManager().getInstanceName()));
+	}
+
+	@Transition(from = "SLAVE", to = "OFFLINE")
+	public void slaveToOffline(Message m, NotificationContext context) {
+		log.trace(String.format("%s transitioning from SLAVE to OFFLINE",
+				context.getManager().getInstanceName()));
+	}
+
+	@Transition(from = "SLAVE", to = "MASTER")
+	public void slaveToMaster(Message m, NotificationContext context) {
+		log.trace(String.format("%s transitioning from SLAVE to MASTER",
+				context.getManager().getInstanceName()));
+	}
+
+	@Transition(from = "MASTER", to = "SLAVE")
+	public void masterToSlave(Message m, NotificationContext context) {
+		log.trace(String.format("%s transitioning from MASTER to SLAVE",
+				context.getManager().getInstanceName()));
+	}
+
+	@Transition(from = "OFFLINE", to = "DROPPED")
+	public void offlineToDropped(Message m, NotificationContext context) {
+		log.trace(String.format("%s transitioning from OFFLINE to DROPPED",
+				context.getManager().getInstanceName()));
+	}
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/e38aa54b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/managed/ManagedFactory.java
----------------------------------------------------------------------
diff --git a/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/managed/ManagedFactory.java b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/managed/ManagedFactory.java
new file mode 100644
index 0000000..f51d9c0
--- /dev/null
+++ b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/managed/ManagedFactory.java
@@ -0,0 +1,30 @@
+package org.apache.helix.metamanager.managed;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import org.apache.helix.participant.statemachine.StateModelFactory;
+
+public class ManagedFactory extends StateModelFactory<Managed> {
+
+	@Override
+	public Managed createNewStateModel(String partitionName) {
+		return new Managed();
+	}
+}

http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/e38aa54b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/managed/ManagedProcess.java
----------------------------------------------------------------------
diff --git a/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/managed/ManagedProcess.java b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/managed/ManagedProcess.java
new file mode 100644
index 0000000..387c459
--- /dev/null
+++ b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/managed/ManagedProcess.java
@@ -0,0 +1,85 @@
+package org.apache.helix.metamanager.managed;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import org.apache.helix.HelixManager;
+import org.apache.helix.HelixManagerFactory;
+import org.apache.helix.InstanceType;
+import org.apache.log4j.Logger;
+
+public class ManagedProcess
+{
+	static final Logger log = Logger.getLogger(ManagedProcess.class);
+	
+  private String clusterName;
+  private String zkAddress;
+  private String instanceName;
+  private HelixManager participantManager;
+
+ public ManagedProcess(String clusterName, String zkAddress, String instanceName)
+  {
+    this.clusterName = clusterName;
+    this.zkAddress = zkAddress;
+    this.instanceName = instanceName;
+
+  }
+
+  public void start() throws Exception
+  {
+    log.info("STARTING "+ instanceName);
+    participantManager = HelixManagerFactory.getZKHelixManager(clusterName,
+        instanceName, InstanceType.PARTICIPANT, zkAddress);
+    participantManager.getStateMachineEngine().registerStateModelFactory(
+        "MasterSlave", new ManagedFactory());
+    participantManager.connect();
+    log.info("STARTED "+ instanceName);
+
+  }
+
+  public void stop()
+  {
+    if (participantManager != null)
+    {
+      participantManager.disconnect();
+    }
+  }
+
+  public static void main(String[] args) throws Exception
+  {
+    final String zkAddress = args[0];
+    final String clusterName = args[1];
+    final String instanceName = args[2];
+
+    // Give a unique id to each process, most commonly used format hostname_port
+    final ManagedProcess managerProcess = new ManagedProcess(clusterName, zkAddress,
+        instanceName);
+    Runtime.getRuntime().addShutdownHook(new Thread()
+    {
+      @Override
+      public void run()
+      {
+    	  log.info("Shutting down " + instanceName);
+        managerProcess.stop();
+      }
+    });
+    managerProcess.start();
+    Thread.currentThread().join();
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/e38aa54b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/managed/ShellContainerProvider.java
----------------------------------------------------------------------
diff --git a/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/managed/ShellContainerProvider.java b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/managed/ShellContainerProvider.java
new file mode 100644
index 0000000..107f2c6
--- /dev/null
+++ b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/managed/ShellContainerProvider.java
@@ -0,0 +1,85 @@
+package org.apache.helix.metamanager.managed;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+
+import org.apache.helix.metamanager.ClusterContainerProvider;
+import org.apache.log4j.Logger;
+
+public class ShellContainerProvider implements ClusterContainerProvider {
+
+	static final Logger log = Logger.getLogger(ShellContainerProvider.class);
+	
+	static final String REQUIRED_TYPE = "container";
+	static final String RUN_COMMAND = "/bin/sh";
+	
+	// global view of processes required
+	static final Object staticLock = new Object();
+	static final Map<String, ShellProcess> processes = new HashMap<String, ShellProcess>();
+
+	final String zkAddress;
+	final String clusterName;
+	final String command;
+	final String providerName;
+	
+	public ShellContainerProvider(String zkAddress, String clusterName, String owner, String command) {
+		this.zkAddress = zkAddress;
+		this.clusterName = clusterName;
+		this.command = command;
+		this.providerName = owner;
+	}
+
+	@Override
+	public void create(String id, String type) throws Exception {
+		synchronized (staticLock) {
+			if(processes.containsKey(id))
+				throw new IllegalArgumentException(String.format("Process '%s' already exists", id));
+			
+			if(!type.equals(REQUIRED_TYPE))
+				throw new IllegalArgumentException(String.format("Type '%s' not supported", type));
+			
+			log.info(String.format("Running container '%s' (zkAddress='%s', clusterName='%s', command='%s')", id, zkAddress, clusterName, command));
+			
+			ProcessBuilder builder = new ProcessBuilder(RUN_COMMAND, command, zkAddress, clusterName, id);
+			Process process = builder.start();
+			
+			processes.put(id, new ShellProcess(id, providerName, process));
+		}
+	}
+	
+	@Override
+	public void destroy(String id) throws Exception {
+		synchronized (staticLock) {
+			if(!processes.containsKey(id))
+				throw new IllegalArgumentException(String.format("Process '%s' does not exists", id));
+			
+			log.info(String.format("Destroying container '%s'", id));
+			
+			ShellProcess shell = processes.remove(id);
+			shell.process.destroy();
+		}
+	}
+	
+	@Override
+	public void destroyAll() {
+		synchronized (staticLock) {
+			log.info("Destroying all processes");
+			for(ShellProcess process : new HashSet<ShellProcess>(processes.values())) {
+				try { destroy(process.id); } catch (Exception ignore) {}
+			}
+		}
+	}
+	
+	static class ShellProcess {
+		final String id;
+		final String owner;
+		final Process process;
+
+		public ShellProcess(String id, String owner, Process process) {
+			this.id = id;
+			this.owner = owner;
+			this.process = process;
+		}		
+	}
+}

http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/e38aa54b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/managed/ShellProcessProvider.java
----------------------------------------------------------------------
diff --git a/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/managed/ShellProcessProvider.java b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/managed/ShellProcessProvider.java
new file mode 100644
index 0000000..0def0f5
--- /dev/null
+++ b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/managed/ShellProcessProvider.java
@@ -0,0 +1,148 @@
+package org.apache.helix.metamanager.managed;
+
+import java.lang.reflect.Field;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+
+import org.apache.helix.metamanager.ClusterContainerProvider;
+import org.apache.log4j.Logger;
+
+public class ShellProcessProvider implements ClusterContainerProvider {
+
+	static final Logger log = Logger.getLogger(ShellProcessProvider.class);
+	
+	static final String REQUIRED_TYPE = "container";
+	static final String RUN_COMMAND = "/bin/sh";
+	static final String KILL_COMMAND = "kill -s SIGINT %d";
+	
+	Map<String, ProcessBuilder> builders = new HashMap<String, ProcessBuilder>();
+	Map<String, Process> processes = new HashMap<String, Process>();
+	Map<String, String> id2connection = new HashMap<String, String>();
+
+	int connectionCounter = 0;
+	
+	final String zkAddress;
+	final String clusterName;
+	final int basePort;
+	final String command;
+	
+	public ShellProcessProvider(String zkAddress, String clusterName, int basePort, String command) {
+		this.zkAddress = zkAddress;
+		this.clusterName = clusterName;
+		this.basePort = basePort;
+		this.command = command;
+	}
+
+	@Override
+	public synchronized String create(String id, String type) throws Exception {
+		if(builders.containsKey(id))
+			throw new IllegalArgumentException(String.format("Process '%s' already exists", id));
+		
+		if(!type.equals(REQUIRED_TYPE))
+			throw new IllegalArgumentException(String.format("Type '%s' not supported", type));
+		
+		String connection = "localhost_" + (basePort + connectionCounter);
+		connectionCounter++;
+		
+		log.info(String.format("Running container '%s' (zkAddress='%s', clusterName='%s', connection='%s', command='%s')", id, zkAddress, clusterName, connection, command));
+		
+		ProcessBuilder builder = new ProcessBuilder(RUN_COMMAND, command, zkAddress, clusterName, connection);
+		
+		builders.put(id, builder);
+		id2connection.put(id, connection);
+		
+		return connection;
+	}
+	
+	public synchronized void start(String id) throws Exception {
+		if(!builders.containsKey(id))
+			throw new IllegalArgumentException(String.format("Process '%s' does not exists", id));
+		
+		if(processes.containsKey(id))
+			throw new IllegalArgumentException(String.format("Process '%s' already running", id));
+		
+		log.info(String.format("Starting container '%s'", id));
+		
+		Process p = builders.get(id).start();
+		
+		processes.put(id, p);
+	}
+	
+	public synchronized void stop(String id) throws Exception {
+		if(!builders.containsKey(id))
+			throw new IllegalArgumentException(String.format("Process '%s' does not exists", id));
+		
+		if(!processes.containsKey(id))
+			throw new IllegalArgumentException(String.format("Process '%s' not running", id));
+		
+		log.info(String.format("Stopping container '%s'", id));
+		
+		Process p = processes.get(id);
+		
+		int pid = getUnixPID(p);
+		Runtime.getRuntime().exec(String.format(KILL_COMMAND, pid));
+		
+		int retVal = p.waitFor();
+		if(retVal != 130) {
+			log.warn(String.format("Process %d returned %d (should be 130, SIGINT)", pid, retVal));
+		}
+		
+		processes.remove(id);
+		
+	}
+
+	@Override
+	public synchronized String destroy(String id) throws Exception {
+		if(!builders.containsKey(id))
+			throw new IllegalArgumentException(String.format("Process '%s' does not exists", id));
+		
+		log.info(String.format("Destroying container '%s'", id));
+		
+		if(processes.containsKey(id)) {
+			log.warn(String.format("Forcibly terminating running container '%s'", id));
+			processes.get(id).destroy();
+			processes.remove(id);
+		}
+		
+		String connection = id2connection.get(id);
+
+		builders.remove(id);
+		id2connection.remove(id);
+		
+		return connection;
+	}
+	
+	public synchronized void destroyAll() {
+		log.info("Destroying all processes");
+		for(String id : new HashSet<String>(processes.keySet())) {
+			try {
+				destroy(id);
+			} catch (Exception ignore) {
+				// ignore
+			}
+		}
+	}
+	
+	// TODO get PID independently of platform
+    static int getUnixPID(Process process) throws IllegalArgumentException, IllegalAccessException, NoSuchFieldException {
+        if (process.getClass().getName().equals("java.lang.UNIXProcess")) {
+            Class<?> proc = process.getClass();
+            Field field = proc.getDeclaredField("pid");
+            Object value = getFieldValue(field, process);
+            return ((Integer) value).intValue();
+        } else {
+            throw new IllegalArgumentException("Not a UNIXProcess");
+        }
+    }
+    
+    static Object getFieldValue(Field field, Object object) throws IllegalArgumentException, IllegalAccessException {
+    	Object value;
+    	boolean accessible = field.isAccessible();
+    	field.setAccessible(true);
+    	value = field.get(object);
+    	field.setAccessible(accessible);
+    	return value;
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-helix/blob/e38aa54b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/managed/YarnContainerProvider.java
----------------------------------------------------------------------
diff --git a/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/managed/YarnContainerProvider.java b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/managed/YarnContainerProvider.java
new file mode 100644
index 0000000..629788e
--- /dev/null
+++ b/recipes/meta-cluster-manager/src/main/java/org/apache/helix/metamanager/managed/YarnContainerProvider.java
@@ -0,0 +1,37 @@
+package org.apache.helix.metamanager.managed;
+
+import org.apache.helix.metamanager.ClusterContainerProvider;
+
+public class YarnContainerProvider implements ClusterContainerProvider {
+
+	@Override
+	public String create(String id, String type) throws Exception {
+		// TODO Auto-generated method stub
+		return null;
+	}
+
+	@Override
+	public String destroy(String id) throws Exception {
+		// TODO Auto-generated method stub
+		return null;
+	}
+
+	@Override
+	public void start(String id) throws Exception {
+		// TODO Auto-generated method stub
+
+	}
+
+	@Override
+	public void stop(String id) throws Exception {
+		// TODO Auto-generated method stub
+
+	}
+
+	@Override
+	public void destroyAll() {
+		// TODO Auto-generated method stub
+
+	}
+
+}


Mime
View raw message