You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@drill.apache.org by ar...@apache.org on 2018/03/04 17:13:44 UTC
[04/12] drill git commit: DRILL-1170: YARN integration for Drill
http://git-wip-us.apache.org/repos/asf/drill/blob/f2ac8749/drill-yarn/src/main/java/org/apache/drill/yarn/client/StopCommand.java
----------------------------------------------------------------------
diff --git a/drill-yarn/src/main/java/org/apache/drill/yarn/client/StopCommand.java b/drill-yarn/src/main/java/org/apache/drill/yarn/client/StopCommand.java
new file mode 100644
index 0000000..95f7bf3
--- /dev/null
+++ b/drill-yarn/src/main/java/org/apache/drill/yarn/client/StopCommand.java
@@ -0,0 +1,223 @@
+/*
+ * 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.
+ */
+package org.apache.drill.yarn.client;
+
+import org.apache.drill.yarn.core.DoYUtil;
+import org.apache.drill.yarn.core.DrillOnYarnConfig;
+import org.apache.drill.yarn.core.YarnClientException;
+import org.apache.drill.yarn.core.YarnRMClient;
+import org.apache.hadoop.yarn.api.records.YarnApplicationState;
+
+import com.typesafe.config.Config;
+
+/**
+ * Perform a semi-graceful shutdown of the Drill-on-YARN AM. We send a message
+ * to the AM to request shutdown because the YARN-provided message just kills
+ * the AM. (There seems to be no way to get YARN to call its own
+ * AMRMClientAsync.CallbackHandler.onShutdownRequest message.) The AM, however,
+ * cannot gracefully shut down the drill-bits because Drill itself has no
+ * graceful shutdown. But, at least this technique gives the AM a fighting
+ * chance to do graceful shutdown in the future.
+ */
+
+public class StopCommand extends ClientCommand {
+ /**
+ * Poll the YARN RM to check the stop status of the AM. Periodically poll,
+ * waiting to get an app state that indicates app completion.
+ */
+
+ private static class StopMonitor {
+ StatusCommand.Reporter reporter;
+ private YarnApplicationState state;
+ private int pollWaitSec;
+ private int shutdownWaitSec;
+
+ StopMonitor(Config config, StatusCommand.Reporter reporter) {
+ this.reporter = reporter;
+ pollWaitSec = config.getInt(DrillOnYarnConfig.CLIENT_POLL_SEC);
+ if (pollWaitSec < 1) {
+ pollWaitSec = 1;
+ }
+ shutdownWaitSec = config.getInt(DrillOnYarnConfig.CLIENT_STOP_WAIT_SEC);
+ }
+
+ boolean run(boolean verbose) throws ClientException {
+ System.out.print("Stopping...");
+ try {
+ int attemptCount = shutdownWaitSec / pollWaitSec;
+ for (int attempt = 0; attempt < attemptCount; attempt++) {
+ if (!poll()) {
+ break;
+ }
+ }
+ } catch (ClientException e) {
+ System.out.println();
+ throw e;
+ }
+ if (reporter.isStopped()) {
+ System.out.println(" Stopped.");
+ reporter.showFinalStatus();
+ return true;
+ } else {
+ System.out.println();
+ System.out.println(
+ "Application Master is slow to stop, use YARN to check status.");
+ return false;
+ }
+ }
+
+ private boolean poll() throws ClientException {
+ try {
+ Thread.sleep(pollWaitSec * 1000);
+ } catch (InterruptedException e) {
+ return false;
+ }
+ reporter.getReport();
+ if (reporter.isStopped()) {
+ return false;
+ }
+ YarnApplicationState newState = reporter.getState();
+ if (newState == state) {
+ System.out.print(".");
+ return true;
+ }
+ updateState(newState);
+ return true;
+ }
+
+ private void updateState(YarnApplicationState newState) {
+ YarnApplicationState oldState = state;
+ state = newState;
+ if (oldState == null) {
+ return;
+ }
+ System.out.println();
+ System.out.print("Application State: ");
+ System.out.println(state.toString());
+ System.out.print("Stopping...");
+ }
+ }
+
+ private Config config;
+ private YarnRMClient client;
+
+ @Override
+ public void run() throws ClientException {
+ config = DrillOnYarnConfig.config();
+ client = getClient();
+ System.out
+ .println("Stopping Application ID: " + client.getAppId().toString());
+
+ // First get an application report to ensure that the AM is,
+ // in fact, running, and to get the HTTP endpoint.
+
+ StatusCommand.Reporter reporter = new StatusCommand.Reporter(client);
+ try {
+ reporter.getReport();
+ } catch (ClientException e) {
+ reporter = null;
+ }
+
+ // Handle the case of an already stopped app.
+
+ boolean stopped = true;
+ if (reporter == null || reporter.isStopped()) {
+ System.out.println("Application is not running.");
+ } else {
+ // Try to stop the server by sending a STOP REST request.
+
+ if (opts.force) {
+ System.out.println("Forcing shutdown");
+ } else {
+ stopped = gracefulStop(reporter.getAmUrl());
+ }
+
+ // If that did not work, then forcibly kill the AM.
+ // YARN will forcibly kill the AM's containers.
+ // Not pretty, but it works.
+
+ if (opts.force || !stopped) {
+ forcefulStop();
+ }
+
+ // Wait for the AM to stop. The AM may refuse to stop in
+ // the time allowed to wait.
+
+ stopped = new StopMonitor(config, reporter).run(opts.verbose);
+ }
+
+ // If the AM is gone because it started out dead or
+ // we killed it, then forget its App Id.
+
+ if (stopped) {
+ removeAppIdFile();
+ }
+ }
+
+ /**
+ * Do a graceful shutdown by using the AM's REST API call to request stop.
+ * Include the master key with the request to differentiate this request from
+ * accidental uses of the stop REST API.
+ *
+ * @param report
+ * @return
+ */
+
+ private boolean gracefulStop(String baseUrl) {
+ try {
+ if (DoYUtil.isBlank(baseUrl)) {
+ return false;
+ }
+ SimpleRestClient restClient = new SimpleRestClient();
+ String tail = "rest/stop";
+ String masterKey = config.getString(DrillOnYarnConfig.HTTP_REST_KEY);
+ if (!DoYUtil.isBlank(masterKey)) {
+ tail += "?key=" + masterKey;
+ }
+ if (opts.verbose) {
+ System.out.println("Stopping with POST " + baseUrl + "/" + tail);
+ }
+ String result = restClient.send(baseUrl, tail, true);
+ if (result.contains("\"ok\"")) {
+ return true;
+ }
+ System.err.println(
+ "Failed to stop the application master. Response = " + result);
+ return false;
+ } catch (ClientException e) {
+ System.err.println(e.getMessage());
+ System.out.println("Resorting to forced kill");
+ return false;
+ }
+ }
+
+ /**
+ * If the graceful approach did not work, resort to a forceful request. This
+ * asks the AM's NM to kill the AM process.
+ *
+ * @throws ClientException
+ */
+
+ private void forcefulStop() throws ClientException {
+ try {
+ client.killApplication();
+ } catch (YarnClientException e) {
+ throw new ClientException("Failed to stop application master", e);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/f2ac8749/drill-yarn/src/main/java/org/apache/drill/yarn/client/package-info.java
----------------------------------------------------------------------
diff --git a/drill-yarn/src/main/java/org/apache/drill/yarn/client/package-info.java b/drill-yarn/src/main/java/org/apache/drill/yarn/client/package-info.java
new file mode 100644
index 0000000..c03c2fa
--- /dev/null
+++ b/drill-yarn/src/main/java/org/apache/drill/yarn/client/package-info.java
@@ -0,0 +1,31 @@
+/*
+ * 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.
+ */
+
+/**
+ * Implements a "YARN client" for Drill-on-YARN. The client uploads files to
+ * DFS, then requests that YARN start the Application Master. Much fiddling
+ * about is required to support this, such as zipping up the user's configuration,
+ * creating a local file with the app id so we can get app status and shut down
+ * the app, etc.
+ * <p>
+ * Divided into a main program ({@link DrillOnYarn}) and a series of commands.
+ * Some commands are further divided into tasks. Builds on the
+ * YARN and DFS facades defined in the core module.
+ */
+
+package org.apache.drill.yarn.client;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/drill/blob/f2ac8749/drill-yarn/src/main/java/org/apache/drill/yarn/core/AppSpec.java
----------------------------------------------------------------------
diff --git a/drill-yarn/src/main/java/org/apache/drill/yarn/core/AppSpec.java b/drill-yarn/src/main/java/org/apache/drill/yarn/core/AppSpec.java
new file mode 100644
index 0000000..34d4ad1
--- /dev/null
+++ b/drill-yarn/src/main/java/org/apache/drill/yarn/core/AppSpec.java
@@ -0,0 +1,169 @@
+/*
+ * 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.
+ */
+package org.apache.drill.yarn.core;
+
+import java.io.IOException;
+import java.io.PrintStream;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.yarn.api.records.ApplicationSubmissionContext;
+import org.apache.hadoop.yarn.api.records.ContainerLaunchContext;
+import org.apache.hadoop.yarn.api.records.Priority;
+import org.apache.hadoop.yarn.api.records.Resource;
+import org.apache.hadoop.yarn.client.api.YarnClientApplication;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.util.Records;
+
+/**
+ * Abstract description of a remote process launch that describes the many
+ * details needed to launch a process on a remote node. The YARN launch
+ * specification is a mess to work with; this class provides a simpler facade to
+ * gather the information, then turns around and builds the required YARN
+ * object.
+ * <p>
+ * Based on <a href="https://github.com/hortonworks/simple-yarn-app">Simple YARN
+ * App</a>.
+ */
+
+public class AppSpec extends LaunchSpec {
+
+ static final private Log LOG = LogFactory.getLog(LaunchSpec.class);
+
+ /**
+ * The memory required in the allocated container, in MB.
+ */
+
+ public int memoryMb;
+
+ /**
+ * The number of YARN "vcores" (roughly equivalent to CPUs) to allocate to the
+ * process.
+ */
+
+ public int vCores = 1;
+
+ /**
+ * The number of disk resources (that is, disk channels) used by the process.
+ * Available only on some YARN distributions. Fractional values allowed.
+ */
+
+ public double disks;
+
+ /**
+ * The name of the application given to YARN. Appears in the YARN admin UI.
+ */
+
+ public String appName;
+
+ /**
+ * The YARN queue in which to place the application launch request.
+ */
+
+ public String queueName = "default";
+
+ public int priority = 1;
+
+ /**
+ * Whether to run the AM in unmanaged mode. Leave this false for production
+ * code.
+ */
+
+ public boolean unmanaged;
+
+ /**
+ * Optional node label expression for the launch. Selects the nodes on which
+ * the task can run.
+ */
+
+ public String nodeLabelExpr;
+
+ /**
+ * Given this generic description of an application, create the detailed YARN
+ * application submission context required to launch the application.
+ *
+ * @param conf
+ * the YARN configuration obtained by reading the Hadoop
+ * configuration files
+ * @param app
+ * the YARN definition of the client application to be populated from
+ * this generic description
+ * @return the completed application launch context for the given application
+ * @throws IOException
+ * if localized resources are not found in the distributed file
+ * system (such as HDFS)
+ */
+
+ public ApplicationSubmissionContext createAppLaunchContext(
+ YarnConfiguration conf, YarnClientApplication app) throws IOException {
+ ContainerLaunchContext amContainer = createLaunchContext(conf);
+
+ // Finally, set-up ApplicationSubmissionContext for the application
+ ApplicationSubmissionContext appContext = app
+ .getApplicationSubmissionContext();
+ appContext.setApplicationName(appName); // application name
+ appContext.setAMContainerSpec(amContainer);
+ appContext.setResource(getCapability());
+ appContext.setQueue(queueName); // queue
+ appContext.setPriority(Priority.newInstance(priority));
+ if (!DoYUtil.isBlank(nodeLabelExpr)) {
+ LOG.info(
+ "Requesting to run the AM using node expression: " + nodeLabelExpr);
+ appContext.setNodeLabelExpression(nodeLabelExpr);
+ }
+
+ appContext.setUnmanagedAM(unmanaged);
+
+ // Only try the AM once. It will fail if things are misconfigured. Retrying
+ // is unlikely
+ // to fix the configuration problem.
+
+ appContext.setMaxAppAttempts(1);
+
+ // TODO: Security tokens
+
+ return appContext;
+ }
+
+ public Resource getCapability() {
+
+ // Set up resource type requirements for ApplicationMaster
+ Resource capability = Records.newRecord(Resource.class);
+ capability.setMemory(memoryMb);
+ capability.setVirtualCores(vCores);
+ DoYUtil.callSetDiskIfExists(capability, disks);
+ return capability;
+ }
+
+ @Override
+ public void dump(PrintStream out) {
+ out.print("Memory (MB): ");
+ out.println(memoryMb);
+ out.print("Vcores: ");
+ out.println(vCores);
+ out.print("Disks: ");
+ out.println(disks);
+ out.print("Application Name: ");
+ out.println(appName);
+ out.print("Queue: ");
+ out.println(queueName);
+ out.print("Priority: ");
+ out.println(priority);
+ super.dump(out);
+ }
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/f2ac8749/drill-yarn/src/main/java/org/apache/drill/yarn/core/ClusterDef.java
----------------------------------------------------------------------
diff --git a/drill-yarn/src/main/java/org/apache/drill/yarn/core/ClusterDef.java b/drill-yarn/src/main/java/org/apache/drill/yarn/core/ClusterDef.java
new file mode 100644
index 0000000..223b606
--- /dev/null
+++ b/drill-yarn/src/main/java/org/apache/drill/yarn/core/ClusterDef.java
@@ -0,0 +1,212 @@
+/*
+ * 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.
+ */
+package org.apache.drill.yarn.core;
+
+import java.io.PrintStream;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.drill.yarn.appMaster.TaskSpec;
+import org.mortbay.log.Log;
+
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigList;
+import com.typesafe.config.ConfigValue;
+
+public class ClusterDef {
+ // The following keys are relative to the cluster group definition
+
+ public static final String GROUP_NAME = "name";
+ public static final String GROUP_TYPE = "type";
+ public static final String GROUP_SIZE = "count";
+
+ // For the labeled pool
+
+ public static final String DRILLBIT_LABEL = "drillbit-label-expr";
+ public static final String AM_LABEL = "am-label-expr";
+
+ /**
+ * Defined cluster tier types. The value of the type appears as the value of
+ * the {@link $CLUSTER_TYPE} parameter in the config file.
+ */
+
+ public enum GroupType {
+ BASIC,
+ LABELED;
+
+ public static GroupType toEnum(String value) {
+ return GroupType.valueOf( value.toUpperCase() );
+ }
+
+ public String toValue() {
+ return name().toLowerCase();
+ }
+ }
+
+ public static class ClusterGroup {
+ private final String name;
+ private final int count;
+ private final GroupType type;
+
+ public ClusterGroup( Map<String, Object> group, int index, GroupType type ) {
+ this.type = type;
+
+ // Config system has already parsed the value. We insist that the value,
+ // when parsed, was interpreted as an integer. That is, the value had
+ // to be, say 10. Not "10", not 10.0, but just a plain integer.
+
+ try {
+ count = (Integer) group.get(GROUP_SIZE);
+ } catch (ClassCastException e) {
+ throw new IllegalArgumentException(
+ "Expected an integer for " + GROUP_SIZE + " for tier " + index);
+ }
+ Object nameValue = group.get(GROUP_NAME);
+ String theName = null;
+ if (nameValue != null) {
+ theName = nameValue.toString();
+ }
+ if (DoYUtil.isBlank(theName)) {
+ theName = "tier-" + Integer.toString(index);
+ }
+ name = theName;
+ }
+
+
+ public String getName( ) { return name; }
+ public int getCount( ) { return count; }
+ public GroupType getType( ) { return type; }
+
+ public void getPairs(int index, List<NameValuePair> pairs) {
+ String key = DrillOnYarnConfig.append(DrillOnYarnConfig.CLUSTERS,
+ Integer.toString(index));
+ addPairs(pairs, key);
+ }
+
+ protected void addPairs(List<NameValuePair> pairs, String key) {
+ pairs.add(
+ new NameValuePair(DrillOnYarnConfig.append(key, GROUP_NAME), name));
+ pairs.add(
+ new NameValuePair(DrillOnYarnConfig.append(key, GROUP_TYPE), type));
+ pairs.add(
+ new NameValuePair(DrillOnYarnConfig.append(key, GROUP_SIZE), count));
+ }
+
+ public void dump(String prefix, PrintStream out) {
+ out.print(prefix);
+ out.print("name = ");
+ out.println(name);
+ out.print(prefix);
+ out.print("type = ");
+ out.println(type.toValue());
+ out.print(prefix);
+ out.print("count = ");
+ out.println(count);
+ }
+
+ public void modifyTaskSpec(TaskSpec taskSpec) {
+ }
+ }
+
+ public static class BasicGroup extends ClusterGroup {
+
+ public BasicGroup(Map<String, Object> pool, int index) {
+ super(pool, index, GroupType.BASIC);
+ }
+
+ }
+
+ public static class LabeledGroup extends ClusterGroup {
+
+ private final String drillbitLabelExpr;
+
+ public LabeledGroup(Map<String, Object> pool, int index) {
+ super(pool, index, GroupType.LABELED);
+ drillbitLabelExpr = (String) pool.get(DRILLBIT_LABEL);
+ if (drillbitLabelExpr == null) {
+ Log.warn("Labeled pool is missing the drillbit label expression ("
+ + DRILLBIT_LABEL + "), will treat pool as basic.");
+ }
+ }
+
+ public String getLabelExpr( ) { return drillbitLabelExpr; }
+
+ @Override
+ public void dump(String prefix, PrintStream out) {
+ out.print(prefix);
+ out.print("Drillbit label expr = ");
+ out.println((drillbitLabelExpr == null) ? "<none>" : drillbitLabelExpr);
+ }
+
+ @Override
+ protected void addPairs(List<NameValuePair> pairs, String key) {
+ super.addPairs(pairs, key);
+ pairs.add(new NameValuePair(DrillOnYarnConfig.append(key, DRILLBIT_LABEL),
+ drillbitLabelExpr));
+ }
+
+ @Override
+ public void modifyTaskSpec(TaskSpec taskSpec) {
+ taskSpec.containerSpec.nodeLabelExpr = drillbitLabelExpr;
+ }
+ }
+
+ /**
+ * Deserialize a node tier from the configuration file.
+ *
+ * @param n
+ * @return
+ */
+
+ public static ClusterGroup getCluster(Config config, int n) {
+ int index = n + 1;
+ ConfigList tiers = config.getList(DrillOnYarnConfig.CLUSTERS);
+ ConfigValue value = tiers.get(n);
+ if ( value == null ) {
+ throw new IllegalArgumentException( "If cluster group is provided, it cannot be null: group " + index );
+ }
+ @SuppressWarnings("unchecked")
+ Map<String, Object> tier = (Map<String, Object>) value.unwrapped();
+ String type;
+ try {
+ type = tier.get(GROUP_TYPE).toString();
+ } catch (NullPointerException e) {
+ throw new IllegalArgumentException(
+ "Pool type is required for cluster group " + index);
+ }
+ GroupType groupType = GroupType.toEnum(type);
+ if (groupType == null) {
+ throw new IllegalArgumentException(
+ "Undefined type for cluster group " + index + ": " + type);
+ }
+ ClusterGroup tierDef;
+ switch (groupType) {
+ case BASIC:
+ tierDef = new BasicGroup( tier, index );
+ break;
+ case LABELED:
+ tierDef = new LabeledGroup( tier, index );
+ break;
+ default:
+ assert false;
+ throw new IllegalStateException(
+ "Undefined cluster group type: " + groupType);
+ }
+ return tierDef;
+ }
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/f2ac8749/drill-yarn/src/main/java/org/apache/drill/yarn/core/ContainerRequestSpec.java
----------------------------------------------------------------------
diff --git a/drill-yarn/src/main/java/org/apache/drill/yarn/core/ContainerRequestSpec.java b/drill-yarn/src/main/java/org/apache/drill/yarn/core/ContainerRequestSpec.java
new file mode 100644
index 0000000..99a22d7
--- /dev/null
+++ b/drill-yarn/src/main/java/org/apache/drill/yarn/core/ContainerRequestSpec.java
@@ -0,0 +1,125 @@
+/*
+ * 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.
+ */
+package org.apache.drill.yarn.core;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.drill.yarn.appMaster.Scheduler;
+import org.apache.hadoop.yarn.api.records.Priority;
+import org.apache.hadoop.yarn.api.records.Resource;
+import org.apache.hadoop.yarn.client.api.AMRMClient.ContainerRequest;
+import org.apache.hadoop.yarn.util.Records;
+
+/**
+ * Describes a container request in terms of priority, memory, cores and
+ * placement preference. This is a simplified version of the YARN
+ * ContainerRequest structure. This structure is easier to use within the app,
+ * then is translated to the YARN structure when needed.
+ */
+
+public class ContainerRequestSpec {
+ static final Log LOG = LogFactory.getLog(ContainerRequestSpec.class);
+
+ /**
+ * Application-specific priority. Drill-on-Yarn uses the priority to associate
+ * YARN requests with a {@link Scheduler}. When the resource allocation
+ * arrives, we use the priority to trace back to the scheduler that requested
+ * it, and from there to the task to be run in the allocation.
+ * <p>
+ * For this reason, the priority is set by the Drill-on-YARN application; it
+ * is not a user-adjustable value.
+ */
+
+ public int priority = 0;
+
+ /**
+ * Memory, in MB, required by the container.
+ */
+
+ public int memoryMb;
+
+ /**
+ * Number of "virtual cores" required by the task. YARN allocates whole CPU
+ * cores and does not support fractional allocations.
+ */
+
+ public int vCores = 1;
+
+ /**
+ * Number of virtual disks (channels, spindles) to request. Not supported in
+ * Apache YARN, is supported in selected distributions.
+ */
+
+ public double disks;
+
+ /**
+ * Node label expression to apply to this request.
+ */
+
+ public String nodeLabelExpr;
+
+ public List<String> racks = new ArrayList<>();
+ public List<String> hosts = new ArrayList<>();
+
+ /**
+ * Create a YARN ContainerRequest object from the information in this object.
+ *
+ * @return
+ */
+ public ContainerRequest makeRequest() {
+ assert memoryMb != 0;
+
+ Priority priorityRec = Records.newRecord(Priority.class);
+ priorityRec.setPriority(priority);
+
+ Resource capability = Records.newRecord(Resource.class);
+ capability.setMemory(memoryMb);
+ capability.setVirtualCores(vCores);
+ DoYUtil.callSetDiskIfExists(capability, disks);
+
+ boolean relaxLocality = true;
+ String nodeArr[] = null;
+ if (!hosts.isEmpty()) {
+ nodeArr = new String[hosts.size()];
+ hosts.toArray(nodeArr);
+ relaxLocality = false;
+ }
+ String rackArr[] = null;
+ if (!racks.isEmpty()) {
+ nodeArr = new String[racks.size()];
+ racks.toArray(rackArr);
+ relaxLocality = false;
+ }
+ String nodeExpr = null;
+ if (!DoYUtil.isBlank(nodeLabelExpr)) {
+ nodeExpr = nodeLabelExpr;
+ LOG.info("Requesting a container using node expression: " + nodeExpr);
+ }
+
+ // YARN is fragile. To (potentially) pass a node expression, we must use the
+ // 5-argument constructor. The fourth argument (relax locality) MUST be set
+ // to true if we omit the rack and node specs. (Else we get a runtime
+ // error.
+
+ return new ContainerRequest(capability, nodeArr, rackArr, priorityRec,
+ relaxLocality, nodeExpr);
+ }
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/f2ac8749/drill-yarn/src/main/java/org/apache/drill/yarn/core/DfsFacade.java
----------------------------------------------------------------------
diff --git a/drill-yarn/src/main/java/org/apache/drill/yarn/core/DfsFacade.java b/drill-yarn/src/main/java/org/apache/drill/yarn/core/DfsFacade.java
new file mode 100644
index 0000000..09e88ae
--- /dev/null
+++ b/drill-yarn/src/main/java/org/apache/drill/yarn/core/DfsFacade.java
@@ -0,0 +1,345 @@
+/*
+ * 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.
+ */
+package org.apache.drill.yarn.core;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.Map;
+
+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.RemoteIterator;
+import org.apache.hadoop.fs.UnsupportedFileSystemException;
+import org.apache.hadoop.fs.permission.FsAction;
+import org.apache.hadoop.fs.permission.FsPermission;
+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.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.util.ConverterUtils;
+
+import com.typesafe.config.Config;
+
+/**
+ * Facade to the distributed file system (DFS) system that implements
+ * Drill-on-YARN related operations. Some operations are used by both the client
+ * and AM applications.
+ */
+
+public class DfsFacade {
+ public static class DfsFacadeException extends Exception {
+ private static final long serialVersionUID = 1L;
+
+ public DfsFacadeException(String msg) {
+ super(msg);
+ }
+
+ public DfsFacadeException(String msg, Exception e) {
+ super(msg, e);
+ }
+ }
+
+ private FileSystem fs;
+ private Configuration yarnConf;
+ private Config config;
+ private boolean localize;
+
+ public DfsFacade(Config config) {
+ this.config = config;
+ localize = config.getBoolean(DrillOnYarnConfig.LOCALIZE_DRILL);
+ }
+
+ public boolean isLocalized() {
+ return localize;
+ }
+
+ public void connect() throws DfsFacadeException {
+ loadYarnConfig();
+ String dfsConnection = config.getString(DrillOnYarnConfig.DFS_CONNECTION);
+ try {
+ if (DoYUtil.isBlank(dfsConnection)) {
+ fs = FileSystem.get(yarnConf);
+ } else {
+ URI uri;
+ try {
+ uri = new URI(dfsConnection);
+ } catch (URISyntaxException e) {
+ throw new DfsFacadeException(
+ "Illformed DFS connection: " + dfsConnection, e);
+ }
+ fs = FileSystem.get(uri, yarnConf);
+ }
+ } catch (IOException e) {
+ throw new DfsFacadeException("Failed to create the DFS", e);
+ }
+ }
+
+ /**
+ * Lazy loading of YARN configuration since it takes a long time to load.
+ * (YARN provides no caching, sadly.)
+ */
+
+ private void loadYarnConfig() {
+ if (yarnConf == null) {
+ yarnConf = new YarnConfiguration();
+ // On some distributions, lack of proper configuration causes
+ // DFS to default to the local file system. So, a local file
+ // system generally means that the config is wrong, or running
+ // the wrong build of Drill for the user's environment.
+ URI fsUri = FileSystem.getDefaultUri( yarnConf );
+ if(fsUri.toString().startsWith("file:/")) {
+ System.err.println("Warning: Default DFS URI is for a local file system: " + fsUri);
+ }
+ }
+ }
+
+ public static class Localizer {
+ private final DfsFacade dfs;
+ protected File localArchivePath;
+ protected Path dfsArchivePath;
+ FileStatus fileStatus;
+ private String label;
+
+ /**
+ * Resources to be localized (downloaded) to each AM or drillbit node.
+ */
+
+ public Localizer(DfsFacade dfs, File archivePath, String label) {
+ this(dfs, archivePath, dfs.getUploadPath(archivePath), label);
+ }
+
+ public Localizer(DfsFacade dfs, File archivePath, String destName,
+ String label) {
+ this(dfs, archivePath, dfs.getUploadPath(destName), label);
+ }
+
+ public Localizer(DfsFacade dfs, String destPath) {
+ this( dfs, null, new Path(destPath), null );
+ }
+
+ public Localizer(DfsFacade dfs, File archivePath, Path destPath, String label) {
+ this.dfs = dfs;
+ dfsArchivePath = destPath;
+ this.label = label;
+ localArchivePath = archivePath;
+ }
+
+ public String getBaseName() {
+ return localArchivePath.getName();
+ }
+
+ public String getDestPath() {
+ return dfsArchivePath.toString();
+ }
+
+ public void upload() throws DfsFacadeException {
+ dfs.uploadArchive(localArchivePath, dfsArchivePath, label);
+ fileStatus = null;
+ }
+
+ /**
+ * The client may check file status multiple times. Cache it here so we
+ * only retrieve the status once. Cache it here so that the client
+ * doen't have to do the caching.
+ *
+ * @return
+ * @throws DfsFacadeException
+ */
+
+ private FileStatus getStatus() throws DfsFacadeException {
+ if (fileStatus == null) {
+ fileStatus = dfs.getFileStatus(dfsArchivePath);
+ }
+ return fileStatus;
+ }
+
+ public void defineResources(Map<String, LocalResource> resources,
+ String key) throws DfsFacadeException {
+ // Put the application archive, visible to only the application.
+ // Because it is an archive, it will be expanded by YARN prior to launch
+ // of the AM.
+
+ LocalResource drillResource = dfs.makeResource(dfsArchivePath,
+ getStatus(), LocalResourceType.ARCHIVE,
+ LocalResourceVisibility.APPLICATION);
+ resources.put(key, drillResource);
+ }
+
+ public boolean filesMatch() {
+ FileStatus status;
+ try {
+ status = getStatus();
+ } catch (DfsFacadeException e) {
+
+ // An exception is DFS's way of tell us the file does
+ // not exist.
+
+ return false;
+ }
+ return status.getLen() == localArchivePath.length();
+ }
+
+ public String getLabel() {
+ return label;
+ }
+
+ public boolean destExists() throws IOException {
+ return dfs.exists(dfsArchivePath);
+ }
+ }
+
+ public boolean exists(Path path) throws IOException {
+ return fs.exists(path);
+ }
+
+ public Path getUploadPath(File localArchiveFile) {
+ return getUploadPath(localArchiveFile.getName());
+ }
+
+ public Path getUploadPath(String baseName) {
+ String dfsDirStr = config.getString(DrillOnYarnConfig.DFS_APP_DIR);
+
+ Path appDir;
+ if (dfsDirStr.startsWith("/")) {
+ appDir = new Path(dfsDirStr);
+ } else {
+ Path home = fs.getHomeDirectory();
+ appDir = new Path(home, dfsDirStr);
+ }
+ return new Path(appDir, baseName);
+ }
+
+ public void uploadArchive(File localArchiveFile, Path destPath, String label)
+ throws DfsFacadeException {
+ // Create the application upload directory if it does not yet exist.
+
+ String dfsDirStr = config.getString(DrillOnYarnConfig.DFS_APP_DIR);
+ Path appDir = new Path(dfsDirStr);
+ try {
+ // If the directory does not exist, create it, giving this user
+ // (only) read and write access.
+
+ if (!fs.isDirectory(appDir)) {
+ fs.mkdirs(appDir, new FsPermission(FsAction.READ_WRITE, FsAction.NONE, FsAction.NONE));
+ }
+ } catch (IOException e) {
+ throw new DfsFacadeException(
+ "Failed to create DFS directory: " + dfsDirStr, e);
+ }
+
+ // The file must be an archive type so YARN knows to extract its contents.
+
+ String baseName = localArchiveFile.getName();
+ if (DrillOnYarnConfig.findSuffix(baseName) == null) {
+ throw new DfsFacadeException(
+ label + " archive must be .tar.gz, .tgz or .zip: " + baseName);
+ }
+
+ Path srcPath = new Path(localArchiveFile.getAbsolutePath());
+
+ // Do the upload, replacing the old archive.
+
+ try {
+ // TODO: Specify file permissions and owner.
+
+ fs.copyFromLocalFile(false, true, srcPath, destPath);
+ } catch (IOException e) {
+ throw new DfsFacadeException(
+ "Failed to upload " + label + " archive to DFS: "
+ + localArchiveFile.getAbsolutePath() + " --> " + destPath,
+ e);
+ }
+ }
+
+ private FileStatus getFileStatus(Path dfsPath) throws DfsFacadeException {
+ try {
+ return fs.getFileStatus(dfsPath);
+ } catch (IOException e) {
+ throw new DfsFacadeException(
+ "Failed to get DFS status for file: " + dfsPath, e);
+ }
+ }
+
+ /**
+ * Create a local resource definition for YARN. A local resource is one that
+ * must be localized onto the remote node prior to running a command on that
+ * node.
+ * <p>
+ * YARN uses the size and timestamp are used to check if the file has changed
+ * on HDFS to check if YARN can use an existing copy, if any.
+ * <p>
+ * Resources are made public.
+ *
+ * @param conf
+ * Configuration created from the Hadoop config files, in this case,
+ * identifies the target file system.
+ * @param resourcePath
+ * the path (relative or absolute) to the file on the configured file
+ * system (usually HDFS).
+ * @return a YARN local resource records that contains information about path,
+ * size, type, resource and so on that YARN requires.
+ * @throws IOException
+ * if the resource does not exist on the configured file system
+ */
+
+ public LocalResource makeResource(Path dfsPath, FileStatus dfsFileStatus,
+ LocalResourceType type, LocalResourceVisibility visibility)
+ throws DfsFacadeException {
+ URL destUrl;
+ try {
+ destUrl = ConverterUtils.getYarnUrlFromPath(
+ FileContext.getFileContext().makeQualified(dfsPath));
+ } catch (UnsupportedFileSystemException e) {
+ throw new DfsFacadeException(
+ "Unable to convert dfs file to a URL: " + dfsPath.toString(), e);
+ }
+ LocalResource resource = LocalResource.newInstance(destUrl, type,
+ visibility, dfsFileStatus.getLen(),
+ dfsFileStatus.getModificationTime());
+ return resource;
+ }
+
+ public void removeDrillFile(String fileName) throws DfsFacadeException {
+ Path destPath = getUploadPath(fileName);
+ try {
+ fs.delete(destPath, false);
+ } catch (IOException e) {
+ throw new DfsFacadeException(
+ "Failed to delete file: " + destPath.toString(), e);
+ }
+
+ // Remove the Drill directory, but only if it is now empty.
+
+ Path dir = destPath.getParent();
+ try {
+ RemoteIterator<FileStatus> iter = fs.listStatusIterator(dir);
+ if (!iter.hasNext()) {
+ fs.delete(dir, false);
+ }
+ } catch (IOException e) {
+ throw new DfsFacadeException(
+ "Failed to delete directory: " + dir.toString(), e);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/f2ac8749/drill-yarn/src/main/java/org/apache/drill/yarn/core/DoYUtil.java
----------------------------------------------------------------------
diff --git a/drill-yarn/src/main/java/org/apache/drill/yarn/core/DoYUtil.java b/drill-yarn/src/main/java/org/apache/drill/yarn/core/DoYUtil.java
new file mode 100644
index 0000000..3c1d17d
--- /dev/null
+++ b/drill-yarn/src/main/java/org/apache/drill/yarn/core/DoYUtil.java
@@ -0,0 +1,189 @@
+/*
+ * 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.
+ */
+package org.apache.drill.yarn.core;
+
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.security.CodeSource;
+import java.text.DateFormat;
+import java.text.SimpleDateFormat;
+import java.util.Date;
+import java.util.List;
+import java.util.TimeZone;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.yarn.api.records.Container;
+
+public class DoYUtil {
+ static final private Log LOG = LogFactory.getLog(DoYUtil.class);
+
+ private DoYUtil() {
+ }
+
+ public static String join(String separator, List<String> list) {
+ StringBuilder buf = new StringBuilder();
+ String sep = "";
+ for (String item : list) {
+ buf.append(sep);
+ buf.append(item);
+ sep = separator;
+ }
+ return buf.toString();
+ }
+
+ public static void addNonEmpty(List<String> list, String value) {
+ if ( ! isBlank( value ) ) {
+ list.add(value.trim( ));
+ }
+ }
+
+ public static boolean isBlank(String str) {
+ return str == null || str.trim().isEmpty();
+ }
+
+ public static String toIsoTime(long timestamp) {
+
+ // Uses old-style dates rather than java.time because
+ // the code still must compile for JDK 7.
+
+ DateFormat fmt = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+ fmt.setTimeZone(TimeZone.getDefault());
+ return fmt.format(new Date(timestamp));
+ }
+
+ public static String labelContainer(Container container) {
+ StringBuilder buf = new StringBuilder()
+ .append("[id: ")
+ .append(container.getId())
+ .append(", host: ")
+ .append(container.getNodeId().getHost())
+ .append(", priority: ")
+ .append(container.getPriority())
+ .append("]");
+ return buf.toString();
+ }
+
+ /**
+ * Utility method to display YARN container information in a useful way for
+ * log messages.
+ *
+ * @param container
+ * @return
+ */
+
+ public static String describeContainer(Container container) {
+ StringBuilder buf = new StringBuilder()
+ .append("[id: ")
+ .append(container.getId())
+ .append(", host: ")
+ .append(container.getNodeId().getHost())
+ .append(", priority: ")
+ .append(container.getPriority())
+ .append(", memory: ")
+ .append(container.getResource().getMemory())
+ .append(" MB, vcores: ")
+ .append(container.getResource().getVirtualCores())
+ .append("]");
+ return buf.toString();
+ }
+
+ /**
+ * The tracking URL given to YARN is a redirect URL. When giving the URL to
+ * the user, "unwrap" that redirect URL to get the actual site URL.
+ *
+ * @param trackingUrl
+ * @return
+ */
+
+ public static String unwrapAmUrl(String trackingUrl) {
+ return trackingUrl.replace("/redirect", "/");
+ }
+
+ public static Object dynamicCall(Object target, String fnName, Object args[],
+ Class<?> types[]) {
+
+ // First, look for the method using the names and types provided.
+
+ final String methodLabel = target.getClass().getName() + "." + fnName;
+ Method m;
+ try {
+ m = target.getClass().getMethod(fnName, types);
+ } catch (NoSuchMethodException e) {
+
+ // Ignore, but log: the method does not exist in this distribution.
+
+ StringBuilder buf = new StringBuilder();
+ if (types != null) {
+ String sep = "";
+ for (Class<?> type : types) {
+ buf.append(sep);
+ buf.append(type.getName());
+ sep = ",";
+ }
+ }
+ LOG.trace("Not supported in this YARN distribution: " + methodLabel + "("
+ + buf.toString() + ")");
+ CodeSource src = target.getClass().getProtectionDomain().getCodeSource();
+ if (src != null) {
+ java.net.URL jar = src.getLocation();
+ LOG.trace("Class found in URL: " + jar.toString());
+ }
+ return null;
+ } catch (SecurityException e) {
+ LOG.error("Security prevents dynamic method calls", e);
+ return null;
+ }
+
+ // Next, call the method with the arguments provided.
+
+ Object ret = null;
+ try {
+ ret = m.invoke(target, args);
+ } catch (IllegalAccessException | IllegalArgumentException
+ | InvocationTargetException e) {
+ LOG.error("Failed to dynamically call " + methodLabel, e);
+ return null;
+ }
+ StringBuilder buf = new StringBuilder();
+ if (args != null) {
+ String sep = "";
+ for (Object arg : args) {
+ buf.append(sep);
+ buf.append(arg == null ? "null" : arg.toString());
+ sep = ",";
+ }
+ }
+ LOG.trace(
+ "Successfully called " + methodLabel + "( " + buf.toString() + ")");
+
+ // Return any return value. Will be null if the method is returns void.
+
+ return ret;
+ }
+
+ public static void callSetDiskIfExists(Object target, double arg) {
+ dynamicCall(target, "setDisks", new Object[] { arg },
+ new Class<?>[] { Double.TYPE });
+ }
+
+ public static double callGetDiskIfExists(Object target) {
+ Object ret = dynamicCall(target, "getDisks", null, null);
+ return (ret == null) ? 0.0 : (Double) ret;
+ }
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/f2ac8749/drill-yarn/src/main/java/org/apache/drill/yarn/core/DoyConfigException.java
----------------------------------------------------------------------
diff --git a/drill-yarn/src/main/java/org/apache/drill/yarn/core/DoyConfigException.java b/drill-yarn/src/main/java/org/apache/drill/yarn/core/DoyConfigException.java
new file mode 100644
index 0000000..422b89b
--- /dev/null
+++ b/drill-yarn/src/main/java/org/apache/drill/yarn/core/DoyConfigException.java
@@ -0,0 +1,30 @@
+/*
+ * 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.
+ */
+package org.apache.drill.yarn.core;
+
+public class DoyConfigException extends Exception {
+ private static final long serialVersionUID = 1L;
+
+ public DoyConfigException(String msg) {
+ super(msg);
+ }
+
+ public DoyConfigException(String msg, Exception e) {
+ super(msg, e);
+ }
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/f2ac8749/drill-yarn/src/main/java/org/apache/drill/yarn/core/DrillOnYarnConfig.java
----------------------------------------------------------------------
diff --git a/drill-yarn/src/main/java/org/apache/drill/yarn/core/DrillOnYarnConfig.java b/drill-yarn/src/main/java/org/apache/drill/yarn/core/DrillOnYarnConfig.java
new file mode 100644
index 0000000..38ecd1c
--- /dev/null
+++ b/drill-yarn/src/main/java/org/apache/drill/yarn/core/DrillOnYarnConfig.java
@@ -0,0 +1,841 @@
+/*
+ * 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.
+ */
+package org.apache.drill.yarn.core;
+
+import java.io.File;
+import java.io.PrintStream;
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.drill.common.config.CommonConstants;
+import org.apache.drill.common.config.DrillConfig;
+import org.apache.drill.common.scanner.ClassPathScanner;
+import org.apache.drill.common.scanner.persistence.ScanResult;
+import org.apache.drill.exec.ExecConstants;
+
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigException;
+import com.typesafe.config.ConfigFactory;
+
+/**
+ * Configuration used within the Drill-on-YARN code. Configuration comes from four
+ * sources (in order of precedence):
+ * <ol>
+ * <li>System properties</li>
+ * <li>$SITE_DIR/drill-on-yarn.conf</li>
+ * <li>Distribution-specific properties in $SITE_HOME/conf/doy-distrib.conf</li>
+ * <li>Drill-on-YARN defaults in drill-on-yarn-defaults.conf. (Which should be
+ * disjoint from the Drill settings.)</li>
+ * <li>Drill properties (via the Drill override system)</li>
+ * </ol>
+ * <p>
+ * Defines constants for each property, including some defined in Drill. This provides
+ * a uniform property access interface even if some properties migrate between DoY and
+ * Drill proper.
+ */
+
+public class DrillOnYarnConfig {
+ public static final String DEFAULTS_FILE_NAME = "drill-on-yarn-defaults.conf";
+ public static final String DISTRIB_FILE_NAME = "doy-distrib.conf";
+ public static final String CONFIG_FILE_NAME = "drill-on-yarn.conf";
+
+ public static final String DRILL_ON_YARN_PARENT = "drill.yarn";
+ public static final String DOY_CLIENT_PARENT = append(DRILL_ON_YARN_PARENT, "client");
+ public static final String DOY_AM_PARENT = append(DRILL_ON_YARN_PARENT, "am");
+ public static final String DOY_DRILLBIT_PARENT = append(DRILL_ON_YARN_PARENT, "drillbit");
+ public static final String FILES_PARENT = append(DRILL_ON_YARN_PARENT, "drill-install");
+ public static final String DFS_PARENT = append(DRILL_ON_YARN_PARENT, "dfs");
+ public static final String HTTP_PARENT = append(DRILL_ON_YARN_PARENT, "http");
+ public static final String YARN_PARENT = append(DRILL_ON_YARN_PARENT, "yarn");
+ public static final String HADOOP_PARENT = append(DRILL_ON_YARN_PARENT, "hadoop");
+ public static final String CLIENT_PARENT = append(DRILL_ON_YARN_PARENT, "client");
+
+ public static final String APP_NAME = append(DRILL_ON_YARN_PARENT, "app-name");
+ public static final String CLUSTER_ID = ExecConstants.SERVICE_NAME;
+
+ public static final String DFS_CONNECTION = append(DFS_PARENT, "connection");
+ public static final String DFS_APP_DIR = append(DFS_PARENT, "app-dir");
+
+ public static final String YARN_QUEUE = append(YARN_PARENT, "queue");
+ public static final String YARN_PRIORITY = append(YARN_PARENT, "priority");
+
+ public static final String DRILL_ARCHIVE_PATH = append(FILES_PARENT, "client-path");
+ public static final String DRILL_DIR_NAME = append(FILES_PARENT, "dir-name");
+
+ /**
+ * Key used for the Drill archive file in the AM launch config.
+ */
+
+ public static final String DRILL_ARCHIVE_KEY = append(FILES_PARENT, "drill-key");
+ public static final String SITE_ARCHIVE_KEY = append(FILES_PARENT, "site-key");
+ public static final String LOCALIZE_DRILL = append(FILES_PARENT, "localize");
+ public static final String CONF_AS_SITE = append(FILES_PARENT, "conf-as-site");
+ public static final String DRILL_HOME = append(FILES_PARENT, "drill-home");
+ public static final String SITE_DIR = append(FILES_PARENT, "site-dir");
+ public static final String JAVA_LIB_PATH = append(FILES_PARENT, "library-path");
+
+ public static final String HADOOP_HOME = append(HADOOP_PARENT, "home");
+ public static final String HADOOP_CLASSPATH = append(HADOOP_PARENT, "class-path");
+ public static final String HBASE_CLASSPATH = append(HADOOP_PARENT, "hbase-class-path");
+
+ public static final String MEMORY_KEY = "memory-mb";
+ public static final String VCORES_KEY = "vcores";
+ public static final String DISKS_KEY = "disks";
+ public static final String VM_ARGS_KEY = "vm-args";
+ public static final String HEAP_KEY = "heap";
+
+ public static final String AM_MEMORY = append(DOY_AM_PARENT, MEMORY_KEY);
+ public static final String AM_VCORES = append(DOY_AM_PARENT, VCORES_KEY);
+ public static final String AM_DISKS = append(DOY_AM_PARENT, DISKS_KEY);
+ public static final String AM_NODE_LABEL_EXPR = append(DOY_AM_PARENT, "node-label-expr");
+ public static final String AM_HEAP = append(DOY_AM_PARENT, HEAP_KEY);
+ public static final String AM_VM_ARGS = append(DOY_AM_PARENT, VM_ARGS_KEY);
+ public static final String AM_POLL_PERIOD_MS = append(DOY_AM_PARENT, "poll-ms");
+ public static final String AM_TICK_PERIOD_MS = append(DOY_AM_PARENT, "tick-ms");
+ public static final String AM_PREFIX_CLASSPATH = append(DOY_AM_PARENT, "prefix-class-path");
+ public static final String AM_CLASSPATH = append(DOY_AM_PARENT, "class-path");
+ public static final String AM_DEBUG_LAUNCH = append(DOY_AM_PARENT, "debug-launch");
+ public static final String AM_ENABLE_AUTO_SHUTDOWN = append(DOY_AM_PARENT, "auto-shutdown");
+
+ public static final String DRILLBIT_MEMORY = append(DOY_DRILLBIT_PARENT, MEMORY_KEY);
+ public static final String DRILLBIT_VCORES = append(DOY_DRILLBIT_PARENT, VCORES_KEY);
+ public static final String DRILLBIT_DISKS = append(DOY_DRILLBIT_PARENT, DISKS_KEY);
+ public static final String DRILLBIT_VM_ARGS = append(DOY_DRILLBIT_PARENT, VM_ARGS_KEY);
+ public static final String DRILLBIT_HEAP = append(DOY_DRILLBIT_PARENT, HEAP_KEY);
+ public static final String DRILLBIT_DIRECT_MEM = append(DOY_DRILLBIT_PARENT, "max-direct-memory");
+ public static final String DRILLBIT_CODE_CACHE = append(DOY_DRILLBIT_PARENT, "code-cache");
+ public static final String DRILLBIT_LOG_GC = append(DOY_DRILLBIT_PARENT, "log-gc");
+ public static final String DRILLBIT_PREFIX_CLASSPATH = append( DOY_DRILLBIT_PARENT, "prefix-class-path");
+ public static final String DRILLBIT_EXTN_CLASSPATH = append( DOY_DRILLBIT_PARENT, "extn-class-path");
+ public static final String DRILLBIT_CLASSPATH = append(DOY_DRILLBIT_PARENT, "class-path");
+ public static final String DRILLBIT_MAX_RETRIES = append(DOY_DRILLBIT_PARENT, "max-retries");
+ public static final String DRILLBIT_DEBUG_LAUNCH = append(DOY_DRILLBIT_PARENT, "debug-launch");
+ public static final String DRILLBIT_HTTP_PORT = ExecConstants.HTTP_PORT;
+ public static final String DISABLE_YARN_LOGS = append(DOY_DRILLBIT_PARENT, "disable-yarn-logs");
+ public static final String DRILLBIT_USER_PORT = ExecConstants.INITIAL_USER_PORT;
+ public static final String DRILLBIT_BIT_PORT = ExecConstants.INITIAL_BIT_PORT;
+ public static final String DRILLBIT_USE_HTTPS = ExecConstants.HTTP_ENABLE_SSL;
+ public static final String DRILLBIT_MAX_EXTRA_NODES = append(DOY_DRILLBIT_PARENT, "max-extra-nodes");
+ public static final String DRILLBIT_REQUEST_TIMEOUT_SEC = append(DOY_DRILLBIT_PARENT, "request-timeout-secs");
+
+ public static final String ZK_CONNECT = ExecConstants.ZK_CONNECTION;
+ public static final String ZK_ROOT = ExecConstants.ZK_ROOT;
+ public static final String ZK_FAILURE_TIMEOUT_MS = ExecConstants.ZK_TIMEOUT;
+ public static final String ZK_RETRY_COUNT = ExecConstants.ZK_RETRY_TIMES;
+ public static final String ZK_RETRY_DELAY_MS = ExecConstants.ZK_RETRY_DELAY;
+
+ // Names selected to be parallel to Drillbit HTTP config.
+
+ public static final String HTTP_ENABLED = append(HTTP_PARENT, "enabled");
+ public static final String HTTP_ENABLE_SSL = append(HTTP_PARENT, "ssl-enabled");
+ public static final String HTTP_PORT = append(HTTP_PARENT, "port");
+ public static final String HTTP_AUTH_TYPE = append(HTTP_PARENT, "auth-type");
+ public static final String HTTP_REST_KEY = append(HTTP_PARENT, "rest-key");
+ public static final String HTTP_SESSION_MAX_IDLE_SECS = append(HTTP_PARENT, "session-max-idle-secs");
+ public static final String HTTP_DOCS_LINK = append(HTTP_PARENT, "docs-link");
+ public static final String HTTP_REFRESH_SECS = append(HTTP_PARENT, "refresh-secs");
+ public static final String HTTP_USER_NAME = append(HTTP_PARENT, "user-name");
+ public static final String HTTP_PASSWORD = append(HTTP_PARENT, "password");
+
+ public static final String AUTH_TYPE_NONE = "none";
+ public static final String AUTH_TYPE_DRILL = "drill";
+ public static final String AUTH_TYPE_SIMPLE = "simple";
+
+ public static final String CLIENT_POLL_SEC = append(CLIENT_PARENT, "poll-sec");
+ public static final String CLIENT_START_WAIT_SEC = append(CLIENT_PARENT, "start-wait-sec");
+ public static final String CLIENT_STOP_WAIT_SEC = append(CLIENT_PARENT, "stop-wait-sec");
+
+ public static final String CLUSTERS = append(DRILL_ON_YARN_PARENT, "cluster");
+
+ /**
+ * Name of the subdirectory of the container directory that will hold
+ * localized Drill distribution files. This name must be consistent between AM
+ * launch request and AM launch, and between Drillbit launch request and
+ * Drillbit launch. This name is fixed; there is no reason for the user to
+ * change it as it is visible only in the YARN container environment.
+ */
+
+ public static String LOCAL_DIR_NAME = "drill";
+
+ // Environment variables used to pass information from the Drill-on-YARN
+ // Client to the AM, or from the AM to the Drillbit launch script.
+
+ public static final String APP_ID_ENV_VAR = "DRILL_AM_APP_ID";
+ public static final String DRILL_ARCHIVE_ENV_VAR = "DRILL_ARCHIVE";
+ public static final String SITE_ARCHIVE_ENV_VAR = "SITE_ARCHIVE";
+ public static final String DRILL_HOME_ENV_VAR = "DRILL_HOME";
+ public static final String DRILL_SITE_ENV_VAR = "DRILL_CONF_DIR";
+ public static final String AM_HEAP_ENV_VAR = "DRILL_AM_HEAP";
+ public static final String AM_JAVA_OPTS_ENV_VAR = "DRILL_AM_JAVA_OPTS";
+ public static final String DRILL_CLASSPATH_ENV_VAR = "DRILL_CLASSPATH";
+ public static final String DRILL_CLASSPATH_PREFIX_ENV_VAR = "DRILL_CLASSPATH_PREFIX";
+ public static final String DOY_LIBPATH_ENV_VAR = "DOY_JAVA_LIB_PATH";
+ public static final String DRILL_DEBUG_ENV_VAR = "DRILL_DEBUG";
+
+ /**
+ * Special value for the DRILL_DIR_NAME parameter to indicate to use the base
+ * name of the archive as the Drill home path.
+ */
+
+ private static final Object BASE_NAME_MARKER = "<base>";
+
+ /**
+ * The name of the Drill site archive stored in dfs. Since the archive is
+ * created by the client as a temp file, it's local name has no meaning; we
+ * use this standard name on dfs.
+ */
+
+ public static final String SITE_ARCHIVE_NAME = "site.tar.gz";
+
+ protected static DrillOnYarnConfig instance;
+ private File drillSite;
+ private File drillHome;
+ private static DrillConfig drillConfig;
+ private Config config;
+ private ScanResult classPathScan;
+
+ public static String append(String parent, String key) {
+ return parent + "." + key;
+ }
+
+ // Protected only to allow creating a test version of this class.
+
+ protected DrillOnYarnConfig( ) {
+ }
+
+ public static DrillOnYarnConfig load() throws DoyConfigException {
+ instance = new DrillOnYarnConfig();
+ instance.doLoad(Thread.currentThread().getContextClassLoader());
+ return instance;
+ }
+
+ /**
+ * Load the config.
+ * @param cl class loader to use for resource searches (except defaults).
+ * Allows test to specify a specialized version.
+ * <p>
+ * Implemented in a way that allows unit testing. The parseUrl( ) methods
+ * let us mock the files; the load( ) methods seem to not actually use the
+ * provided class loader.
+ *
+ * @throws DoyConfigException
+ */
+ protected void doLoad(ClassLoader cl) throws DoyConfigException {
+ Config drillConfig = loadDrillConfig();
+
+ // Resolution order, larger numbers take precedence.
+ // 1. Drill-on-YARN defaults.
+ // File is at root of the package tree.
+
+ URL url = DrillOnYarnConfig.class.getResource(DEFAULTS_FILE_NAME);
+ if (url == null) {
+ throw new IllegalStateException(
+ "Drill-on-YARN defaults file is required: " + DEFAULTS_FILE_NAME);
+ }
+ config = ConfigFactory.parseURL(url).withFallback(drillConfig);
+
+ // 2. Optional distribution-specific configuration-file.
+ // (Lets a vendor, for example, specify the default DFS upload location
+ // without tinkering with the user's own settings.
+
+ url = cl.getResource(DISTRIB_FILE_NAME);
+ if (url != null) {
+ config = ConfigFactory.parseURL(url).withFallback(config);
+ }
+
+ // 3. User's Drill-on-YARN configuration.
+ // Optional since defaults are fine & ZK comes from drill-override.conf.
+
+ url = cl.getResource(CONFIG_FILE_NAME);
+ if (url != null) {
+ config = ConfigFactory.parseURL(url).withFallback(config);
+ }
+
+ // 4. System properties
+ // Allows -Dfoo=bar on the command line.
+ // But, note that substitutions are NOT allowed in system properties!
+
+ config = ConfigFactory.systemProperties().withFallback(config);
+
+ // Resolution allows ${foo.bar} syntax in values, but only for values
+ // from config files, not from system properties.
+
+ config = config.resolve();
+ }
+
+ private static Config loadDrillConfig() {
+ drillConfig = DrillConfig
+ .create(CommonConstants.CONFIG_OVERRIDE_RESOURCE_PATHNAME);
+ return drillConfig.resolve();
+ }
+
+ public DrillConfig getDrillConfig() {
+ return drillConfig;
+ }
+
+ /**
+ * Return Drill's class path scan. This is used only in the main thread during
+ * initialization. Not needed by the client, so done in an unsynchronized,
+ * lazy fashion.
+ *
+ * @return
+ */
+
+ public ScanResult getClassPathScan() {
+ if (classPathScan == null) {
+ classPathScan = ClassPathScanner.fromPrescan(drillConfig);
+ }
+ return classPathScan;
+ }
+
+ /**
+ * Obtain Drill home from the DRILL_HOME environment variable set by
+ * drill-config.sh, which is called from drill-on-yarn.sh. When debugging,
+ * DRILL_HOME must be set in the environment.
+ * <p>
+ * This information is required only by the client to prepare for uploads to
+ * DFS.
+ *
+ * @throws DoyConfigException
+ */
+
+ public void setClientPaths() throws DoyConfigException {
+ setClientDrillHome();
+ setSiteDir();
+ }
+
+ private void setClientDrillHome() throws DoyConfigException {
+ // Try the environment variable that should have been
+ // set in drill-on-yarn.sh (for the client) or in the
+ // launch environment (for the AM.)
+
+ String homeDir = getEnv(DRILL_HOME_ENV_VAR);
+
+ // For ease in debugging, allow setting the Drill home in
+ // drill-on-yarn.conf.
+ // This setting is also used for a non-localized run.
+
+ if (DoYUtil.isBlank(homeDir)) {
+ homeDir = config.getString(DRILL_HOME);
+ }
+ if (DoYUtil.isBlank(homeDir)) {
+ throw new DoyConfigException(
+ "The DRILL_HOME environment variable must point to your Drill install.");
+ }
+ drillHome = new File(homeDir);
+ }
+
+ /**
+ * All environment variable access goes through this function to allow unit
+ * tests to replace this function to set test values. (The Java environment is
+ * immutable, so it is not possible for unit tests to change the actual
+ * environment.)
+ *
+ * @param key
+ * @return
+ */
+
+ protected String getEnv(String key) {
+ return System.getenv(key);
+ }
+
+ /**
+ * On both the client and the AM, the site directory is optional. If provided,
+ * it was set with the --config (or --site) option to the script that launched
+ * the client or AM. In both cases, the script sets the drill.yarn.siteDir
+ * system property (and leaks the DRILL_HOME environment variable.)
+ * <p>
+ * For ease of debugging, if neither of those are set, this method uses the
+ * location of the drill-on-yarn configuration file to infer the site
+ * directory.
+ * <p>
+ * On the client, the site directory will be the "original" directory that
+ * contains the user's "master" files. On the AM, the site directory is a
+ * localized version of the client directory. Because of the way tar works,
+ * both the client and AM site directories have the same name; though the path
+ * to that name obviously differs.
+ *
+ * @throws DoyConfigException
+ */
+
+ private void setSiteDir() throws DoyConfigException {
+ // The site directory is the one where the config file lives.
+ // This should have been set in an environment variable by the launch
+ // script.
+
+ String sitePath = getEnv("DRILL_CONF_DIR");
+ if (!DoYUtil.isBlank(sitePath)) {
+ drillSite = new File(sitePath);
+ } else {
+
+ // Otherwise, let's guess it from the config file. This version assists
+ // in debugging as it reduces setup steps.
+
+ ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
+ if (classLoader == null) {
+ classLoader = DrillOnYarnConfig.class.getClassLoader();
+ }
+
+ URL url = classLoader.getResource(CommonConstants.CONFIG_OVERRIDE_RESOURCE_PATHNAME);
+ if (url == null) {
+ throw new DoyConfigException(
+ "Drill configuration file is missing: " + CommonConstants.CONFIG_OVERRIDE_RESOURCE_PATHNAME);
+ }
+ File confFile;
+ try {
+ java.nio.file.Path confPath = Paths.get(url.toURI());
+ confFile = confPath.toFile();
+ } catch (URISyntaxException e) {
+ throw new DoyConfigException(
+ "Invalid path to Drill-on-YARN configuration file: "
+ + url.toString(),
+ e);
+ }
+ drillSite = confFile.getParentFile();
+ }
+
+ // Verify that the site directory is not just $DRILL_HOME/conf.
+ // Since the calling script does not differentiate between the two cases.
+ // But, treat $DRILL_HOME/conf as the site directory if:
+ // 1. The conf-as-site property is true, or
+ // 2. The Drill archive resides within $DRILL_HOME.
+ //
+ // The above situations occur in certain distributions that
+ // ship the archive inside the site directory and don't use a
+ // site directory.
+
+ if (drillHome.equals(drillSite.getParentFile())
+ && !config.getBoolean(CONF_AS_SITE)) {
+ String drillArchivePath = config
+ .getString(DrillOnYarnConfig.DRILL_ARCHIVE_PATH);
+ if (!DoYUtil.isBlank(drillArchivePath)) {
+ File archiveFile = new File(drillArchivePath);
+ if (!archiveFile.isAbsolute() && !archiveFile.getAbsolutePath()
+ .startsWith(drillHome.getAbsolutePath())) {
+ drillSite = null;
+ }
+ }
+ }
+ }
+
+ /**
+ * Retrieve the AM Drill home location from the DRILL_HOME variable set in the
+ * drill-am.sh launch script.
+ *
+ * @throws DoyConfigException
+ */
+
+ public void setAmDrillHome() throws DoyConfigException {
+ String drillHomeStr = getEnv(DRILL_HOME_ENV_VAR);
+ drillHome = new File(drillHomeStr);
+ setSiteDir();
+ }
+
+ public Config getConfig() {
+ return instance.config;
+ }
+
+ public static DrillOnYarnConfig instance() {
+ assert instance != null;
+ return instance;
+ }
+
+ public static Config config() {
+ return instance().getConfig();
+ }
+
+ /**
+ * Return the Drill home on this machine as inferred from the config file
+ * contents or location.
+ *
+ * @return
+ */
+
+ public File getLocalDrillHome() {
+ return drillHome;
+ }
+
+ public void dump() {
+ dump(System.out);
+ }
+
+ private static final String keys[] = {
+ // drill.yarn
+
+ APP_NAME,
+ CLUSTER_ID,
+
+ // drill.yarn.dfs
+
+ DFS_CONNECTION,
+ DFS_APP_DIR,
+
+ // drill.yarn.hadoop
+
+ HADOOP_HOME,
+ HADOOP_CLASSPATH,
+ HBASE_CLASSPATH,
+
+ // drill.yarn.yarn
+
+ YARN_QUEUE,
+ YARN_PRIORITY,
+
+ // drill.yarn.drill-install
+
+ DRILL_ARCHIVE_PATH,
+ DRILL_DIR_NAME,
+ LOCALIZE_DRILL,
+ CONF_AS_SITE,
+ DRILL_HOME,
+ DRILL_ARCHIVE_KEY,
+ SITE_ARCHIVE_KEY,
+ JAVA_LIB_PATH,
+
+ // drill.yarn.client
+
+ CLIENT_POLL_SEC,
+ CLIENT_START_WAIT_SEC,
+ CLIENT_STOP_WAIT_SEC,
+
+ // drill.yarn.am
+
+ AM_MEMORY,
+ AM_VCORES,
+ AM_DISKS,
+ AM_NODE_LABEL_EXPR,
+ AM_VM_ARGS,
+ AM_HEAP,
+ AM_POLL_PERIOD_MS,
+ AM_TICK_PERIOD_MS,
+ AM_PREFIX_CLASSPATH,
+ AM_CLASSPATH,
+ AM_DEBUG_LAUNCH,
+ AM_ENABLE_AUTO_SHUTDOWN,
+
+ // drill.yarn.zk
+
+ ZK_CONNECT,
+ ZK_ROOT,
+ ZK_RETRY_COUNT,
+ ZK_RETRY_DELAY_MS,
+ ZK_FAILURE_TIMEOUT_MS,
+
+ // drill.yarn.drillbit
+
+ DRILLBIT_MEMORY,
+ DRILLBIT_VCORES,
+ DRILLBIT_DISKS,
+ DRILLBIT_VM_ARGS,
+ DRILLBIT_HEAP,
+ DRILLBIT_DIRECT_MEM,
+ DRILLBIT_CODE_CACHE,
+ DRILLBIT_PREFIX_CLASSPATH,
+ DRILLBIT_EXTN_CLASSPATH,
+ DRILLBIT_CLASSPATH,
+ DRILLBIT_MAX_RETRIES,
+ DRILLBIT_DEBUG_LAUNCH,
+ DRILLBIT_MAX_EXTRA_NODES,
+ DRILLBIT_REQUEST_TIMEOUT_SEC,
+ DISABLE_YARN_LOGS,
+ DRILLBIT_HTTP_PORT,
+ DRILLBIT_USER_PORT,
+ DRILLBIT_BIT_PORT,
+ DRILLBIT_USE_HTTPS,
+
+ // drill.yarn.http
+
+ HTTP_ENABLED,
+ HTTP_ENABLE_SSL,
+ HTTP_PORT,
+ HTTP_AUTH_TYPE,
+ HTTP_SESSION_MAX_IDLE_SECS,
+ HTTP_DOCS_LINK,
+ HTTP_REFRESH_SECS,
+ // Do not include AM_REST_KEY: it is supposed to be secret.
+ // Same is true of HTTP_USER_NAME and HTTP_PASSWORD
+ };
+
+ private static String envVars[] = {
+ APP_ID_ENV_VAR,
+ DRILL_HOME_ENV_VAR,
+ DRILL_SITE_ENV_VAR,
+ AM_HEAP_ENV_VAR,
+ AM_JAVA_OPTS_ENV_VAR,
+ DRILL_CLASSPATH_PREFIX_ENV_VAR,
+ DRILL_CLASSPATH_ENV_VAR,
+ DRILL_ARCHIVE_ENV_VAR,
+ SITE_ARCHIVE_ENV_VAR,
+ DRILL_DEBUG_ENV_VAR
+ };
+
+ private void dump(PrintStream out) {
+ for (String key : keys) {
+ out.print(key);
+ out.print(" = ");
+ try {
+ out.println(config.getString(key));
+ } catch (ConfigException.Missing e) {
+ out.println("<missing>");
+ }
+ }
+ out.print(CLUSTERS);
+ out.println("[");
+ for (int i = 0; i < clusterGroupCount(); i++) {
+ ClusterDef.ClusterGroup cluster = ClusterDef.getCluster(config, i);
+ out.print(i);
+ out.println(" = {");
+ cluster.dump(" ", out);
+ out.println(" }");
+ }
+ out.println("]");
+ }
+
+ public void dumpEnv(PrintStream out) {
+ out.print("environment");
+ out.println("[");
+ for (String envVar : envVars) {
+ String value = getEnv(envVar);
+ out.print(envVar);
+ out.print(" = ");
+ if (value == null) {
+ out.print("<unset>");
+ } else {
+ out.print("\"");
+ out.print(value);
+ out.print("\"");
+ }
+ out.println();
+ }
+ out.println("]");
+ }
+
+ public List<NameValuePair> getPairs() {
+ List<NameValuePair> pairs = new ArrayList<>();
+ for (String key : keys) {
+ pairs.add(new NameValuePair(key, config.getString(key)));
+ }
+ for (int i = 0; i < clusterGroupCount(); i++) {
+ ClusterDef.ClusterGroup pool = ClusterDef.getCluster(config, i);
+ pool.getPairs(i, pairs);
+ }
+
+ // Add environment variables as "pseudo" properties,
+ // prefixed with "envt.".
+
+ for (String envVar : envVars) {
+ pairs.add(new NameValuePair("envt." + envVar, getEnv(envVar)));
+ }
+ return pairs;
+ }
+
+ public static String clusterGroupKey(int index, String key) {
+ return CLUSTERS + "." + index + "." + key;
+ }
+
+ public int clusterGroupCount() {
+ return config.getList(CLUSTERS).size();
+ }
+
+ private static String suffixes[] = { ".tar.gz", ".tgz", ".zip" };
+
+ public static String findSuffix(String baseName) {
+ baseName = baseName.toLowerCase();
+ for (String extn : suffixes) {
+ if (baseName.endsWith(extn)) {
+ return extn;
+ }
+ }
+ return null;
+ }
+
+ /**
+ * Get the location of Drill home on a remote machine, relative to the
+ * container working directory. Used when constructing a launch context.
+ * Assumes either the absolute path from the config file, or a constructed
+ * path to the localized Drill on the remote node. YARN examples use "./foo"
+ * to refer to container resources. But, since we cannot be sure when such a
+ * path is evaluated, we explicitly use YARN's PWD environment variable to get
+ * the absolute path.
+ *
+ * @return the remote path, with the "$PWD" environment variable.
+ * @throws DoyConfigException
+ */
+
+ public String getRemoteDrillHome() throws DoyConfigException {
+ // If the application is not localized, then the user can tell us the remote
+ // path in the config file. Otherwise, we assume that the remote path is the
+ // same as the local path.
+
+ if (!config.getBoolean(LOCALIZE_DRILL)) {
+ String drillHomePath = config.getString(DRILL_HOME);
+ if (DoYUtil.isBlank(drillHomePath)) {
+ drillHomePath = drillHome.getAbsolutePath();
+ }
+ return drillHomePath;
+ }
+
+ // The application is localized. Work out the location within the container
+ // directory. The path starts with the "key" we specify when uploading the
+ // Drill archive; YARN expands the archive into a folder of that name.
+
+ String drillHome = "$PWD/" + config.getString(DRILL_ARCHIVE_KEY);
+
+ String home = config.getString(DRILL_DIR_NAME);
+ if (DoYUtil.isBlank(home)) {
+ // Assume the archive expands without a subdirectory.
+ }
+
+ // If the special "<base>" marker is used, assume that the path depends
+ // on the name of the archive, which we know from the config file.
+
+ else if (home.equals(BASE_NAME_MARKER)) {
+
+ // Otherwise, assume that the archive expands to a directory with the
+ // same name as the archive itself (minus the archive suffix.)
+
+ String drillArchivePath = config
+ .getString(DrillOnYarnConfig.DRILL_ARCHIVE_PATH);
+ if (DoYUtil.isBlank(drillArchivePath)) {
+ throw new DoyConfigException("Required config property not set: "
+ + DrillOnYarnConfig.DRILL_ARCHIVE_PATH);
+ }
+ File localArchiveFile = new File(drillArchivePath);
+ home = localArchiveFile.getName();
+ String suffix = findSuffix(home);
+ if (suffix == null) {
+ throw new DoyConfigException(DrillOnYarnConfig.DRILL_ARCHIVE_PATH
+ + " does not name a valid archive: " + drillArchivePath);
+ }
+ drillHome += "/" + home.substring(0, home.length() - suffix.length());
+ } else {
+ // If the user told us the name of the directory within the archive,
+ // use it.
+
+ drillHome += "/" + home;
+ }
+ return drillHome;
+ }
+
+ /**
+ * Get the optional remote site directory name. This name will include the
+ * absolute path for a non-localized application. It will return the path
+ * relative to the container for a localized application. In the localized
+ * case, the site archive is tar'ed relative to the site directory so that its
+ * contents are unarchived directly into the YARN-provided folder (with the
+ * name of the archive) key. That is, if the site directory on the client is
+ * /var/drill/my-site, the contents of the tar file will be
+ * "./drill-override.conf", etc., and the remote location is
+ * $PWD/site-key/drill-override.conf, where site-key is the key name used to
+ * localize the site archive.
+ *
+ * @return
+ */
+
+ public String getRemoteSiteDir() {
+ // If the application does not use a site directory, then return null.
+
+ if (!hasSiteDir()) {
+ return null;
+ }
+
+ // If the application is not localized, then use the remote site path
+ // provided in the config file. Otherwise, assume that the remote path
+ // is the same as the local path.
+
+ if (!config.getBoolean(LOCALIZE_DRILL)) {
+ String drillSitePath = config.getString(SITE_DIR);
+ if (DoYUtil.isBlank(drillSitePath)) {
+ drillSitePath = drillSite.getAbsolutePath();
+ }
+ return drillSitePath;
+ }
+
+ // Work out the site directory name as above for the Drill directory.
+ // The caller must include a archive subdirectory name if required.
+
+ return "$PWD/" + config.getString(SITE_ARCHIVE_KEY);
+ }
+
+ /**
+ * Return the app ID file to use for this client run. The file is in the
+ * directory that holds the site dir (if a site dir is used), else the
+ * directory that holds Drill home (otherwise.) Not that the file does NOT go
+ * into the site dir or Drill home as we upload these directories (via
+ * archives) to DFS so we don't want to change them by adding a file.
+ * <p>
+ * It turns out that Drill allows two distinct clusters to share the same ZK
+ * root and/or cluster ID (just not the same combination), so the file name
+ * contains both parts.
+ *
+ * @param clusterId
+ * @return
+ */
+
+ public File getLocalAppIdFile() {
+ String rootDir = config.getString(DrillOnYarnConfig.ZK_ROOT);
+ String clusterId = config.getString(DrillOnYarnConfig.CLUSTER_ID);
+ String key = rootDir + "-" + clusterId;
+ String appIdFileName = key + ".appid";
+ File appIdDir;
+ if (hasSiteDir()) {
+ appIdDir = drillSite.getParentFile();
+ } else {
+ appIdDir = drillHome.getParentFile();
+ }
+ return new File(appIdDir, appIdFileName);
+ }
+
+ public boolean hasSiteDir() {
+ return drillSite != null;
+ }
+
+ public File getLocalSiteDir() {
+ return drillSite;
+ }
+
+ /**
+ * Returns the DFS path to the localized Drill archive. This is an AM-only
+ * method as it relies on an environment variable set by the client. It is set
+ * only if the application is localized, it is not set for a non-localized
+ * run.
+ *
+ * @return
+ */
+
+ public String getDrillArchiveDfsPath() {
+ return getEnv(DrillOnYarnConfig.DRILL_ARCHIVE_ENV_VAR);
+ }
+
+ /**
+ * Returns the DFS path to the localized site archive. This is an AM-only
+ * method as it relies on an environment variable set by the client. This
+ * variable is optional; if not set then the AM can infer that the application
+ * does not use a site archive (configuration files reside in
+ * $DRILL_HOME/conf), or the application is not localized.
+ *
+ * @return
+ */
+
+ public String getSiteArchiveDfsPath() {
+ return getEnv(DrillOnYarnConfig.SITE_ARCHIVE_ENV_VAR);
+ }
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/f2ac8749/drill-yarn/src/main/java/org/apache/drill/yarn/core/LaunchSpec.java
----------------------------------------------------------------------
diff --git a/drill-yarn/src/main/java/org/apache/drill/yarn/core/LaunchSpec.java b/drill-yarn/src/main/java/org/apache/drill/yarn/core/LaunchSpec.java
new file mode 100644
index 0000000..6c22874
--- /dev/null
+++ b/drill-yarn/src/main/java/org/apache/drill/yarn/core/LaunchSpec.java
@@ -0,0 +1,248 @@
+/*
+ * 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.
+ */
+package org.apache.drill.yarn.core;
+
+import java.io.IOException;
+import java.io.PrintStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.hadoop.yarn.api.ApplicationConstants;
+import org.apache.hadoop.yarn.api.ApplicationConstants.Environment;
+import org.apache.hadoop.yarn.api.records.ContainerLaunchContext;
+import org.apache.hadoop.yarn.api.records.LocalResource;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.util.Records;
+
+/**
+ * Abstract description of a remote process launch that describes the many
+ * details needed to launch a process on a remote node.
+ * <p>
+ * Based on <a href="https://github.com/hortonworks/simple-yarn-app">Simple YARN
+ * App</a>.
+ */
+
+public class LaunchSpec {
+ /**
+ * List of (key, file) pairs to be localized to the node before running the
+ * command. The file must exist in a distributed file system (such as HDFS)
+ * visible to both the client and remote node. Typically, the path is relative
+ * or absolute within the file system defined by the fs.defaultFS parameter in
+ * core-site.xml.
+ * <p>
+ * TODO: Can the value also be a URL such as
+ * <p>
+ * <code>hdfs://somehost:1234//path/to/file
+ * <p>
+ * The key is used as (what?).
+ */
+
+ public Map<String, LocalResource> resources = new HashMap<>();
+
+ /**
+ * Defines environment variables to be set on the remote host before launching
+ * the remote app. Note: do not set CLASSPATH here; use {@link #classPath}
+ * instead.
+ */
+
+ public Map<String, String> env = new HashMap<>();
+
+ /**
+ * Set to the name of the OS command to run when we wish to run a non-Java
+ * command.
+ */
+
+ public String command;
+
+ /**
+ * Set to the name of the Java main class (the one with the main method) when
+ * we wish to run a Java command.
+ */
+
+ public String mainClass;
+
+ /**
+ * Set to the application-specific class path for the Java application. These
+ * values are added to the Hadoop-provided values. These items are relative to
+ * (what?), use (what variables) to refer to the localized application
+ * directory.
+ */
+
+ public List<String> classPath = new ArrayList<>();
+
+ /**
+ * Optional VM arguments to pass to the JVM when running a Java class; ignored
+ * when running an OS command.
+ */
+
+ public List<String> vmArgs = new ArrayList<>();
+
+ /**
+ * Arguments to the remote command.
+ */
+
+ public List<String> cmdArgs = new ArrayList<>();
+
+ public LaunchSpec() {
+ }
+
+ /**
+ * Create the command line to run on the remote node. The command can either
+ * be a simple OS command (if the {@link #command} member is set) or can be a
+ * Java class (if the {@link #mainClass} member is set. If the command is
+ * Java, then we pass along optional Java VM arguments.
+ * <p>
+ * In all cases we append arguments to the command itself, and redirect stdout
+ * and stderr to log files.
+ *
+ * @return the complete command string
+ */
+
+ public String getCommand() {
+ List<String> cmd = new ArrayList<>();
+ if (command != null) {
+ cmd.add(command);
+ } else {
+ assert mainClass != null;
+
+ // JAVA_HOME is provided by YARN.
+
+ cmd.add(Environment.JAVA_HOME.$$() + "/bin/java");
+ cmd.addAll(vmArgs);
+ if (!classPath.isEmpty()) {
+ cmd.add("-cp");
+ cmd.add(DoYUtil.join(":", classPath));
+ }
+ cmd.add(mainClass);
+ }
+ cmd.addAll(cmdArgs);
+ cmd.add("1>" + ApplicationConstants.LOG_DIR_EXPANSION_VAR + "/stdout");
+ cmd.add("2>" + ApplicationConstants.LOG_DIR_EXPANSION_VAR + "/stderr");
+
+ // Java 8
+ // return String.join( " ", cmd );
+ return DoYUtil.join(" ", cmd);
+ }
+
+ /**
+ * Given this generic description of an application, create the detailed YARN
+ * application submission context required to launch the application.
+ *
+ * @param conf
+ * the YARN configuration obtained by reading the Hadoop
+ * configuration files
+ * @return the completed application launch context for the given application
+ * @throws IOException
+ * if localized resources are not found in the distributed file
+ * system (such as HDFS)
+ */
+
+ public ContainerLaunchContext createLaunchContext(YarnConfiguration conf)
+ throws IOException {
+ // Set up the container launch context
+ ContainerLaunchContext container = Records
+ .newRecord(ContainerLaunchContext.class);
+
+ // Set up the list of commands to run. Here, we assume that we run only
+ // one command.
+
+ container.setCommands(Collections.singletonList(getCommand()));
+
+ // Add localized resources
+
+ container.setLocalResources(resources);
+
+ // Environment.
+
+ container.setEnvironment(env);
+
+ return container;
+ }
+
+ public void dump(PrintStream out) {
+ if (command != null) {
+ out.print("Command: ");
+ out.println(command);
+ }
+ if (mainClass != null) {
+ out.print("Main Class: ");
+ out.println(mainClass);
+ out.println("VM Args:");
+ if (vmArgs.isEmpty()) {
+ out.println(" None");
+ } else {
+ for (String item : vmArgs) {
+ out.print(" ");
+ out.println(item);
+ }
+ }
+ out.println("Class Path:");
+ if (classPath.isEmpty()) {
+ out.println(" None");
+ } else {
+ for (String item : classPath) {
+ out.print(" ");
+ out.println(item);
+ }
+ }
+ }
+ out.println("Program Args:");
+ if (cmdArgs.isEmpty()) {
+ out.println(" None");
+ } else {
+ for (String item : cmdArgs) {
+ out.print(" ");
+ out.println(item);
+ }
+ }
+ out.println("Environment:");
+ if (env.isEmpty()) {
+ out.println(" None");
+ } else {
+ for (String key : env.keySet()) {
+ out.print(" ");
+ out.print(key);
+ out.print("=");
+ out.println(env.get(key));
+ }
+ }
+ out.println("Resources: ");
+ if (resources.isEmpty()) {
+ out.println(" None");
+ } else {
+ for (String key : resources.keySet()) {
+ out.print(" Key: ");
+ out.println(key);
+ LocalResource resource = resources.get(key);
+ out.print(" URL: ");
+ out.println(resource.getResource().toString());
+ out.print(" Size: ");
+ out.println(resource.getSize());
+ out.print(" Timestamp: ");
+ out.println(DoYUtil.toIsoTime(resource.getTimestamp()));
+ out.print(" Type: ");
+ out.println(resource.getType().toString());
+ out.print(" Visiblity: ");
+ out.println(resource.getVisibility().toString());
+ }
+ }
+ }
+}