http://git-wip-us.apache.org/repos/asf/drill/blob/f2ac8749/drill-yarn/src/main/java/org/apache/drill/yarn/core/NameValuePair.java ---------------------------------------------------------------------- diff --git a/drill-yarn/src/main/java/org/apache/drill/yarn/core/NameValuePair.java b/drill-yarn/src/main/java/org/apache/drill/yarn/core/NameValuePair.java new file mode 100644 index 0000000..5872ab9 --- /dev/null +++ b/drill-yarn/src/main/java/org/apache/drill/yarn/core/NameValuePair.java @@ -0,0 +1,46 @@ +/* + * 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 NameValuePair { + private String name; + private Object value; + + public NameValuePair(String name, Object value) { + this.name = name; + this.value = value; + } + + public String getName() { + return name; + } + + public Object getValue() { + return value; + } + + public String getQuotedValue() { + if (value == null) { + return "<unset>"; + } + if (value instanceof String) { + return "\"" + value + "\""; + } + return value.toString(); + } +}
http://git-wip-us.apache.org/repos/asf/drill/blob/f2ac8749/drill-yarn/src/main/java/org/apache/drill/yarn/core/YarnClientException.java ---------------------------------------------------------------------- diff --git a/drill-yarn/src/main/java/org/apache/drill/yarn/core/YarnClientException.java b/drill-yarn/src/main/java/org/apache/drill/yarn/core/YarnClientException.java new file mode 100644 index 0000000..62dd468 --- /dev/null +++ b/drill-yarn/src/main/java/org/apache/drill/yarn/core/YarnClientException.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 YarnClientException extends Exception { + private static final long serialVersionUID = -1411110715738266578L; + + public YarnClientException(String msg) { + super(msg); + } + + public YarnClientException(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/YarnRMClient.java ---------------------------------------------------------------------- diff --git a/drill-yarn/src/main/java/org/apache/drill/yarn/core/YarnRMClient.java b/drill-yarn/src/main/java/org/apache/drill/yarn/core/YarnRMClient.java new file mode 100644 index 0000000..8905ce3 --- /dev/null +++ b/drill-yarn/src/main/java/org/apache/drill/yarn/core/YarnRMClient.java @@ -0,0 +1,207 @@ +/* + * 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 org.apache.hadoop.security.token.Token; +import org.apache.hadoop.yarn.api.protocolrecords.GetNewApplicationResponse; +import org.apache.hadoop.yarn.api.records.ApplicationAttemptId; +import org.apache.hadoop.yarn.api.records.ApplicationId; +import org.apache.hadoop.yarn.api.records.ApplicationReport; +import org.apache.hadoop.yarn.api.records.ApplicationSubmissionContext; +import org.apache.hadoop.yarn.api.records.YarnApplicationState; +import org.apache.hadoop.yarn.client.api.YarnClient; +import org.apache.hadoop.yarn.client.api.YarnClientApplication; +import org.apache.hadoop.yarn.conf.YarnConfiguration; +import org.apache.hadoop.yarn.exceptions.YarnException; +import org.apache.hadoop.yarn.security.AMRMTokenIdentifier; + +/** + * YARN resource manager client implementation for Drill. Provides a wrapper + * around the YARN client interface to the Resource Manager. Used by the client + * app to start the Drill application master. + * <p> + * Based on + * <a href="https://github.com/hortonworks/simple-yarn-app";>simple-yarn-app</a> + */ + +public class YarnRMClient { + private YarnConfiguration conf; + private YarnClient yarnClient; + + /** + * Application ID. Semantics are such that each session of Drill-on-YARN works + * with no more than one application ID. + */ + + private ApplicationId appId; + private YarnClientApplication app; + + public YarnRMClient() { + this(new YarnConfiguration()); + } + + public YarnRMClient(ApplicationId appId) { + this(); + this.appId = appId; + } + + public YarnRMClient(YarnConfiguration conf) { + this.conf = conf; + yarnClient = YarnClient.createYarnClient(); + yarnClient.init(conf); + yarnClient.start(); + } + + public GetNewApplicationResponse createAppMaster() + throws YarnClientException { + // Create application via yarnClient + // Response is a new application ID along with cluster capacity info + + try { + app = yarnClient.createApplication(); + } catch (YarnException | IOException e) { + throw new YarnClientException("Create application failed", e); + } + GetNewApplicationResponse response = app.getNewApplicationResponse(); + appId = response.getApplicationId(); + return response; + } + + public void submitAppMaster(AppSpec spec) throws YarnClientException { + if (app == null) { + throw new IllegalStateException("call createAppMaster( ) first"); + } + + ApplicationSubmissionContext appContext; + try { + appContext = spec.createAppLaunchContext(conf, app); + } catch (IOException e) { + throw new YarnClientException("Create app launch context failed", e); + } + + // Submit application + try { + yarnClient.submitApplication(appContext); + } catch (YarnException | IOException e) { + throw new YarnClientException("Submit application failed", e); + } + } + + public ApplicationId getAppId() { + return appId; + } + + public ApplicationReport getAppReport() throws YarnClientException { + try { + return yarnClient.getApplicationReport(appId); + } catch (YarnException | IOException e) { + throw new YarnClientException("Get application report failed", e); + } + } + + /** + * Waits for the application to start. This version is somewhat informal, the + * intended use is when debugging unmanaged applications. + * + * @throws YarnClientException + */ + public ApplicationAttemptId waitForStart() throws YarnClientException { + ApplicationReport appReport; + YarnApplicationState appState; + ApplicationAttemptId attemptId; + for (;;) { + appReport = getAppReport(); + appState = appReport.getYarnApplicationState(); + attemptId = appReport.getCurrentApplicationAttemptId(); + if (appState != YarnApplicationState.NEW + && appState != YarnApplicationState.NEW_SAVING + && appState != YarnApplicationState.SUBMITTED) { + break; + } + System.out.println("App State: " + appState); + try { + Thread.sleep(1000); + } catch (InterruptedException e) { + // Should never occur. + } + } + if (appState != YarnApplicationState.ACCEPTED) { + throw new YarnClientException( + "Application start failed with status " + appState); + } + + return attemptId; + } + + /** + * Wait for the application to enter one of the completion states. This is an + * informal implementation useful for testing. + * + * @throws YarnClientException + */ + + public void waitForCompletion() throws YarnClientException { + ApplicationReport appReport; + YarnApplicationState appState; + for (;;) { + appReport = getAppReport(); + appState = appReport.getYarnApplicationState(); + if (appState == YarnApplicationState.FINISHED + || appState == YarnApplicationState.KILLED + || appState == YarnApplicationState.FAILED) { + break; + } + try { + Thread.sleep(100); + } catch (InterruptedException e) { + // Should never occur. + } + } + + System.out.println("Application " + appId + " finished with" + " state " + + appState + " at " + appReport.getFinishTime()); + } + + public Token<AMRMTokenIdentifier> getAMRMToken() throws YarnClientException { + try { + return yarnClient.getAMRMToken(appId); + } catch (YarnException | IOException e) { + throw new YarnClientException("Get AM/RM token failed", e); + } + } + + /** + * Return standard class path entries from the YARN application class path. + */ + + public String[] getYarnAppClassPath() { + return conf.getStrings(YarnConfiguration.YARN_APPLICATION_CLASSPATH, + YarnConfiguration.DEFAULT_YARN_APPLICATION_CLASSPATH); + } + + public void killApplication() throws YarnClientException { + try { + yarnClient.killApplication(appId); + } catch (YarnException | IOException e) { + throw new YarnClientException( + "Kill failed for application: " + appId.toString()); + } + } +} http://git-wip-us.apache.org/repos/asf/drill/blob/f2ac8749/drill-yarn/src/main/java/org/apache/drill/yarn/core/package-info.java ---------------------------------------------------------------------- diff --git a/drill-yarn/src/main/java/org/apache/drill/yarn/core/package-info.java b/drill-yarn/src/main/java/org/apache/drill/yarn/core/package-info.java new file mode 100644 index 0000000..aaa0fff --- /dev/null +++ b/drill-yarn/src/main/java/org/apache/drill/yarn/core/package-info.java @@ -0,0 +1,24 @@ +/* + * 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. + */ + +/** + * Holds functionality common to the Drill-on-YARN client and Application Master (AM). + * Includes configuration, utilities, and wrappers around various YARN data classes. + */ + +package org.apache.drill.yarn.core; \ 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/package-info.java ---------------------------------------------------------------------- diff --git a/drill-yarn/src/main/java/org/apache/drill/yarn/package-info.java b/drill-yarn/src/main/java/org/apache/drill/yarn/package-info.java new file mode 100644 index 0000000..170dfa8 --- /dev/null +++ b/drill-yarn/src/main/java/org/apache/drill/yarn/package-info.java @@ -0,0 +1,35 @@ +/* + * 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. + */ + +/** + * Hosts Apache Drill under Apache Hadoop YARN. Consists of two main + * components as required by YARN: a client application which uses YARN to + * start the Drill cluster, and an Application Master (AM) which manages + * the cluster. The AM in turn starts, manages and stops drillbits. + * <p> + * Much of the functionality is simply plumbing to get YARN to do what is + * needed. The core of the AM is a "cluster controller" which starts, + * monitors and stops Drillbits, tracking their state transitions though + * the several lifecycle stages that result. + * <p> + * Note about logs here: Drill-on-YARN is a YARN application and so it + * uses the same logging system used by the YARN code. This is different + * than that used by Drill. + */ + +package org.apache.drill.yarn; \ 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/zk/AMRegistry.java ---------------------------------------------------------------------- diff --git a/drill-yarn/src/main/java/org/apache/drill/yarn/zk/AMRegistry.java b/drill-yarn/src/main/java/org/apache/drill/yarn/zk/AMRegistry.java new file mode 100644 index 0000000..9cc95e5 --- /dev/null +++ b/drill-yarn/src/main/java/org/apache/drill/yarn/zk/AMRegistry.java @@ -0,0 +1,145 @@ +/* + * 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.zk; + +import org.apache.zookeeper.CreateMode; +import org.apache.zookeeper.KeeperException.NodeExistsException; + +/** + * Register this App Master in ZK to prevent duplicates. + * <p> + * Possible enhancement is to put the registry in some well-known location, such + * as /drill-am, + */ +public class AMRegistry { + private static final String AM_REGISTRY = "/drill-on-yarn"; + + private ZKClusterCoordinator zkCoord; + @SuppressWarnings("unused") + private String amHost; + @SuppressWarnings("unused") + private int amPort; + @SuppressWarnings("unused") + private String amAppId; + + private String zkRoot; + + private String clusterId; + + public AMRegistry(ZKClusterCoordinator zkCoord) { + this.zkCoord = zkCoord; + } + + public void useLocalRegistry(String zkRoot, String clusterId) { + this.zkRoot = zkRoot; + this.clusterId = clusterId; + } + + /** + * Register this AM as an ephemeral znode in ZK. The structure of ZK is as + * follows: + * + * <pre> + * /drill + * . <cluster-id> + * . . <Drillbit GUID> (Value is Proto-encoded drillbit info) + * . drill-on-yarn + * . . <cluster-id> (value: amHost:port) + * </pre> + * <p> + * The structure acknowledges that the cluster-id znode may be renamed, and + * there may be multiple cluster IDs for a single drill root node. (Odd, but + * supported.) To address this, we put the AM registrations in their own + * (persistent) znode: drill-on-yarn. Each is keyed by the cluster ID (so we + * can find it), and holds the host name, HTTP port and Application ID of the + * AM. + * <p> + * When the AM starts, it atomically checks and sets the AM registration. If + * another AM already is running, then this AM will fail, displaying a log + * error message with the host, port and (most importantly) app ID so the user + * can locate the problem. + * + * @throws ZKRuntimeException + */ + + public void register(String amHost, int amPort, String amAppId) + throws ZKRuntimeException { + this.amHost = amHost; + this.amPort = amPort; + this.amAppId = amAppId; + try { + + // The znode to hold AMs may or may not exist. Create it if missing. + + try { + zkCoord.getCurator().create().withMode(CreateMode.PERSISTENT) + .forPath(AM_REGISTRY, new byte[0]); + } catch (NodeExistsException e) { + // OK + } + + // Try to create the AM registration. + + String amPath = AM_REGISTRY + "/" + clusterId; + String content = amHost + ":" + Integer.toString(amPort) + ":" + amAppId; + try { + zkCoord.getCurator().create().withMode(CreateMode.EPHEMERAL) + .forPath(amPath, content.getBytes("UTF-8")); + } catch (NodeExistsException e) { + + // ZK says that a node exists, which means that another AM is already + // running. + // Display an error, handling the case where the AM just disappeared, + // the + // registration is badly formatted, etc. + + byte data[] = zkCoord.getCurator().getData().forPath(amPath); + String existing; + if (data == null) { + existing = "Unknown"; + } else { + String packed = new String(data, "UTF-8"); + String unpacked[] = packed.split(":"); + if (unpacked.length < 3) { + existing = packed; + } else { + existing = unpacked[0] + ", port: " + unpacked[1] + + ", Application ID: " + unpacked[2]; + } + } + + // Die with a clear (we hope!) error message. + + throw new ZKRuntimeException( + "FAILED! An Application Master already exists for " + zkRoot + "/" + + clusterId + " on host: " + existing); + } + } catch (ZKRuntimeException e) { + + // Something bad happened with ZK. + + throw e; + } catch (Exception e) { + + // Something bad happened with ZK. + + throw new ZKRuntimeException("Failed to create AM registration node", e); + } + } + +} http://git-wip-us.apache.org/repos/asf/drill/blob/f2ac8749/drill-yarn/src/main/java/org/apache/drill/yarn/zk/ZKClusterCoordinator.java ---------------------------------------------------------------------- diff --git a/drill-yarn/src/main/java/org/apache/drill/yarn/zk/ZKClusterCoordinator.java b/drill-yarn/src/main/java/org/apache/drill/yarn/zk/ZKClusterCoordinator.java new file mode 100644 index 0000000..7c5f5f3 --- /dev/null +++ b/drill-yarn/src/main/java/org/apache/drill/yarn/zk/ZKClusterCoordinator.java @@ -0,0 +1,318 @@ +/* + * 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.zk; + +import static com.google.common.base.Throwables.propagate; +import static com.google.common.collect.Collections2.transform; + +import java.io.IOException; +import java.util.Collection; +import java.util.Collections; +import java.util.HashSet; +import java.util.Set; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.apache.curator.RetryPolicy; +import org.apache.curator.framework.CuratorFramework; +import org.apache.curator.framework.CuratorFrameworkFactory; +import org.apache.curator.framework.state.ConnectionState; +import org.apache.curator.framework.state.ConnectionStateListener; +import org.apache.curator.retry.RetryNTimes; +import org.apache.curator.x.discovery.ServiceCache; +import org.apache.curator.x.discovery.ServiceDiscovery; +import org.apache.curator.x.discovery.ServiceDiscoveryBuilder; +import org.apache.curator.x.discovery.ServiceInstance; +import org.apache.curator.x.discovery.details.ServiceCacheListener; +import org.apache.drill.common.AutoCloseables; +import org.apache.drill.exec.coord.ClusterCoordinator; +import org.apache.drill.exec.coord.DistributedSemaphore; +import org.apache.drill.exec.coord.DrillServiceInstanceHelper; +import org.apache.drill.exec.coord.store.CachingTransientStoreFactory; +import org.apache.drill.exec.coord.store.TransientStore; +import org.apache.drill.exec.coord.store.TransientStoreConfig; +import org.apache.drill.exec.coord.store.TransientStoreFactory; +import org.apache.drill.exec.coord.zk.ZKRegistrationHandle; +import org.apache.drill.exec.coord.zk.ZkDistributedSemaphore; +import org.apache.drill.exec.coord.zk.ZkEphemeralStore; +import org.apache.drill.exec.coord.zk.ZkTransientStoreFactory; +import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint; +import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint.State; + +import com.google.common.base.Function; + +/** + * Manages cluster coordination utilizing zookeeper. + * <p> + * This is a clone of the Drill class + * org.apache.drill.exec.coord.zk.ZKClusterCoordinator with a number of + * modifications: + * <ul> + * <li>Removed dependency on the Drill config system. That system uses Google's + * Guava library version 18, which conflicts with the earlier versions used by + * YARN and Hadoop, which resulted in runtime undefined method exceptions.</li> + * <li>Instead of getting config information out of the Drill config, the + * parameters are instead passed directly.</li> + * <li>Adds support for the drillbits registered event which was neither needed + * nor implemented by Drill.</li> + * <li>Use the YARN logging system instead of Drill's.</li> + * </ul> + * <p> + * This class should be replaced by the Drill version if/when the Guava + * conflicts can be resolved (and when registered Drillbit notifications are + * added to the Drill version.) + */ + +public class ZKClusterCoordinator extends ClusterCoordinator { + + protected static final Log logger = LogFactory + .getLog(ZKClusterCoordinator.class); + + private CuratorFramework curator; + private ServiceDiscovery<DrillbitEndpoint> discovery; + private volatile Collection<DrillbitEndpoint> endpoints = Collections + .emptyList(); + private final String serviceName; + private final CountDownLatch initialConnection = new CountDownLatch(1); + private final TransientStoreFactory factory; + private ServiceCache<DrillbitEndpoint> serviceCache; + + public ZKClusterCoordinator(String connect, String zkRoot, String clusterId, + int retryCount, int retryDelayMs, int connectTimeoutMs) + throws IOException { + logger.debug("ZK connect: " + connect + ", zkRoot: " + zkRoot + + ", clusterId: " + clusterId); + + this.serviceName = clusterId; + RetryPolicy rp = new RetryNTimes(retryCount, retryDelayMs); + curator = CuratorFrameworkFactory.builder().namespace(zkRoot) + .connectionTimeoutMs(connectTimeoutMs).retryPolicy(rp) + .connectString(connect).build(); + curator.getConnectionStateListenable() + .addListener(new InitialConnectionListener()); + curator.start(); + discovery = newDiscovery(); + factory = CachingTransientStoreFactory + .of(new ZkTransientStoreFactory(curator)); + } + + public CuratorFramework getCurator() { + return curator; + } + + @Override + public void start(long millisToWait) throws Exception { + logger.debug("Starting ZKClusterCoordination."); + discovery.start(); + + if (millisToWait != 0) { + boolean success = this.initialConnection.await(millisToWait, + TimeUnit.MILLISECONDS); + if (!success) { + throw new IOException(String.format( + "Failure to connect to the zookeeper cluster service within the allotted time of %d milliseconds.", + millisToWait)); + } + } else { + this.initialConnection.await(); + } + + serviceCache = discovery.serviceCacheBuilder().name(serviceName).build(); + serviceCache.addListener(new EndpointListener()); + serviceCache.start(); + updateEndpoints(); + } + + private class InitialConnectionListener implements ConnectionStateListener { + + @Override + public void stateChanged(CuratorFramework client, + ConnectionState newState) { + if (newState == ConnectionState.CONNECTED) { + initialConnection.countDown(); + client.getConnectionStateListenable().removeListener(this); + } + } + + } + + private class EndpointListener implements ServiceCacheListener { + @Override + public void stateChanged(CuratorFramework client, + ConnectionState newState) { + } + + @Override + public void cacheChanged() { + logger.debug("Got cache changed --> updating endpoints"); + updateEndpoints(); + } + } + + @Override + public void close() throws Exception { + // discovery attempts to close its caches(ie serviceCache) already. however, + // being good citizens we make sure to + // explicitly close serviceCache. Not only that we make sure to close + // serviceCache before discovery to prevent + // double releasing and disallowing jvm to spit bothering warnings. simply + // put, we are great! + AutoCloseables.close(serviceCache, discovery, curator, factory); + } + + @Override + public RegistrationHandle register(DrillbitEndpoint data) { + try { + ServiceInstance<DrillbitEndpoint> serviceInstance = newServiceInstance( + data); + discovery.registerService(serviceInstance); + return new ZKRegistrationHandle(serviceInstance.getId(), data); + } catch (Exception e) { + throw propagate(e); + } + } + + @Override + public void unregister(RegistrationHandle handle) { + if (!(handle instanceof ZKRegistrationHandle)) { + throw new UnsupportedOperationException( + "Unknown handle type: " + handle.getClass().getName()); + } + + // when Drillbit is unregistered, clean all the listeners registered in CC. + this.listeners.clear(); + + ZKRegistrationHandle h = (ZKRegistrationHandle) handle; + try { + ServiceInstance<DrillbitEndpoint> serviceInstance = ServiceInstance + .<DrillbitEndpoint> builder().address("").port(0).id(h.id) + .name(serviceName).build(); + discovery.unregisterService(serviceInstance); + } catch (Exception e) { + propagate(e); + } + } + + @Override + public Collection<DrillbitEndpoint> getAvailableEndpoints() { + return this.endpoints; + } + + @Override + public DistributedSemaphore getSemaphore(String name, int maximumLeases) { + return new ZkDistributedSemaphore(curator, "/semaphore/" + name, + maximumLeases); + } + + @Override + public <V> TransientStore<V> getOrCreateTransientStore( + final TransientStoreConfig<V> config) { + final ZkEphemeralStore<V> store = (ZkEphemeralStore<V>) factory + .getOrCreateStore(config); + return store; + } + + private synchronized void updateEndpoints() { + try { + Collection<DrillbitEndpoint> newDrillbitSet = transform( + discovery.queryForInstances(serviceName), + new Function<ServiceInstance<DrillbitEndpoint>, DrillbitEndpoint>() { + @Override + public DrillbitEndpoint apply( + ServiceInstance<DrillbitEndpoint> input) { + return input.getPayload(); + } + }); + + // set of newly dead bits : original bits - new set of active bits. + Set<DrillbitEndpoint> unregisteredBits = new HashSet<>(endpoints); + unregisteredBits.removeAll(newDrillbitSet); + + // Set of newly live bits : new set of active bits - original bits. + Set<DrillbitEndpoint> registeredBits = new HashSet<>(newDrillbitSet); + registeredBits.removeAll(endpoints); + + endpoints = newDrillbitSet; + + if (logger.isDebugEnabled()) { + StringBuilder builder = new StringBuilder(); + builder.append("Active drillbit set changed. Now includes "); + builder.append(newDrillbitSet.size()); + builder.append(" total bits."); + if (!newDrillbitSet.isEmpty()) { + builder.append(" New active drillbits: \n"); + } + for (DrillbitEndpoint bit : newDrillbitSet) { + builder.append('\t'); + builder.append(bit.getAddress()); + builder.append(':'); + builder.append(bit.getUserPort()); + builder.append(':'); + builder.append(bit.getControlPort()); + builder.append(':'); + builder.append(bit.getDataPort()); + builder.append('\n'); + } + logger.debug(builder.toString()); + } + + // Notify the drillbit listener for newly unregistered bits. + if (!(unregisteredBits.isEmpty())) { + drillbitUnregistered(unregisteredBits); + } + // Notify the drillbit listener for newly registered bits. + if (!(registeredBits.isEmpty())) { + drillbitRegistered(registeredBits); + } + + } catch (Exception e) { + logger.error("Failure while update Drillbit service location cache.", e); + } + } + + protected ServiceInstance<DrillbitEndpoint> newServiceInstance( + DrillbitEndpoint endpoint) throws Exception { + return ServiceInstance.<DrillbitEndpoint> builder().name(serviceName) + .payload(endpoint).build(); + } + + protected ServiceDiscovery<DrillbitEndpoint> newDiscovery() { + return ServiceDiscoveryBuilder.builder(DrillbitEndpoint.class).basePath("/") + .client(curator).serializer(DrillServiceInstanceHelper.SERIALIZER) + .build(); + } + + @Override + public Collection<DrillbitEndpoint> getOnlineEndPoints() { + + // Not used in DoY + + throw new UnsupportedOperationException(); + } + + @Override + public RegistrationHandle update(RegistrationHandle handle, State state) { + + // Not used in DoY + + throw new UnsupportedOperationException(); + } + +} http://git-wip-us.apache.org/repos/asf/drill/blob/f2ac8749/drill-yarn/src/main/java/org/apache/drill/yarn/zk/ZKClusterCoordinatorDriver.java ---------------------------------------------------------------------- diff --git a/drill-yarn/src/main/java/org/apache/drill/yarn/zk/ZKClusterCoordinatorDriver.java b/drill-yarn/src/main/java/org/apache/drill/yarn/zk/ZKClusterCoordinatorDriver.java new file mode 100644 index 0000000..3f83ff2 --- /dev/null +++ b/drill-yarn/src/main/java/org/apache/drill/yarn/zk/ZKClusterCoordinatorDriver.java @@ -0,0 +1,315 @@ +/* + * 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.zk; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + +import org.apache.curator.framework.CuratorFramework; +import org.apache.curator.framework.state.ConnectionState; +import org.apache.curator.framework.state.ConnectionStateListener; +import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint; +import org.apache.drill.exec.work.foreman.DrillbitStatusListener; +import org.apache.drill.yarn.appMaster.AMRegistrar; + +/** + * Driver class for the ZooKeeper cluster coordinator. Provides defaults for + * most options, but allows customizing each. Provides a {@link #build()} method + * to create <i>and start</i> the ZK service. Obtains the initial set of + * Drillbits (which should be empty for a YARN-defined cluster) which can be + * retrieved after building. + * <p> + * Maintains the ZK connection and monitors for disconnect. This class simply + * detects a disconnect timeout, it does not send a disconnect event itself to + * avoid creating a timer thread just for this purpose. Instead, the caller can + * poll {@link #hasFailed()}. + * <p> + * Defaults match those in Drill. (Actual Drill defaults are not yet used due to + * code incompatibility issues.) + */ + +public class ZKClusterCoordinatorDriver implements AMRegistrar { + private static final Pattern ZK_COMPLEX_STRING = Pattern + .compile("(^.*?)/(.*)/([^/]*)$"); + + // Defaults are taken from java-exec's drill-module.conf + + private String connect = "localhost:2181"; + private String clusterId = "drillbits1"; + private String zkRoot = "drill"; + private int retryCount = 7200; + private int connectTimeoutMs = 5_000; + private int retryDelayMs = 500; + + // Default timeout before we declare that ZK is down: 2 minutes. + + private int failureTimeoutMs = 120_000; + + // Maximum ZK startup wait defaults to 30 seconds. It is only 10 seconds + // in the Drill implementation. + + private int maxStartWaitMs = 30_000; + + // Expected ports used to match ZK registries with + // containers. ZK lists the ports as part of its key, we have to anticipate + // these values in order to match. + + private int userPort = 31010; + private int controlPort = 31011; + private int dataPort = 31012; + + private List<DrillbitEndpoint> initialEndpoints; + private ConnectionStateListener stateListener = new ConnectionStateListener() { + + @Override + public void stateChanged(CuratorFramework client, + ConnectionState newState) { + ZKClusterCoordinatorDriver.this.stateChanged(newState); + } + }; + + private ZKClusterCoordinator zkCoord; + + private long connectionLostTime; + + private AMRegistry amRegistry; + + public ZKClusterCoordinatorDriver() { + } + + /** + * Specify connect string in the form: host:/zkRoot/clusterId + * + * @param connect + * @return + * @throws ZKConfigException + */ + public ZKClusterCoordinatorDriver setConnect(String connect) + throws ZKConfigException { + + // check if this is a complex zk string. If so, parse into components. + Matcher m = ZK_COMPLEX_STRING.matcher(connect); + if (!m.matches()) { + throw new ZKConfigException("Bad connect string: " + connect); + } + this.connect = m.group(1); + zkRoot = m.group(2); + clusterId = m.group(3); + return this; + } + + public ZKClusterCoordinatorDriver setConnect(String connect, String zkRoot, + String clusterId) { + this.connect = connect; + this.zkRoot = zkRoot; + this.clusterId = clusterId; + return this; + } + + public ZKClusterCoordinatorDriver setRetryCount(int n) { + retryCount = n; + return this; + } + + public ZKClusterCoordinatorDriver setConnectTimeoutMs(int ms) { + connectTimeoutMs = ms; + return this; + } + + public ZKClusterCoordinatorDriver setRetryDelayMs(int ms) { + retryDelayMs = ms; + return this; + } + + public ZKClusterCoordinatorDriver setMaxStartWaitMs(int ms) { + maxStartWaitMs = ms; + return this; + } + + public ZKClusterCoordinatorDriver setFailureTimoutMs(int ms) { + failureTimeoutMs = ms; + return this; + } + + public ZKClusterCoordinatorDriver setPorts(int userPort, int controlPort, + int dataPort) { + this.userPort = userPort; + this.controlPort = controlPort; + this.dataPort = dataPort; + return this; + } + + /** + * Builds and starts the ZooKeeper cluster coordinator, translating any errors + * that occur. After this call, the listener will start receiving messages. + * + * @return + * @throws ZKRuntimeException + * if ZK startup fails + */ + public ZKClusterCoordinatorDriver build() throws ZKRuntimeException { + try { + zkCoord = new ZKClusterCoordinator(connect, zkRoot, clusterId, retryCount, + retryDelayMs, connectTimeoutMs); + } catch (IOException e) { + throw new ZKRuntimeException( + "Failed to initialize the ZooKeeper cluster coordination", e); + } + try { + zkCoord.start(maxStartWaitMs); + } catch (Exception e) { + throw new ZKRuntimeException( + "Failed to start the ZooKeeper cluster coordination after " + + maxStartWaitMs + " ms.", + e); + } + initialEndpoints = new ArrayList<>(zkCoord.getAvailableEndpoints()); + zkCoord.getCurator().getConnectionStateListenable() + .addListener(stateListener); + amRegistry = new AMRegistry(zkCoord); + amRegistry.useLocalRegistry(zkRoot, clusterId); + return this; + } + + public void addDrillbitListener(DrillbitStatusListener listener) { + zkCoord.addDrillbitStatusListener(listener); + } + + public void removeDrillbitListener(DrillbitStatusListener listener) { + zkCoord.removeDrillbitStatusListener(listener); + } + + /** + * Returns the set of Drillbits registered at the time of the {@link #build()} + * call. Should be empty for a cluster managed by YARN. + * + * @return + */ + + public List<DrillbitEndpoint> getInitialEndpoints() { + return initialEndpoints; + } + + /** + * Convenience method to convert a Drillbit to a string. Note that ZK does not + * advertise the HTTP port, so it does not appear in the generated string. + * + * @param bit + * @return + */ + + public static String asString(DrillbitEndpoint bit) { + return formatKey(bit.getAddress(), bit.getUserPort(), bit.getControlPort(), + bit.getDataPort()); + } + + public String toKey(String host) { + return formatKey(host, userPort, controlPort, dataPort); + } + + public static String formatKey(String host, int userPort, int controlPort, + int dataPort) { + StringBuilder buf = new StringBuilder(); + buf.append(host).append(":").append(userPort).append(':') + .append(controlPort).append(':').append(dataPort); + return buf.toString(); + } + + /** + * Translate ZK connection events into a connected/disconnected state along + * with the time of the first disconnect not followed by a connect. + * + * @param newState + */ + + protected void stateChanged(ConnectionState newState) { + switch (newState) { + case CONNECTED: + case READ_ONLY: + case RECONNECTED: + if (connectionLostTime != 0) { + ZKClusterCoordinator.logger.info("ZK connection regained"); + } + connectionLostTime = 0; + break; + case LOST: + case SUSPENDED: + if (connectionLostTime == 0) { + ZKClusterCoordinator.logger.info("ZK connection lost"); + connectionLostTime = System.currentTimeMillis(); + } + break; + } + } + + /** + * Reports our best guess as to whether ZK has failed. We assume ZK has failed + * if we received a connection lost notification without a subsequent connect + * notification, and we received the disconnect notification log enough ago + * that we assume that a timeout has occurred. + * + * @return + */ + + public boolean hasFailed() { + if (connectionLostTime == 0) { + return false; + } + return System.currentTimeMillis() - connectionLostTime > failureTimeoutMs; + } + + public long getLostConnectionDurationMs() { + if (connectionLostTime == 0) { + return 0; + } + return System.currentTimeMillis() - connectionLostTime; + } + + public void close() { + if (zkCoord == null) { + return; + } + zkCoord.getCurator().getConnectionStateListenable() + .removeListener(stateListener); + try { + zkCoord.close(); + } catch (Exception e) { + ZKClusterCoordinator.logger.error("Error occurred on ZK close, ignored", + e); + } + zkCoord = null; + } + + @Override + public void register(String amHost, int amPort, String appId) + throws AMRegistrationException { + try { + amRegistry.register(amHost, amPort, appId); + } catch (ZKRuntimeException e) { + throw new AMRegistrationException(e); + } + } + + @Override + public void deregister() { + // Nothing to do: ZK does it for us. + } +} http://git-wip-us.apache.org/repos/asf/drill/blob/f2ac8749/drill-yarn/src/main/java/org/apache/drill/yarn/zk/ZKConfigException.java ---------------------------------------------------------------------- diff --git a/drill-yarn/src/main/java/org/apache/drill/yarn/zk/ZKConfigException.java b/drill-yarn/src/main/java/org/apache/drill/yarn/zk/ZKConfigException.java new file mode 100644 index 0000000..700d84b --- /dev/null +++ b/drill-yarn/src/main/java/org/apache/drill/yarn/zk/ZKConfigException.java @@ -0,0 +1,26 @@ +/* + * 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.zk; + +public class ZKConfigException extends Exception { + private static final long serialVersionUID = 1L; + + public ZKConfigException(String msg) { + super(msg); + } +} http://git-wip-us.apache.org/repos/asf/drill/blob/f2ac8749/drill-yarn/src/main/java/org/apache/drill/yarn/zk/ZKRegistry.java ---------------------------------------------------------------------- diff --git a/drill-yarn/src/main/java/org/apache/drill/yarn/zk/ZKRegistry.java b/drill-yarn/src/main/java/org/apache/drill/yarn/zk/ZKRegistry.java new file mode 100644 index 0000000..0426578 --- /dev/null +++ b/drill-yarn/src/main/java/org/apache/drill/yarn/zk/ZKRegistry.java @@ -0,0 +1,582 @@ +/* + * 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.zk; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint; +import org.apache.drill.exec.work.foreman.DrillbitStatusListener; +import org.apache.drill.yarn.appMaster.AMWrapperException; +import org.apache.drill.yarn.appMaster.EventContext; +import org.apache.drill.yarn.appMaster.Pollable; +import org.apache.drill.yarn.appMaster.RegistryHandler; +import org.apache.drill.yarn.appMaster.Task; +import org.apache.drill.yarn.appMaster.TaskLifecycleListener; + +/** + * AM-specific implementation of a Drillbit registry backed by ZooKeeper. + * Listens to ZK events for registering a Drillbit and deregistering. Alerts the + * Cluster Controller of these events. + * <p> + * Locking strategy: Receives events from both ZK and the cluster controller, + * both of which must be synchronized. To prevent deadlocks, this class NEVER + * calls into the cluster controller while holding a lock. This prevents the + * following: + * <p> + * ClusterController --> ZKRegistry (OK) <br> + * ZK --> ZKRegistry (OK) <br> + * ZK --> ZKRegistry --> Cluster Controller (bad) + * <p> + * In the case of registration, ZK calls the registry which must alert the + * cluster controller. Cluster controller alerting is handled outside the ZK + * update critical section. + * <p> + * Because ZK events are occur relatively infrequently, any deadlock will occur + * once in a blue moon, which will make it very hard to reproduce. So, extreme + * caution is needed at design time to prevent the problem. + */ + +public class ZKRegistry + implements TaskLifecycleListener, DrillbitStatusListener, Pollable { + /** + * State of each Drillbit that we've discovered through ZK or launched via the + * AM. The tracker is where we combine the ZK information with AM to correlate + * overall Drillbit health. + */ + + protected static class DrillbitTracker { + /** + * A Drillbit can be in one of four states. + */ + + public enum State { + + /** + * An unmanaged Drillbit is one that has announced itself via ZK, but + * which the AM didn't launch (or has not yet received confirmation from + * YARN that it was launched.) In the normal state, this state either does + * not occur (YARN reports the task launch before the Drillbit registers + * in ZK) or is transient (if the Drillbit registers in ZK before YARN + * gets around to telling the AM that the Drillbit was launched.) A + * Drillbit that stays in the unregistered state is likely one launched + * outside the AM: either launched manually or (possibly), one left from a + * previous, failed AM run (though YARN is supposed to kill left-over + * child processes in that case.) + */ + + UNMANAGED, + + /** + * A new Drillbit is one that the AM has launched, but that has not yet + * registered itself with ZK. This is normally a transient state that + * occurs as ZK registration catches up with the YARN launch notification. + * If a Drillbit says in this state, then something is seriously wrong + * (perhaps a mis-configuration). The cluster controller will patiently + * wait a while, then decide bad things are happening and will ask YARN to + * kill the Drillbit, then will retry a time or two, after which it will + * throw up its hands, blacklist the node, and wait for the admin to sort + * things out. + */ + + NEW, + + /** + * Normal operating state: the AM launched the Drillbit, which then + * dutifully registered itself in ZK. Nothing to see here, move along. + */ + + REGISTERED, + + /** + * The Drillbit was working just fine, but its registration has dropped + * out of ZK for a reason best left to the cluster controller to + * determine. Perhaps the controller has decided to kill the Drillbit. + * Perhaps the Drillbit became unresponsive (in which case the controller + * will kill it and retry) or has died (in which case YARN will alert the + * AM that the process exited.) + */ + + DEREGISTERED + } + + /** + * The common key used between tasks and ZK registrations. The key is of the + * form:<br> + * + * <pre> + * host:port:port:port + * </pre> + */ + + protected final String key; + + /** + * ZK tracking state. + * + * @see {@link State} + */ + + protected State state; + + /** + * For Drillbits started by the AM, the task object for this Drillbit. + */ + + protected Task task; + + /** + * For Drillbits discovered through ZK, the Drill endpoint for the Drillbit. + */ + + protected DrillbitEndpoint endpoint; + + public DrillbitTracker(String key, DrillbitEndpoint endpoint) { + this.key = key; + this.state = DrillbitTracker.State.UNMANAGED; + this.endpoint = endpoint; + } + + public DrillbitTracker(String key, Task task) { + this.key = key; + this.task = task; + state = DrillbitTracker.State.NEW; + } + + /** + * Mark that a YARN-managed task has become registered in ZK. This indicates + * that the task has come online. Tell the task to update its state to + * record that the task is, in fact, registered in ZK. This indicates a + * normal, healthy task. + * + * @param tracker + */ + + private void becomeRegistered() { + state = DrillbitTracker.State.REGISTERED; + } + + /** + * Mark that a YARN-managed Drillbit has dropped out of ZK. + * + * @param registryHandler + */ + + public void becomeUnregistered() { + assert state == DrillbitTracker.State.REGISTERED; + state = DrillbitTracker.State.DEREGISTERED; + endpoint = null; + } + } + + public static final String CONTROLLER_PROPERTY = "zk"; + + public static final int UPDATE_PERIOD_MS = 20_000; + + public static final String ENDPOINT_PROPERTY = "endpoint"; + + private static final Log LOG = LogFactory.getLog(ZKRegistry.class); + + /** + * Map of host:port:port:port keys to tracking objects. Identifies the + * Drillbits discovered from ZK, started by the controller, or (ideally) both. + */ + + private Map<String, DrillbitTracker> registry = new HashMap<>(); + + /** + * Interface to Drill's cluster coordinator. + */ + + private ZKClusterCoordinatorDriver zkDriver; + + /** + * Drill's cluster coordinator (or, at least, Drill-on-YARN's version of it. + */ + + private RegistryHandler registryHandler; + + /** + * Last check of ZK status. + */ + + private long lastUpdateTime; + + public ZKRegistry(ZKClusterCoordinatorDriver zkDriver) { + this.zkDriver = zkDriver; + } + + /** + * Called during AM startup to initialize ZK. Checks if any Drillbits are + * already running. These are "unmanaged" because the AM could not have + * started them (since they predate the AM.) + */ + + public void start(RegistryHandler controller) { + this.registryHandler = controller; + try { + zkDriver.build(); + } catch (ZKRuntimeException e) { + LOG.error("Failed to start ZK monitoring", e); + throw new AMWrapperException("Failed to start ZK monitoring", e); + } + for (DrillbitEndpoint dbe : zkDriver.getInitialEndpoints()) { + String key = toKey(dbe); + registry.put(key, new DrillbitTracker(key, dbe)); + + // Blacklist the host for each unmanaged drillbit. + + controller.reserveHost(dbe.getAddress()); + + LOG.warn("Host " + dbe.getAddress() + + " already running a Drillbit outside of YARN."); + } + zkDriver.addDrillbitListener(this); + } + + /** + * Convert a Drillbit endpoint to a string key used in the (zk-->task) map. + * Note that the string format here must match the one used in + * {@link #toKey(Task)} to map a task to string key. + * + * @param dbe + * the Drillbit endpoint from ZK + * @return a string key for this object + */ + + private String toKey(DrillbitEndpoint dbe) { + return ZKClusterCoordinatorDriver.asString(dbe); + } + + /** + * Convert a task to a string key used in the (zk-->task) map. Note that the + * string format here must match the one used in + * {@link #toKey(DrillbitEndpoint)} to map a drillbit endpoint to string key. + * + * @param task + * the task tracked by the cluster controller + * @return a string key for this object + */ + + private String toKey(Task task) { + return zkDriver.toKey(task.getHostName()); + } + + // private String toKey(Container container) { + // return zkDriver.toKey(container.getNodeId().getHost()); + // } + + public static class AckEvent { + Task task; + DrillbitEndpoint endpoint; + + public AckEvent(Task task, DrillbitEndpoint endpoint) { + this.task = task; + this.endpoint = endpoint; + } + } + + /** + * Callback from ZK to indicate that one or more drillbits have become + * registered. We handle registrations in a critical section, then alert the + * cluster controller outside the critical section. + */ + + @Override + public void drillbitRegistered(Set<DrillbitEndpoint> registeredDrillbits) { + List<AckEvent> updates = registerDrillbits(registeredDrillbits); + for (AckEvent event : updates) { + if (event.task == null) { + registryHandler.reserveHost(event.endpoint.getAddress()); + } else { + registryHandler.startAck(event.task, ENDPOINT_PROPERTY, event.endpoint); + } + } + } + + private synchronized List<AckEvent> registerDrillbits( + Set<DrillbitEndpoint> registeredDrillbits) { + List<AckEvent> events = new ArrayList<>(); + for (DrillbitEndpoint dbe : registeredDrillbits) { + AckEvent event = drillbitRegistered(dbe); + if (event != null) { + events.add(event); + } + } + return events; + } + + /** + * Called when a drillbit has become registered. There are two cases. Either + * this is a normal registration of a previously-started task, or this is an + * unmanaged drillbit for which we have no matching task. + */ + + private AckEvent drillbitRegistered(DrillbitEndpoint dbe) { + String key = toKey(dbe); + DrillbitTracker tracker = registry.get(key); + if (tracker == null) { + // Unmanaged drillbit case + + LOG.info("Registration of unmanaged drillbit: " + key); + tracker = new DrillbitTracker(key, dbe); + registry.put(key, tracker); + return new AckEvent(null, dbe); + } + + // Managed drillbit case. Might be we lost, then regained + // ZK connection. + + if (tracker.state == DrillbitTracker.State.REGISTERED) { + LOG.info("Re-registration of known drillbit: " + key); + return null; + } + + // Otherwise, the Drillbit has just registered with ZK. + // Or, if the ZK connection was lost and regained, the + // state changes from DEREGISTERED --> REGISTERED + + LOG.info("Drillbit registered: " + key + ", task: " + tracker.task.toString() ); + tracker.endpoint = dbe; + tracker.becomeRegistered(); + return new AckEvent(tracker.task, dbe); + } + + /** + * Callback from ZK to indicate that one or more drillbits have become + * deregistered from ZK. We handle the deregistrations in a critical section, + * but updates to the cluster controller outside of a critical section. + */ + + @Override + public void drillbitUnregistered( + Set<DrillbitEndpoint> unregisteredDrillbits) { + List<AckEvent> updates = unregisterDrillbits(unregisteredDrillbits); + for (AckEvent event : updates) { + registryHandler.completionAck(event.task, ENDPOINT_PROPERTY); + } + } + + private synchronized List<AckEvent> unregisterDrillbits( + Set<DrillbitEndpoint> unregisteredDrillbits) { + List<AckEvent> events = new ArrayList<>(); + for (DrillbitEndpoint dbe : unregisteredDrillbits) { + AckEvent event = drillbitUnregistered(dbe); + if (event != null) { + events.add(event); + } + } + return events; + } + + /** + * Handle the case that a drillbit becomes unregistered. There are three + * cases. + * <ol> + * <li>The deregistration is for a drillbit that is not in the registry table. + * Indicates a code error.</li> + * <li>The drillbit is unmanaged. This occurs for drillbits started and + * stopped outside of YARN.</li> + * <li>Normal case of deregistration of a YARN-managed drillbit. Inform the + * controller of this event.</li> + * </ol> + * + * @param dbe + */ + + private AckEvent drillbitUnregistered(DrillbitEndpoint dbe) { + String key = toKey(dbe); + DrillbitTracker tracker = registry.get(key); + assert tracker != null; + if (tracker == null) { + // Something is terribly wrong. + // Have seen this when a user kills the Drillbit just after it starts. Evidently, the + // Drillbit registers with ZK just before it is killed, but before DoY hears about + // the registration. + + LOG.error("Internal error - Unexpected drillbit unregistration: " + key); + return null; + } + if (tracker.state == DrillbitTracker.State.UNMANAGED) { + // Unmanaged drillbit + + assert tracker.task == null; + LOG.info("Unmanaged drillbit unregistered: " + key); + registry.remove(key); + registryHandler.releaseHost(dbe.getAddress()); + return null; + } + LOG.info("Drillbit unregistered: " + key + ", task: " + tracker.task.toString() ); + tracker.becomeUnregistered(); + return new AckEvent(tracker.task, dbe); + } + + /** + * Listen for selected YARN task state changes. Called from within the cluster + * controller's critical section. + */ + + @Override + public synchronized void stateChange(Event event, EventContext context) { + switch (event) { + case ALLOCATED: + taskCreated(context.task); + break; + case ENDED: + taskEnded(context.task); + break; + default: + break; + } + } + + /** + * Indicates that the cluster controller has created a task that we expect to + * be monitored by ZK. We handle two cases: the normal case in which we later + * receive a ZK notification. And, the unusual case in which we've already + * received the ZK notification and we now match that notification with this + * task. (The second case could occur if latency causes us to receive the ZK + * notification before we learn from the NM that the task is alive.) + * + * @param task + */ + + private void taskCreated(Task task) { + String key = toKey(task); + DrillbitTracker tracker = registry.get(key); + if (tracker == null) { + // Normal case: no ZK registration yet. + + registry.put(key, new DrillbitTracker(key, task)); + } else if (tracker.state == DrillbitTracker.State.UNMANAGED) { + // Unusual case: ZK registration came first. + + LOG.info("Unmanaged drillbit became managed: " + key); + tracker.task = task; + tracker.becomeRegistered(); + + // Note: safe to call this here as we are already in the controller + // critical section. + + registryHandler.startAck(task, ENDPOINT_PROPERTY, tracker.endpoint); + } else { + LOG.error(task.getLabel() + " - Drillbit registry in wrong state " + + tracker.state + " for new task: " + key); + } + } + + /** + * Report whether the given task is still registered in ZK. Called while + * waiting for a deregistration event to catch possible cases where the + * messages is lost. The message should never be lost, but we've seen + * cases where tasks hang in this state. This is a potential work-around. + * + * @param task + * @return + */ + + public synchronized boolean isRegistered(Task task) { + String key = toKey(task); + DrillbitTracker tracker = registry.get(key); + if (tracker==null) { + return false; + } + return tracker.state == DrillbitTracker.State.REGISTERED; + } + + /** + * Mark that a task (YARN container) has ended. Updates the (zk --> task) + * registry by removing the task. The cluster controller state machine + * monitors ZK and does not end the task until the ZK registration for that + * task drops. As a result, the entry here should be in the deregistered state + * or something is seriously wrong. + * + * @param task + */ + + private void taskEnded(Task task) { + + // If the task has no host name then the task is being cancelled before + // a YARN container was allocated. Just ignore such a case. + + if (task.getHostName() == null) { + return; + } + String key = toKey(task); + DrillbitTracker tracker = registry.get(key); + assert tracker != null; + assert tracker.state == DrillbitTracker.State.DEREGISTERED; + registry.remove(key); + } + + /** + * Periodically check ZK status. If the ZK connection has timed out, something + * is very seriously wrong. Shut the whole Drill cluster down since Drill + * cannot operate without ZooKeeper. + * <p> + * This method should not be synchronized. It checks only the ZK state, not + * internal state. Further, if we do reconnect to ZK, then a ZK thread may + * attempt to update this registry, which will acquire a synchronization lock. + */ + + @Override + public void tick(long curTime) { + if (lastUpdateTime + UPDATE_PERIOD_MS < curTime) { + return; + } + + lastUpdateTime = curTime; + if (zkDriver.hasFailed()) { + int secs = (int) ((zkDriver.getLostConnectionDurationMs() + 500) / 1000); + LOG.error( + "ZooKeeper connection lost, failing after " + secs + " seconds."); + registryHandler.registryDown(); + } + } + + public void finish(RegistryHandler handler) { + zkDriver.removeDrillbitListener(this); + zkDriver.close(); + } + + public synchronized List<String> listUnmanagedDrillits() { + List<String> drillbits = new ArrayList<>(); + for (DrillbitTracker item : registry.values()) { + if (item.state == DrillbitTracker.State.UNMANAGED) { + drillbits.add(item.key); + } + } + return drillbits; + } + + /** + * Get the current registry for testing. Why for testing? Because this is + * unsynchronized. In production code, the map may change out from under you. + * + * @return + */ + + protected Map<String, DrillbitTracker> getRegistryForTesting() { + return registry; + } +} http://git-wip-us.apache.org/repos/asf/drill/blob/f2ac8749/drill-yarn/src/main/java/org/apache/drill/yarn/zk/ZKRuntimeException.java ---------------------------------------------------------------------- diff --git a/drill-yarn/src/main/java/org/apache/drill/yarn/zk/ZKRuntimeException.java b/drill-yarn/src/main/java/org/apache/drill/yarn/zk/ZKRuntimeException.java new file mode 100644 index 0000000..4e1b115 --- /dev/null +++ b/drill-yarn/src/main/java/org/apache/drill/yarn/zk/ZKRuntimeException.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.zk; + +public class ZKRuntimeException extends Exception { + private static final long serialVersionUID = 1L; + + public ZKRuntimeException(String msg, Exception e) { + super(msg, e); + } + + public ZKRuntimeException(String msg) { + super(msg); + } +} http://git-wip-us.apache.org/repos/asf/drill/blob/f2ac8749/drill-yarn/src/main/java/org/apache/drill/yarn/zk/package-info.java ---------------------------------------------------------------------- diff --git a/drill-yarn/src/main/java/org/apache/drill/yarn/zk/package-info.java b/drill-yarn/src/main/java/org/apache/drill/yarn/zk/package-info.java new file mode 100644 index 0000000..14bb427 --- /dev/null +++ b/drill-yarn/src/main/java/org/apache/drill/yarn/zk/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. + */ + +/** + * Interface between the Application Master and ZooKeeper. Classes here manage two + * registrations: Drillbits and the AM itself. + * <p> + * Drillbit registration is used to confirm that Drillbits have indeed come online. + * If Drillbits fail to come online, then the AM concludes that somethign went wrong. + * If Drilbits drop offline unexpectedly, the AM concludes that the Drillbit is sick + * and restarts it. + * <p> + * The AM registry prevents two AMs from attempting to manage the same cluster. + */ + +package org.apache.drill.yarn.zk; \ No newline at end of file http://git-wip-us.apache.org/repos/asf/drill/blob/f2ac8749/drill-yarn/src/main/resources/drill-am/config.ftl ---------------------------------------------------------------------- diff --git a/drill-yarn/src/main/resources/drill-am/config.ftl b/drill-yarn/src/main/resources/drill-am/config.ftl new file mode 100644 index 0000000..8405c1f --- /dev/null +++ b/drill-yarn/src/main/resources/drill-am/config.ftl @@ -0,0 +1,41 @@ +<#-- 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. --> + +<#include "*/generic.ftl"> +<#macro page_head> +</#macro> + +<#macro page_body> + <h4>Fully Resolved Configuration Settings</h4> + <p> + + <table class="table table-hover" style="width: auto;"> + <tr> + <th>Configuration Key</td> + <th>Value</td> + </tr> + <#list model as pair> + <tr> + <td>${pair.getName()}</td> + <td>${pair.getQuotedValue()}</td> + </tr> + </#list> + </table> + <p> + To modify these values: + <ol> + <li>Edit <code>$DRILL_SITE/drill-on-yarn.conf</code> (for the drill.yarn settings),</li> + <li>Edit <code>$DRILL_SITE/drill-override.conf</code> (for the drill.exec settings).</li> + <li>Restart your Drill cluster using the Drill-on-YARN client.</li> + </ol> +</#macro> + +<@page_html/> http://git-wip-us.apache.org/repos/asf/drill/blob/f2ac8749/drill-yarn/src/main/resources/drill-am/confirm.ftl ---------------------------------------------------------------------- diff --git a/drill-yarn/src/main/resources/drill-am/confirm.ftl b/drill-yarn/src/main/resources/drill-am/confirm.ftl new file mode 100644 index 0000000..515293d --- /dev/null +++ b/drill-yarn/src/main/resources/drill-am/confirm.ftl @@ -0,0 +1,70 @@ +<#-- 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. --> + +<#include "*/generic.ftl"> +<#macro page_head> +</#macro> + +<#macro page_body> + <h4><#if model.getType( ) == "STOPPED"> + Stop Drill Cluster + <#else> + Resize Drill Cluster + </#if></h4> + <p> + + <#if model.getType( ) == "RESIZED"> + <div class="alert alert-success"> + <strong>Success!</strong> Cluster resizing to ${model.getValue( )} nodes. + </div> + <#elseif model.getType( ) == "CANCELLED"> + <div class="alert alert-info"> + <strong>Success!</strong> Drillbit ${model.getValue( )} was cancelled. + </div> + <#elseif model.getType( ) == "NULL_RESIZE"> + <div class="alert alert-info"> + <strong>Note:</strong> The new size of ${model.getValue( )} is the + same as the current cluster size. + </div> + <#elseif model.getType( ) == "INVALID_RESIZE"> + <div class="alert alert-danger"> + <strong>Error!</strong> Invalid cluster resize level: ${model.getValue( )}. + Please <a href="/manage">try again</a>. + </div> + <#elseif model.getType( ) == "INVALID_GROW"> + <div class="alert alert-danger"> + <strong>Error!</strong> Invalid cluster grow amount: ${model.getValue( )}. + Please <a href="/manage">try again</a>. + </div> + <#elseif model.getType( ) == "INVALID_SHRINK"> + <div class="alert alert-danger"> + <strong>Error!</strong> Invalid cluster shrink amount: ${model.getValue( )}. + Please <a href="/manage">try again</a>. + </div> + <#elseif model.getType( ) == "INVALID_TASK"> + <div class="alert alert-danger"> + <strong>Error!</strong> Invalid Drillbit ID: ${model.getValue( )}. + Perhaps the Drillbit has already stopped. + </div> + <#elseif model.getType( ) == "STOPPED"> + <div class="alert alert alert-success"> + <strong>Success!</strong> Cluster is shutting down. + </div> + Pages on this site will be unavailable until the cluster restarts. + </#if> + <#if model.getType( ) == "CANCELLED"> + Return to the <a href="/drillbits">Drillbits page</a>. + <#elseif model.getType( ) != "STOPPED"> + Return to the <a href="/manage">Management page</a>. + </#if> +</#macro> + +<@page_html/> http://git-wip-us.apache.org/repos/asf/drill/blob/f2ac8749/drill-yarn/src/main/resources/drill-am/generic.ftl ---------------------------------------------------------------------- diff --git a/drill-yarn/src/main/resources/drill-am/generic.ftl b/drill-yarn/src/main/resources/drill-am/generic.ftl new file mode 100644 index 0000000..b76a917 --- /dev/null +++ b/drill-yarn/src/main/resources/drill-am/generic.ftl @@ -0,0 +1,78 @@ +<#-- 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. --> + +<#-- Adapted from the Drill generic.ftl, adjusted for use in the AM. --> + +<#macro page_head> +</#macro> + +<#macro page_body> +</#macro> + +<#macro page_html> + <meta http-equiv="X-UA-Compatible" content="IE=edge"> + + <title>Apache Drill - Application Master</title> + <link rel="shortcut icon" href="/static/img/drill.ico"> + + <link href="/static/css/bootstrap.min.css" rel="stylesheet"> + <link href="/drill-am/static/css/drill-am.css" rel="stylesheet"> + + <script src="/static/js/jquery.min.js"></script> + <script src="/static/js/bootstrap.min.js"></script> + + <!-- HTML5 shim and Respond.js IE8 support of HTML5 elements and media queries --> + <!--[if lt IE 9]> + <script src="/static/js/html5shiv.js"></script> + <script src="/static/js/1.4.2/respond.min.js"></script> + <![endif]--> + + <@page_head/> + </head> + <body role="document"> + <div class="navbar navbar-inverse navbar-fixed-top" role="navigation"> + <div class="container-fluid"> + <div class="navbar-header"> + <button type="button" class="navbar-toggle" data-toggle="collapse" data-target=".navbar-collapse"> + <span class="sr-only">Toggle navigation</span> + <span class="icon-bar"></span> + <span class="icon-bar"></span> + <span class="icon-bar"></span> + </button> + <a class="navbar-brand" href="/">Apache Drill</a> + </div> + <div class="navbar-collapse collapse"> + <ul class="nav navbar-nav"> + <li><a href="/config">Configuration</a></li> + <li><a href="/drillbits">Drillbits</a></li> + <li><a href="/manage">Manage</a></li> + <li><a href="/history">History</a></li> + </ul> + <ul class="nav navbar-nav navbar-right"> + <li><a href="${docsLink}">Documentation</a> + <#if showLogin == true > + <li><a href="/login">Log In</a> + </#if> + <#if showLogout == true > + <li><a href="/logout">Log Out (${loggedInUserName})</a> + </#if> + </ul> + </div> + </div> + </div> + + <div class="container-fluid drill-am" role="main"> + <h3>YARN Application Master – ${clusterName}</h3> + <@page_body/> + </div> + </body> + </html> +</#macro> http://git-wip-us.apache.org/repos/asf/drill/blob/f2ac8749/drill-yarn/src/main/resources/drill-am/history.ftl ---------------------------------------------------------------------- diff --git a/drill-yarn/src/main/resources/drill-am/history.ftl b/drill-yarn/src/main/resources/drill-am/history.ftl new file mode 100644 index 0000000..c588d06 --- /dev/null +++ b/drill-yarn/src/main/resources/drill-am/history.ftl @@ -0,0 +1,59 @@ +<#-- 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. --> + +<#include "*/generic.ftl"> +<#macro page_head> + <meta http-equiv="refresh" content="${refreshSecs}" > +</#macro> + +<#macro page_body> + <h4>Drillbit History</h4> + <p> + + <div class="table-responsive"> + <table class="table table-hover"> + <tr> + <th>ID</th> + <th>Try</th> + <th>Pool</th> + <th>Host</th> + <th>Container</th> + <th>Memory (MB)</th> + <th>Virtual Cores</th> + <th>Start Time</th> + <th>End Time</th> + <th>Disposition</th> + </th> + <#assign count=0> + <#list model as task> + <#assign count=count+1> + <tr> + <td><b>${task.getTaskId( )}</b></td> + <td>${task.getTryCount( )}</td> + <td>${task.getGroupName( )}</td> + <td><#if task.hasContainer( )><a href="${task.getNmLink( )}">${task.getHost( )}</a> + <#else> </#if></td> + <td>${task.getContainerId()}</td> + <td>${task.getMemory( )}</td> + <td>${task.getVcores( )}</td> + <td>${task.getStartTime( )}</td> + <td>${task.getEndTime( )}</td> + <td>${task.getDisposition( )}</td> + </tr> + </#list> + </table> + <#if count == 0> + No drillbits have completed. + </#if> + </div> +</#macro> + +<@page_html/> http://git-wip-us.apache.org/repos/asf/drill/blob/f2ac8749/drill-yarn/src/main/resources/drill-am/index.ftl ---------------------------------------------------------------------- diff --git a/drill-yarn/src/main/resources/drill-am/index.ftl b/drill-yarn/src/main/resources/drill-am/index.ftl new file mode 100644 index 0000000..18d6ab5 --- /dev/null +++ b/drill-yarn/src/main/resources/drill-am/index.ftl @@ -0,0 +1,128 @@ +<#-- 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. --> + +<#include "*/generic.ftl"> +<#macro page_head> + <meta http-equiv="refresh" content="${model.getRefreshSecs( )}" > +</#macro> + +<#macro page_body> + <h4>Drill Cluster Status</h4> + + <table class="table table-hover" style="width: auto;"> + <tr> + <td>YARN Application ID:</td> + <td><a href="${model.getRmAppLink( )}" data-toggle="tooltip" title="YARN Resource Manager page for this application"> + ${model.getAppId( )}</a></td> + </tr> + <tr> + <td>YARN Resource Manager:</td> + <td><#if model.getRmLink()?? > <#-- Occurs early in startup before app is fully registered. --> + <a href="${model.getRmLink( )}" data-toggle="tooltip" title="YARN Resource Manager page for this container"> + ${model.getRmHost( )}</a> + <#else>Unavailable + </#if></td> + </tr> + <tr> + <td>YARN Node Manager for AM:</td> + <td><#if model.getNmLink()?? > <#-- Occurs early in startup before app is fully registered. --> + <a href="${model.getNmLink( )}" data-toggle="tooltip" title="YARN Node Manager"> + ${model.getNmHost( )}</a> | + <a href="${model.getNmAppLink( )}" data-toggle="tooltip" title="YARN Node Manager page for this application">App info</a> + <#else>Unavailable + </#if></td> + </tr> + <tr> + <td>ZooKeeper Hosts:</td> + <td><span data-toggle="tooltip" title="ZooKeeper connection string."> + ${model.getZkConnectionStr( )}</span></td> + </tr> + <tr> + <td>ZooKeeper Root:</td> + <td><span data-toggle="tooltip" title="ZooKeeper Drill root."> + ${model.getZkRoot( )}</span></td> + </tr> + <tr> + <td>ZooKeeper Cluster ID:</td> + <td><span data-toggle="tooltip" title="ZooKeeper Drill cluster-id."> + ${model.getZkClusterId( )}</span></td> + </tr> + <tr> + <td>State:</td> + <td><span data-toggle="tooltip" title="${model.getStateHint( )}"> + ${model.getState( )}</span></td> + </tr> + <tr> + <td>Target Drillbit Count:</td> + <td>${model.getTargetCount( )}</td> + </tr> + <tr> + <td>Live Drillbit Count:</td> + <td>${model.getLiveCount( )}</td> + </tr> + <#if model.getUnmanagedCount( ) gt 0 > + <tr> + <td style="color: red">Unmanaged Drillbit Count:</td> + <td>${model.getUnmanagedCount( )}</td> + </tr> + </#if> + <#if model.getBlacklistCount( ) gt 0 > + <tr> + <td style="color: red">Blacklisted Node Count:</td> + <td>${model.getBlacklistCount( )}</td> + </tr> + </#if> + <tr> + <td>Total Drill Virtual Cores:</td> + <td>${model.getDrillTotalVcores( )}</td> + </tr> + <tr> + <td>Total Drill Memory (MB):</td> + <td>${model.getDrillTotalMemory( )}</td> + </tr> + <#if model.supportsDiskResource( ) > + <tr> + <td>Total Drill Disks:</td> + <td>${model.getDrillTotalDisks( )}</td> + </tr> + </#if> + </table> + <table class="table table-hover" style="width: auto;"> + <tr> + <th>Group</th> + <th>Type</th> + <th>Target Drillbit Count</th> + <th>Total Drillbits</th> + <th>Live Drillbits</th> + <th>Memory per Drillbit (MB)</th> + <th>VCores per Drillbit</th> + <#if model.supportsDiskResource( ) > + <th>Disks per Drillbit</th> + </#if> + </tr> + <#list model.getGroups( ) as group> + <tr> + <td>${group.getName( )}</td> + <td>${group.getType( )}</td> + <td>${group.getTargetCount( )}</td> + <td>${group.getTaskCount( )}</td> + <td>${group.getLiveCount( )}</td> + <td>${group.getMemory( )}</td> + <td>${group.getVcores( )}</td> + <#if model.supportsDiskResource( ) > + <td>${group.getDisks( )}</td> + </#if> + </tr> + </#list> + </table> +</#macro> + +<@page_html/> http://git-wip-us.apache.org/repos/asf/drill/blob/f2ac8749/drill-yarn/src/main/resources/drill-am/login.ftl ---------------------------------------------------------------------- diff --git a/drill-yarn/src/main/resources/drill-am/login.ftl b/drill-yarn/src/main/resources/drill-am/login.ftl new file mode 100644 index 0000000..036229e --- /dev/null +++ b/drill-yarn/src/main/resources/drill-am/login.ftl @@ -0,0 +1,35 @@ +<#-- 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. --> + +<#include "*/generic.ftl"> +<#macro page_head> +</#macro> + +<#macro page_body> + <div align="center" class="table-responsive"> + <form role="form" name="input" action="/j_security_check" method="POST"> + <fieldset> + <div class="form-group"> + <img src="/drill-am/static/img/apache-drill-logo.png" alt="Apache Drill Logo"> + <#if model??> + <div class="alert alert-danger"> + <strong>Error</strong> ${model} + </div> + </#if> + <p><input type="text" size="30" name="j_username" placeholder="Username"></p> + <p><input type="password" size="30" name="j_password" placeholder="Password"></p> + <p><button type="submit" class="btn btn-primary">Log In</button></p> + </div> + </fieldset> + </form> + </div> +</#macro> +<@page_html/> \ No newline at end of file
