http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/51c2b92c/slider-core/src/main/java/org/apache/slider/server/appmaster/state/ProviderAppState.java ---------------------------------------------------------------------- diff --git a/slider-core/src/main/java/org/apache/slider/server/appmaster/state/ProviderAppState.java b/slider-core/src/main/java/org/apache/slider/server/appmaster/state/ProviderAppState.java deleted file mode 100644 index 37e9a7f..0000000 --- a/slider-core/src/main/java/org/apache/slider/server/appmaster/state/ProviderAppState.java +++ /dev/null @@ -1,307 +0,0 @@ -/* - * 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.slider.server.appmaster.state; - -import org.apache.hadoop.yarn.api.records.Container; -import org.apache.hadoop.yarn.api.records.ContainerId; -import org.apache.hadoop.yarn.exceptions.YarnRuntimeException; -import org.apache.slider.api.ClusterDescription; -import org.apache.slider.api.ClusterNode; -import org.apache.slider.api.types.ApplicationLivenessInformation; -import org.apache.slider.api.types.ComponentInformation; -import org.apache.slider.api.types.NodeInformation; -import org.apache.slider.api.types.RoleStatistics; -import org.apache.slider.core.conf.AggregateConf; -import org.apache.slider.core.conf.ConfTreeOperations; -import org.apache.slider.core.exceptions.NoSuchNodeException; -import org.apache.slider.core.registry.docstore.PublishedConfigSet; -import org.apache.slider.core.registry.docstore.PublishedExportsSet; -import org.apache.slider.server.appmaster.web.rest.RestPaths; -import org.apache.slider.server.services.utility.PatternValidator; - -import java.util.ArrayList; -import java.util.Collection; -import java.util.List; -import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; - -/** - * Implementation of {@link StateAccessForProviders}, which means - * state access for providers, web UI and IPC/REST views. - */ -public class ProviderAppState implements StateAccessForProviders { - - - private final Map<String, PublishedConfigSet> publishedConfigSets = - new ConcurrentHashMap<>(5); - private final PublishedExportsSet publishedExportsSets = new PublishedExportsSet(); - private static final PatternValidator validator = new PatternValidator( - RestPaths.PUBLISHED_CONFIGURATION_SET_REGEXP); - private String applicationName; - - private final AppState appState; - - public ProviderAppState(String applicationName, AppState appState) { - this.appState = appState; - this.applicationName = applicationName; - } - - public void setApplicationName(String applicationName) { - this.applicationName = applicationName; - } - - @Override - public String getApplicationName() { - return applicationName; - } - - @Override - public PublishedConfigSet getPublishedSliderConfigurations() { - return getOrCreatePublishedConfigSet(RestPaths.SLIDER_CONFIGSET); - } - - @Override - public PublishedExportsSet getPublishedExportsSet() { - return publishedExportsSets; - } - - @Override - public PublishedConfigSet getPublishedConfigSet(String name) { - return publishedConfigSets.get(name); - } - - @Override - public PublishedConfigSet getOrCreatePublishedConfigSet(String name) { - PublishedConfigSet set = publishedConfigSets.get(name); - if (set == null) { - validator.validate(name); - synchronized (publishedConfigSets) { - // synchronized double check to ensure that there is never an overridden - // config set created - set = publishedConfigSets.get(name); - if (set == null) { - set = new PublishedConfigSet(); - publishedConfigSets.put(name, set); - } - } - } - return set; - } - - @Override - public List<String> listConfigSets() { - - synchronized (publishedConfigSets) { - List<String> sets = new ArrayList<>(publishedConfigSets.keySet()); - return sets; - } - } - - @Override - public Map<Integer, RoleStatus> getRoleStatusMap() { - return appState.getRoleStatusMap(); - } - - - @Override - public Map<ContainerId, RoleInstance> getFailedContainers() { - return appState.getFailedContainers(); - } - - @Override - public Map<ContainerId, RoleInstance> getLiveContainers() { - return appState.getLiveContainers(); - } - - @Override - public ClusterDescription getClusterStatus() { - return appState.getClusterStatus(); - } - - @Override - public ConfTreeOperations getResourcesSnapshot() { - return appState.getResourcesSnapshot(); - } - - @Override - public ConfTreeOperations getAppConfSnapshot() { - return appState.getAppConfSnapshot(); - } - - @Override - public ConfTreeOperations getInternalsSnapshot() { - return appState.getInternalsSnapshot(); - } - - @Override - public boolean isApplicationLive() { - return appState.isApplicationLive(); - } - - @Override - public long getSnapshotTime() { - return appState.getSnapshotTime(); - } - - @Override - public AggregateConf getInstanceDefinitionSnapshot() { - return appState.getInstanceDefinitionSnapshot(); - } - - @Override - public AggregateConf getUnresolvedInstanceDefinition() { - return appState.getUnresolvedInstanceDefinition(); - } - - @Override - public RoleStatus lookupRoleStatus(int key) { - return appState.lookupRoleStatus(key); - } - - @Override - public RoleStatus lookupRoleStatus(Container c) throws YarnRuntimeException { - return appState.lookupRoleStatus(c); - } - - @Override - public RoleStatus lookupRoleStatus(String name) throws YarnRuntimeException { - return appState.lookupRoleStatus(name); - } - - @Override - public List<RoleInstance> cloneOwnedContainerList() { - return appState.cloneOwnedContainerList(); - } - - @Override - public int getNumOwnedContainers() { - return appState.getNumOwnedContainers(); - } - - @Override - public RoleInstance getOwnedContainer(ContainerId id) { - return appState.getOwnedContainer(id); - } - - @Override - public RoleInstance getOwnedContainer(String id) throws NoSuchNodeException { - return appState.getOwnedInstanceByContainerID(id); - } - - @Override - public List<RoleInstance> cloneLiveContainerInfoList() { - return appState.cloneLiveContainerInfoList(); - } - - @Override - public RoleInstance getLiveInstanceByContainerID(String containerId) throws - NoSuchNodeException { - return appState.getLiveInstanceByContainerID(containerId); - } - - @Override - public List<RoleInstance> getLiveInstancesByContainerIDs(Collection<String> containerIDs) { - return appState.getLiveInstancesByContainerIDs(containerIDs); - } - - @Override - public ClusterDescription refreshClusterStatus() { - return appState.refreshClusterStatus(); - } - - @Override - public List<RoleStatus> cloneRoleStatusList() { - return appState.cloneRoleStatusList(); - } - - @Override - public ApplicationLivenessInformation getApplicationLivenessInformation() { - return appState.getApplicationLivenessInformation(); - } - - @Override - public Map<String, Integer> getLiveStatistics() { - return appState.getLiveStatistics(); - } - - @Override - public Map<String, ComponentInformation> getComponentInfoSnapshot() { - return appState.getComponentInfoSnapshot(); - } - - @Override - public Map<String, Map<String, ClusterNode>> getRoleClusterNodeMapping() { - return appState.createRoleToClusterNodeMap(); - } - - @Override - public List<RoleInstance> enumLiveInstancesInRole(String role) { - List<RoleInstance> nodes = new ArrayList<>(); - Collection<RoleInstance> allRoleInstances = cloneLiveContainerInfoList(); - for (RoleInstance node : allRoleInstances) { - if (role.isEmpty() || role.equals(node.role)) { - nodes.add(node); - } - } - return nodes; - } - - @Override - public List<RoleInstance> lookupRoleContainers(String component) { - RoleStatus roleStatus = lookupRoleStatus(component); - List<RoleInstance> ownedContainerList = cloneOwnedContainerList(); - List<RoleInstance> matching = new ArrayList<>(ownedContainerList.size()); - int roleId = roleStatus.getPriority(); - for (RoleInstance instance : ownedContainerList) { - if (instance.roleId == roleId) { - matching.add(instance); - } - } - return matching; - } - - @Override - public ComponentInformation getComponentInformation(String component) { - RoleStatus roleStatus = lookupRoleStatus(component); - ComponentInformation info = roleStatus.serialize(); - List<RoleInstance> containers = lookupRoleContainers(component); - info.containers = new ArrayList<>(containers.size()); - for (RoleInstance container : containers) { - info.containers.add(container.id); - } - return info; - } - - @Override - public Map<String, NodeInformation> getNodeInformationSnapshot() { - return appState.getRoleHistory() - .getNodeInformationSnapshot(appState.buildNamingMap()); - } - - @Override - public NodeInformation getNodeInformation(String hostname) { - return appState.getRoleHistory() - .getNodeInformation(hostname, appState.buildNamingMap()); - } - - @Override - public RoleStatistics getRoleStatistics() { - return appState.getRoleStatistics(); - } -}
http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/51c2b92c/slider-core/src/main/java/org/apache/slider/server/appmaster/state/RoleHistory.java ---------------------------------------------------------------------- diff --git a/slider-core/src/main/java/org/apache/slider/server/appmaster/state/RoleHistory.java b/slider-core/src/main/java/org/apache/slider/server/appmaster/state/RoleHistory.java deleted file mode 100644 index 4e8a4d7..0000000 --- a/slider-core/src/main/java/org/apache/slider/server/appmaster/state/RoleHistory.java +++ /dev/null @@ -1,1101 +0,0 @@ -/* - * 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.slider.server.appmaster.state; - -import com.google.common.annotations.VisibleForTesting; -import com.google.common.base.Preconditions; -import org.apache.hadoop.fs.FileSystem; -import org.apache.hadoop.fs.Path; -import org.apache.hadoop.yarn.api.records.Container; -import org.apache.hadoop.yarn.api.records.NodeReport; -import org.apache.hadoop.yarn.api.records.NodeState; -import org.apache.hadoop.yarn.api.records.Resource; -import org.apache.slider.api.types.NodeInformation; -import org.apache.slider.common.tools.SliderUtils; -import org.apache.slider.core.exceptions.BadConfigException; -import org.apache.slider.providers.ProviderRole; -import org.apache.slider.server.appmaster.management.BoolMetric; -import org.apache.slider.server.appmaster.management.MetricsAndMonitoring; -import org.apache.slider.server.appmaster.management.Timestamp; -import org.apache.slider.server.appmaster.operations.AbstractRMOperation; -import org.apache.slider.server.avro.LoadedRoleHistory; -import org.apache.slider.server.avro.NodeEntryRecord; -import org.apache.slider.server.avro.RoleHistoryHeader; -import org.apache.slider.server.avro.RoleHistoryWriter; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.io.IOException; -import java.util.ArrayList; -import java.util.Collection; -import java.util.Collections; -import java.util.HashMap; -import java.util.LinkedList; -import java.util.List; -import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; - -/** - * The Role History. - * <p> - * Synchronization policy: all public operations are synchronized. - * Protected methods are in place for testing -no guarantees are made. - * <p> - * Inner classes have no synchronization guarantees; they should be manipulated - * in these classes and not externally. - * <p> - * Note that as well as some methods marked visible for testing, there - * is the option for the time generator method, {@link #now()} to - * be overridden so that a repeatable time series can be used. - * - */ -public class RoleHistory { - protected static final Logger log = - LoggerFactory.getLogger(RoleHistory.class); - private final List<ProviderRole> providerRoles; - /** the roles in here are shared with App State */ - private final Map<Integer, RoleStatus> roleStatusMap = new HashMap<>(); - private final AbstractClusterServices recordFactory; - - private long startTime; - - /** Time when saved */ - private final Timestamp saveTime = new Timestamp(0); - - /** If the history was loaded, the time at which the history was saved. - * That is: the time the data was valid */ - private final Timestamp thawedDataTime = new Timestamp(0); - - private NodeMap nodemap; - private int roleSize; - private final BoolMetric dirty = new BoolMetric(false); - private FileSystem filesystem; - private Path historyPath; - private RoleHistoryWriter historyWriter = new RoleHistoryWriter(); - - /** - * When were the nodes updated in a {@link #onNodesUpdated(List)} call? - * If zero: never. - */ - private final Timestamp nodesUpdatedTime = new Timestamp(0); - private final BoolMetric nodeUpdateReceived = new BoolMetric(false); - - private OutstandingRequestTracker outstandingRequests = - new OutstandingRequestTracker(); - - /** - * For each role, lists nodes that are available for data-local allocation, - * ordered by more recently released - to accelerate node selection. - * That is, they are "recently used nodes" - */ - private Map<Integer, LinkedList<NodeInstance>> recentNodes; - - /** - * Instantiate - * @param roles initial role list - * @param recordFactory yarn record factory - * @throws BadConfigException - */ - public RoleHistory(Collection<RoleStatus> roles, AbstractClusterServices recordFactory) throws BadConfigException { - this.recordFactory = recordFactory; - roleSize = roles.size(); - providerRoles = new ArrayList<>(roleSize); - for (RoleStatus role : roles) { - addNewRole(role); - } - reset(); - } - - /** - * Reset the variables -this does not adjust the fixed attributes - * of the history, but the nodemap and failed node map are cleared. - */ - protected synchronized void reset() throws BadConfigException { - - nodemap = new NodeMap(roleSize); - resetAvailableNodeLists(); - outstandingRequests = new OutstandingRequestTracker(); - } - - /** - * Register all metrics with the metrics infra - * @param metrics metrics - */ - public void register(MetricsAndMonitoring metrics) { - metrics.register(RoleHistory.class, dirty, "dirty"); - metrics.register(RoleHistory.class, nodesUpdatedTime, "nodes-updated.time"); - metrics.register(RoleHistory.class, nodeUpdateReceived, "nodes-updated.flag"); - metrics.register(RoleHistory.class, thawedDataTime, "thawed.time"); - metrics.register(RoleHistory.class, saveTime, "saved.time"); - } - - /** - * safety check: make sure the role is unique amongst - * the role stats...which is extended with the new role - * @param roleStatus role - * @throws ArrayIndexOutOfBoundsException - * @throws BadConfigException - */ - protected void putRole(RoleStatus roleStatus) throws BadConfigException { - int index = roleStatus.getKey(); - if (index < 0) { - throw new BadConfigException("Provider " + roleStatus + " id is out of range"); - } - if (roleStatusMap.get(index) != null) { - throw new BadConfigException( - roleStatus.toString() + " id duplicates that of " + - roleStatusMap.get(index)); - } - roleStatusMap.put(index, roleStatus); - } - - /** - * Add a new role - * @param roleStatus new role - */ - public void addNewRole(RoleStatus roleStatus) throws BadConfigException { - log.debug("Validating/adding new role to role history: {} ", roleStatus); - putRole(roleStatus); - this.providerRoles.add(roleStatus.getProviderRole()); - } - - /** - * Lookup a role by ID - * @param roleId role Id - * @return role or null if not found - */ - public ProviderRole lookupRole(int roleId) { - for (ProviderRole role : providerRoles) { - if (role.id == roleId) { - return role; - } - } - return null; - } - - /** - * Clear the lists of available nodes - */ - private synchronized void resetAvailableNodeLists() { - recentNodes = new ConcurrentHashMap<>(roleSize); - } - - /** - * Prepare the history for re-reading its state. - * <p> - * This intended for use by the RoleWriter logic. - * @throws BadConfigException if there is a problem rebuilding the state - */ - private void prepareForReading(RoleHistoryHeader header) - throws BadConfigException { - reset(); - - int roleCountInSource = header.getRoles(); - if (roleCountInSource != roleSize) { - log.warn("Number of roles in source {}" - +" does not match the expected number of {}", - roleCountInSource, - roleSize); - } - //record when the data was loaded - setThawedDataTime(header.getSaved()); - } - - /** - * rebuild the placement history from the loaded role history - * @param loadedRoleHistory loaded history - * @return the number of entries discarded - * @throws BadConfigException if there is a problem rebuilding the state - */ - @VisibleForTesting - public synchronized int rebuild(LoadedRoleHistory loadedRoleHistory) throws BadConfigException { - RoleHistoryHeader header = loadedRoleHistory.getHeader(); - prepareForReading(header); - int discarded = 0; - Long saved = header.getSaved(); - for (NodeEntryRecord nodeEntryRecord : loadedRoleHistory.records) { - Integer roleId = nodeEntryRecord.getRole(); - NodeEntry nodeEntry = new NodeEntry(roleId); - nodeEntry.setLastUsed(nodeEntryRecord.getLastUsed()); - if (nodeEntryRecord.getActive()) { - //if active at the time of save, make the last used time the save time - nodeEntry.setLastUsed(saved); - } - String hostname = SliderUtils.sequenceToString(nodeEntryRecord.getHost()); - ProviderRole providerRole = lookupRole(roleId); - if (providerRole == null) { - // discarding entry - log.info("Discarding history entry with unknown role: {} on host {}", - roleId, hostname); - discarded ++; - } else { - NodeInstance instance = getOrCreateNodeInstance(hostname); - instance.set(roleId, nodeEntry); - } - } - return discarded; - } - - public synchronized long getStartTime() { - return startTime; - } - - public synchronized long getSaveTime() { - return saveTime.get(); - } - - public long getThawedDataTime() { - return thawedDataTime.get(); - } - - public void setThawedDataTime(long thawedDataTime) { - this.thawedDataTime.set(thawedDataTime); - } - - public synchronized int getRoleSize() { - return roleSize; - } - - /** - * Get the total size of the cluster -the number of NodeInstances - * @return a count - */ - public synchronized int getClusterSize() { - return nodemap.size(); - } - - public synchronized boolean isDirty() { - return dirty.get(); - } - - public synchronized void setDirty(boolean dirty) { - this.dirty.set(dirty); - } - - /** - * Tell the history that it has been saved; marks itself as clean - * @param timestamp timestamp -updates the savetime field - */ - public synchronized void saved(long timestamp) { - setDirty(false); - saveTime.set(timestamp); - } - - /** - * Get a clone of the nodemap. - * The instances inside are not cloned - * @return the map - */ - public synchronized NodeMap cloneNodemap() { - return (NodeMap) nodemap.clone(); - } - - /** - * Get snapshot of the node map - * @return a snapshot of the current node state - * @param naming naming map of priority to enty name; entries must be unique. - * It's OK to be incomplete, for those the list falls back to numbers. - */ - public synchronized Map<String, NodeInformation> getNodeInformationSnapshot( - Map<Integer, String> naming) { - Map<String, NodeInformation> result = new HashMap<>(nodemap.size()); - for (Map.Entry<String, NodeInstance> entry : nodemap.entrySet()) { - result.put(entry.getKey(), entry.getValue().serialize(naming)); - } - return result; - } - - /** - * Get the information on a node - * @param hostname hostname - * @param naming naming map of priority to enty name; entries must be unique. - * It's OK to be incomplete, for those the list falls back to numbers. - * @return the information about that host, or null if there is none - */ - public synchronized NodeInformation getNodeInformation(String hostname, - Map<Integer, String> naming) { - NodeInstance nodeInstance = nodemap.get(hostname); - return nodeInstance != null ? nodeInstance.serialize(naming) : null; - } - - /** - * Get the node instance for the specific node -creating it if needed - * @param hostname node address - * @return the instance - */ - public synchronized NodeInstance getOrCreateNodeInstance(String hostname) { - //convert to a string - return nodemap.getOrCreate(hostname); - } - - /** - * Insert a list of nodes into the map; overwrite any with that name. - * This is a bulk operation for testing. - * Important: this does not update the available node lists, these - * must be rebuilt afterwards. - * @param nodes collection of nodes. - */ - @VisibleForTesting - public synchronized void insert(Collection<NodeInstance> nodes) { - nodemap.insert(nodes); - } - - /** - * Get current time. overrideable for test subclasses - * @return current time in millis - */ - protected long now() { - return System.currentTimeMillis(); - } - - /** - * Mark ourselves as dirty - */ - public void touch() { - setDirty(true); - try { - saveHistoryIfDirty(); - } catch (IOException e) { - log.warn("Failed to save history file ", e); - } - } - - /** - * reset the failed recently counters - */ - public synchronized void resetFailedRecently() { - log.info("Resetting failure history"); - nodemap.resetFailedRecently(); - } - - /** - * Get the path used for history files - * @return the directory used for history files - */ - public Path getHistoryPath() { - return historyPath; - } - - /** - * Save the history to its location using the timestamp as part of - * the filename. The saveTime and dirty fields are updated - * @param time timestamp timestamp to use as the save time - * @return the path saved to - * @throws IOException IO problems - */ - @VisibleForTesting - public synchronized Path saveHistory(long time) throws IOException { - Path filename = historyWriter.createHistoryFilename(historyPath, time); - historyWriter.write(filesystem, filename, true, this, time); - saved(time); - return filename; - } - - /** - * Save the history with the current timestamp if it is dirty; - * return the path saved to if this is the case - * @return the path or null if the history was not saved - * @throws IOException failed to save for some reason - */ - public synchronized Path saveHistoryIfDirty() throws IOException { - if (isDirty()) { - return saveHistory(now()); - } else { - return null; - } - } - - /** - * Start up - * @param fs filesystem - * @param historyDir path in FS for history - * @return true if the history was thawed - */ - public boolean onStart(FileSystem fs, Path historyDir) throws BadConfigException { - assert filesystem == null; - filesystem = fs; - historyPath = historyDir; - startTime = now(); - //assume the history is being thawed; this will downgrade as appropriate - return onThaw(); - } - - /** - * Handler for bootstrap event: there was no history to thaw - */ - public void onBootstrap() { - log.debug("Role history bootstrapped"); - } - - /** - * Handle the start process <i>after the history has been rebuilt</i>, - * and after any gc/purge - */ - public synchronized boolean onThaw() throws BadConfigException { - assert filesystem != null; - assert historyPath != null; - boolean thawSuccessful = false; - //load in files from data dir - - LoadedRoleHistory loadedRoleHistory = null; - try { - loadedRoleHistory = historyWriter.loadFromHistoryDir(filesystem, historyPath); - } catch (IOException e) { - log.warn("Exception trying to load history from {}", historyPath, e); - } - if (loadedRoleHistory != null) { - rebuild(loadedRoleHistory); - thawSuccessful = true; - Path loadPath = loadedRoleHistory.getPath(); - log.debug("loaded history from {}", loadPath); - // delete any old entries - try { - int count = historyWriter.purgeOlderHistoryEntries(filesystem, loadPath); - log.debug("Deleted {} old history entries", count); - } catch (IOException e) { - log.info("Ignoring exception raised while trying to delete old entries", - e); - } - - //start is then completed - buildRecentNodeLists(); - } else { - //fallback to bootstrap procedure - onBootstrap(); - } - return thawSuccessful; - } - - - /** - * (After the start), rebuild the availability data structures - */ - @VisibleForTesting - public synchronized void buildRecentNodeLists() { - resetAvailableNodeLists(); - // build the list of available nodes - for (Map.Entry<String, NodeInstance> entry : nodemap.entrySet()) { - NodeInstance ni = entry.getValue(); - for (int i = 0; i < roleSize; i++) { - NodeEntry nodeEntry = ni.get(i); - if (nodeEntry != null && nodeEntry.isAvailable()) { - log.debug("Adding {} for role {}", ni, i); - listRecentNodesForRoleId(i).add(ni); - } - } - } - // sort the resulting arrays - for (int i = 0; i < roleSize; i++) { - sortRecentNodeList(i); - } - } - - /** - * Get the nodes for an ID -may be null - * @param id role ID - * @return potentially null list - */ - @VisibleForTesting - public List<NodeInstance> getRecentNodesForRoleId(int id) { - return recentNodes.get(id); - } - - /** - * Get a possibly empty list of suggested nodes for a role. - * @param id role ID - * @return list - */ - private LinkedList<NodeInstance> listRecentNodesForRoleId(int id) { - LinkedList<NodeInstance> instances = recentNodes.get(id); - if (instances == null) { - synchronized (this) { - // recheck in the synchronized block and recreate - if (recentNodes.get(id) == null) { - recentNodes.put(id, new LinkedList<NodeInstance>()); - } - instances = recentNodes.get(id); - } - } - return instances; - } - - /** - * Sort a the recent node list for a single role - * @param role role to sort - */ - private void sortRecentNodeList(int role) { - List<NodeInstance> nodesForRoleId = getRecentNodesForRoleId(role); - if (nodesForRoleId != null) { - Collections.sort(nodesForRoleId, new NodeInstance.Preferred(role)); - } - } - - /** - * Find a node for use - * @param role role - * @return the instance, or null for none - */ - @VisibleForTesting - public synchronized NodeInstance findRecentNodeForNewInstance(RoleStatus role) { - if (!role.isPlacementDesired()) { - // no data locality policy - return null; - } - int roleId = role.getKey(); - boolean strictPlacement = role.isStrictPlacement(); - NodeInstance nodeInstance = null; - // Get the list of possible targets. - // This is a live list: changes here are preserved - List<NodeInstance> targets = getRecentNodesForRoleId(roleId); - if (targets == null) { - // nothing to allocate on - return null; - } - - int cnt = targets.size(); - log.debug("There are {} node(s) to consider for {}", cnt, role.getName()); - for (int i = 0; i < cnt && nodeInstance == null; i++) { - NodeInstance candidate = targets.get(i); - if (candidate.getActiveRoleInstances(roleId) == 0) { - // no active instances: check failure statistics - if (strictPlacement - || (candidate.isOnline() && !candidate.exceedsFailureThreshold(role))) { - targets.remove(i); - // exit criteria for loop is now met - nodeInstance = candidate; - } else { - // too many failures for this node - log.info("Recent node failures is higher than threshold {}. Not requesting host {}", - role.getNodeFailureThreshold(), candidate.hostname); - } - } - } - - if (nodeInstance == null) { - log.info("No node found for {}", role.getName()); - } - return nodeInstance; - } - - /** - * Find a node for use - * @param role role - * @return the instance, or null for none - */ - @VisibleForTesting - public synchronized List<NodeInstance> findNodeForNewAAInstance(RoleStatus role) { - // all nodes that are live and can host the role; no attempt to exclude ones - // considered failing - return nodemap.findAllNodesForRole(role.getKey(), role.getLabelExpression()); - } - - /** - * Request an instance on a given node. - * An outstanding request is created & tracked, with the - * relevant node entry for that role updated. - *<p> - * The role status entries will also be tracked - * <p> - * Returns the request that is now being tracked. - * If the node instance is not null, it's details about the role is incremented - * - * @param node node to target or null for "any" - * @param role role to request - * @return the request - */ - public synchronized OutstandingRequest requestInstanceOnNode( - NodeInstance node, RoleStatus role, Resource resource) { - OutstandingRequest outstanding = outstandingRequests.newRequest(node, role.getKey()); - outstanding.buildContainerRequest(resource, role, now()); - return outstanding; - } - - /** - * Find a node for a role and request an instance on that (or a location-less - * instance) - * @param role role status - * @return a request ready to go, or null if this is an AA request and no - * location can be found. - */ - public synchronized OutstandingRequest requestContainerForRole(RoleStatus role) { - - if (role.isAntiAffinePlacement()) { - return requestContainerForAARole(role); - } else { - Resource resource = recordFactory.newResource(); - role.copyResourceRequirements(resource); - NodeInstance node = findRecentNodeForNewInstance(role); - return requestInstanceOnNode(node, role, resource); - } - } - - /** - * Find a node for an AA role and request an instance on that (or a location-less - * instance) - * @param role role status - * @return a request ready to go, or null if no location can be found. - */ - public synchronized OutstandingRequest requestContainerForAARole(RoleStatus role) { - List<NodeInstance> nodes = findNodeForNewAAInstance(role); - if (!nodes.isEmpty()) { - OutstandingRequest outstanding = outstandingRequests.newAARequest( - role.getKey(), nodes, role.getLabelExpression()); - Resource resource = recordFactory.newResource(); - role.copyResourceRequirements(resource); - outstanding.buildContainerRequest(resource, role, now()); - return outstanding; - } else { - log.warn("No suitable location for {}", role.getName()); - return null; - } - } - /** - * Get the list of active nodes ... walks the node map so - * is {@code O(nodes)} - * @param role role index - * @return a possibly empty list of nodes with an instance of that node - */ - public synchronized List<NodeInstance> listActiveNodes(int role) { - return nodemap.listActiveNodes(role); - } - - /** - * Get the node entry of a container - * @param container container to look up - * @return the entry - * @throws RuntimeException if the container has no hostname - */ - public NodeEntry getOrCreateNodeEntry(Container container) { - return getOrCreateNodeInstance(container).getOrCreate(container); - } - - /** - * Get the node instance of a container -always returns something - * @param container container to look up - * @return a (possibly new) node instance - * @throws RuntimeException if the container has no hostname - */ - public synchronized NodeInstance getOrCreateNodeInstance(Container container) { - return nodemap.getOrCreate(RoleHistoryUtils.hostnameOf(container)); - } - - /** - * Get the node instance of a host if defined - * @param hostname hostname to look up - * @return a node instance or null - * @throws RuntimeException if the container has no hostname - */ - public synchronized NodeInstance getExistingNodeInstance(String hostname) { - return nodemap.get(hostname); - } - - /** - * Get the node instance of a container <i>if there's an entry in the history</i> - * @param container container to look up - * @return a node instance or null - * @throws RuntimeException if the container has no hostname - */ - public synchronized NodeInstance getExistingNodeInstance(Container container) { - return nodemap.get(RoleHistoryUtils.hostnameOf(container)); - } - - /** - * Perform any pre-allocation operations on the list of allocated containers - * based on knowledge of system state. - * Currently this places requested hosts ahead of unrequested ones. - * @param allocatedContainers list of allocated containers - * @return list of containers potentially reordered - */ - public synchronized List<Container> prepareAllocationList(List<Container> allocatedContainers) { - - //partition into requested and unrequested - List<Container> requested = - new ArrayList<>(allocatedContainers.size()); - List<Container> unrequested = - new ArrayList<>(allocatedContainers.size()); - outstandingRequests.partitionRequests(this, allocatedContainers, requested, unrequested); - - //give the unrequested ones lower priority - requested.addAll(unrequested); - return requested; - } - - /** - * A container has been allocated on a node -update the data structures - * @param container container - * @param desiredCount desired #of instances - * @param actualCount current count of instances - * @return The allocation outcome - */ - public synchronized ContainerAllocationResults onContainerAllocated(Container container, - long desiredCount, - long actualCount) { - int role = ContainerPriority.extractRole(container); - - String hostname = RoleHistoryUtils.hostnameOf(container); - List<NodeInstance> nodeInstances = listRecentNodesForRoleId(role); - ContainerAllocationResults outcome = - outstandingRequests.onContainerAllocated(role, hostname, container); - if (desiredCount <= actualCount) { - // all outstanding requests have been satisfied - // clear all the lists, so returning nodes to the available set - List<NodeInstance> hosts = outstandingRequests.resetOutstandingRequests(role); - if (!hosts.isEmpty()) { - //add the list - log.info("Adding {} hosts for role {}", hosts.size(), role); - nodeInstances.addAll(hosts); - sortRecentNodeList(role); - } - } - return outcome; - } - - /** - * A container has been assigned to a role instance on a node -update the data structures - * @param container container - */ - public void onContainerAssigned(Container container) { - NodeInstance node = getOrCreateNodeInstance(container); - NodeEntry nodeEntry = node.getOrCreate(container); - nodeEntry.onStarting(); - log.debug("Node {} has updated NodeEntry {}", node, nodeEntry); - } - - /** - * Event: a container start has been submitted - * @param container container being started - * @param instance instance bound to the container - */ - public void onContainerStartSubmitted(Container container, - RoleInstance instance) { - // no actions here - } - - /** - * Container start event - * @param container container that just started - */ - public void onContainerStarted(Container container) { - NodeEntry nodeEntry = getOrCreateNodeEntry(container); - nodeEntry.onStartCompleted(); - touch(); - } - - /** - * A container failed to start: update the node entry state - * and return the container to the queue - * @param container container that failed - * @return true if the node was queued - */ - public boolean onNodeManagerContainerStartFailed(Container container) { - return markContainerFinished(container, false, true, ContainerOutcome.Failed); - } - - /** - * Does the RoleHistory have enough information about the YARN cluster - * to start placing AA requests? That is: has it the node map and - * any label information needed? - * @return true if the caller can start requesting AA nodes - */ - public boolean canPlaceAANodes() { - return nodeUpdateReceived.get(); - } - - /** - * Get the last time the nodes were updated from YARN - * @return the update time or zero if never updated. - */ - public long getNodesUpdatedTime() { - return nodesUpdatedTime.get(); - } - - /** - * Update failedNodes and nodemap based on the node state - * - * @param updatedNodes list of updated nodes - * @return true if a review should be triggered. - */ - public synchronized boolean onNodesUpdated(List<NodeReport> updatedNodes) { - log.debug("Updating {} nodes", updatedNodes.size()); - nodesUpdatedTime.set(now()); - nodeUpdateReceived.set(true); - int printed = 0; - boolean triggerReview = false; - for (NodeReport updatedNode : updatedNodes) { - String hostname = updatedNode.getNodeId() == null - ? "" - : updatedNode.getNodeId().getHost(); - NodeState nodeState = updatedNode.getNodeState(); - if (hostname.isEmpty() || nodeState == null) { - log.warn("Ignoring incomplete update"); - continue; - } - if (log.isDebugEnabled() && printed++ < 10) { - // log the first few, but avoid overloading the logs for a full cluster - // update - log.debug("Node \"{}\" is in state {}", hostname, nodeState); - } - // update the node; this also creates an instance if needed - boolean updated = nodemap.updateNode(hostname, updatedNode); - triggerReview |= updated; - } - return triggerReview; - } - - /** - * A container release request was issued - * @param container container submitted - */ - public void onContainerReleaseSubmitted(Container container) { - NodeEntry nodeEntry = getOrCreateNodeEntry(container); - nodeEntry.release(); - } - - /** - * App state notified of a container completed - * @param container completed container - * @return true if the node was queued - */ - public boolean onReleaseCompleted(Container container) { - return markContainerFinished(container, true, false, ContainerOutcome.Failed); - } - - /** - * App state notified of a container completed -but as - * it wasn't being released it is marked as failed - * - * @param container completed container - * @param shortLived was the container short lived? - * @param outcome - * @return true if the node is considered available for work - */ - public boolean onFailedContainer(Container container, - boolean shortLived, - ContainerOutcome outcome) { - return markContainerFinished(container, false, shortLived, outcome); - } - - /** - * Mark a container finished; if it was released then that is treated - * differently. history is {@code touch()}-ed - * - * - * @param container completed container - * @param wasReleased was the container released? - * @param shortLived was the container short lived? - * @param outcome - * @return true if the node was queued - */ - protected synchronized boolean markContainerFinished(Container container, - boolean wasReleased, - boolean shortLived, - ContainerOutcome outcome) { - NodeEntry nodeEntry = getOrCreateNodeEntry(container); - log.info("Finished container for node {}, released={}, shortlived={}", - nodeEntry.rolePriority, wasReleased, shortLived); - boolean available; - if (shortLived) { - nodeEntry.onStartFailed(); - available = false; - } else { - available = nodeEntry.containerCompleted(wasReleased, outcome); - maybeQueueNodeForWork(container, nodeEntry, available); - } - touch(); - return available; - } - - /** - * If the node is marked as available; queue it for assignments. - * Unsynced: requires caller to be in a sync block. - * @param container completed container - * @param nodeEntry node - * @param available available flag - * @return true if the node was queued - */ - private boolean maybeQueueNodeForWork(Container container, - NodeEntry nodeEntry, - boolean available) { - if (available) { - //node is free - nodeEntry.setLastUsed(now()); - NodeInstance ni = getOrCreateNodeInstance(container); - int roleId = ContainerPriority.extractRole(container); - log.debug("Node {} is now available for role id {}", ni, roleId); - listRecentNodesForRoleId(roleId).addFirst(ni); - } - return available; - } - - /** - * Print the history to the log. This is for testing and diagnostics - */ - public synchronized void dump() { - for (ProviderRole role : providerRoles) { - log.info(role.toString()); - List<NodeInstance> instances = listRecentNodesForRoleId(role.id); - log.info(" available: " + instances.size() - + " " + SliderUtils.joinWithInnerSeparator(" ", instances)); - } - - log.info("Nodes in Cluster: {}", getClusterSize()); - for (NodeInstance node : nodemap.values()) { - log.info(node.toFullString()); - } - } - - /** - * Build the mapping entry for persisting to the role history - * @return a mapping object - */ - public synchronized Map<CharSequence, Integer> buildMappingForHistoryFile() { - Map<CharSequence, Integer> mapping = new HashMap<>(getRoleSize()); - for (ProviderRole role : providerRoles) { - mapping.put(role.name, role.id); - } - return mapping; - } - - /** - * Get a clone of the available list - * @param role role index - * @return a clone of the list - */ - @VisibleForTesting - public List<NodeInstance> cloneRecentNodeList(int role) { - return new LinkedList<>(listRecentNodesForRoleId(role)); - } - - /** - * Get a snapshot of the outstanding placed request list - * @return a list of the requests outstanding at the time of requesting - */ - @VisibleForTesting - public List<OutstandingRequest> listPlacedRequests() { - return outstandingRequests.listPlacedRequests(); - } - - /** - * Get a snapshot of the outstanding placed request list - * @return a list of the requests outstanding at the time of requesting - */ - @VisibleForTesting - public List<OutstandingRequest> listOpenRequests() { - return outstandingRequests.listOpenRequests(); - } - - /** - * Escalate operation as triggered by external timer. - * @return a (usually empty) list of cancel/request operations. - */ - public synchronized List<AbstractRMOperation> escalateOutstandingRequests() { - return outstandingRequests.escalateOutstandingRequests(now()); - } - /** - * Escalate operation as triggered by external timer. - * @return a (usually empty) list of cancel/request operations. - */ - public List<AbstractRMOperation> cancelOutstandingAARequests() { - return outstandingRequests.cancelOutstandingAARequests(); - } - - /** - * Cancel a number of outstanding requests for a role -that is, not - * actual containers, just requests for new ones. - * @param role role - * @param toCancel number to cancel - * @return a list of cancellable operations. - */ - public List<AbstractRMOperation> cancelRequestsForRole(RoleStatus role, int toCancel) { - return role.isAntiAffinePlacement() ? - cancelRequestsForAARole(role, toCancel) - : cancelRequestsForSimpleRole(role, toCancel); - } - - /** - * Build the list of requests to cancel from the outstanding list. - * @param role role - * @param toCancel number to cancel - * @return a list of cancellable operations. - */ - private synchronized List<AbstractRMOperation> cancelRequestsForSimpleRole(RoleStatus role, int toCancel) { - Preconditions.checkArgument(toCancel > 0, - "trying to cancel invalid number of requests: " + toCancel); - List<AbstractRMOperation> results = new ArrayList<>(toCancel); - // first scan through the unplaced request list to find all of a role - int roleId = role.getKey(); - List<OutstandingRequest> requests = - outstandingRequests.extractOpenRequestsForRole(roleId, toCancel); - - // are there any left? - int remaining = toCancel - requests.size(); - // ask for some placed nodes - requests.addAll(outstandingRequests.extractPlacedRequestsForRole(roleId, remaining)); - - // build cancellations - for (OutstandingRequest request : requests) { - results.add(request.createCancelOperation()); - } - return results; - } - - /** - * Build the list of requests to cancel for an AA role. This reduces the number - * of outstanding pending requests first, then cancels any active request, - * before finally asking for any placed containers - * @param role role - * @param toCancel number to cancel - * @return a list of cancellable operations. - */ - private synchronized List<AbstractRMOperation> cancelRequestsForAARole(RoleStatus role, int toCancel) { - List<AbstractRMOperation> results = new ArrayList<>(toCancel); - int roleId = role.getKey(); - List<OutstandingRequest> requests = new ArrayList<>(toCancel); - // there may be pending requests which can be cancelled here - long pending = role.getPendingAntiAffineRequests(); - if (pending > 0) { - // there are some pending ones which can be cancelled first - long pendingToCancel = Math.min(pending, toCancel); - log.info("Cancelling {} pending AA allocations, leaving {}", toCancel, - pendingToCancel); - role.setPendingAntiAffineRequests(pending - pendingToCancel); - toCancel -= pendingToCancel; - } - if (toCancel > 0 && role.isAARequestOutstanding()) { - // not enough - log.info("Cancelling current AA request"); - // find the single entry which may be running - requests = outstandingRequests.extractOpenRequestsForRole(roleId, toCancel); - role.cancelOutstandingAARequest(); - toCancel--; - } - - // ask for some excess nodes - if (toCancel > 0) { - requests.addAll(outstandingRequests.extractPlacedRequestsForRole(roleId, toCancel)); - } - - // build cancellations - for (OutstandingRequest request : requests) { - results.add(request.createCancelOperation()); - } - return results; - } - -} http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/51c2b92c/slider-core/src/main/java/org/apache/slider/server/appmaster/state/RoleHistoryUtils.java ---------------------------------------------------------------------- diff --git a/slider-core/src/main/java/org/apache/slider/server/appmaster/state/RoleHistoryUtils.java b/slider-core/src/main/java/org/apache/slider/server/appmaster/state/RoleHistoryUtils.java deleted file mode 100644 index ea6197b..0000000 --- a/slider-core/src/main/java/org/apache/slider/server/appmaster/state/RoleHistoryUtils.java +++ /dev/null @@ -1,50 +0,0 @@ -/* - * 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.slider.server.appmaster.state; - -import org.apache.hadoop.yarn.api.records.Container; -import org.apache.hadoop.yarn.api.records.NodeId; -import org.apache.slider.common.tools.SliderUtils; - -public class RoleHistoryUtils { - - public static String hostnameOf(Container container) { - NodeId nodeId = container.getNodeId(); - if (nodeId== null) { - throw new RuntimeException("Container has no node ID: %s" + - SliderUtils.containerToString(container)); - } - return nodeId.getHost(); - } - - /** - * Decrement a value but hold it at zero. Usually a sanity check - * on counters tracking outstanding operations - * @param val value - * @return decremented value - */ - public static int decToFloor(int val) { - int v = val-1; - if (v < 0) { - v = 0; - } - return v; - } - -} http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/51c2b92c/slider-core/src/main/java/org/apache/slider/server/appmaster/state/RoleHostnamePair.java ---------------------------------------------------------------------- diff --git a/slider-core/src/main/java/org/apache/slider/server/appmaster/state/RoleHostnamePair.java b/slider-core/src/main/java/org/apache/slider/server/appmaster/state/RoleHostnamePair.java deleted file mode 100644 index 920887a..0000000 --- a/slider-core/src/main/java/org/apache/slider/server/appmaster/state/RoleHostnamePair.java +++ /dev/null @@ -1,75 +0,0 @@ -/* - * 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.slider.server.appmaster.state; - -import java.util.Objects; - -public class RoleHostnamePair { - - /** - * requested role - */ - public final int roleId; - - /** - * hostname -will be null if node==null - */ - public final String hostname; - - public RoleHostnamePair(int roleId, String hostname) { - this.roleId = roleId; - this.hostname = hostname; - } - - public int getRoleId() { - return roleId; - } - - public String getHostname() { - return hostname; - } - - @Override - public boolean equals(Object o) { - if (this == o) { - return true; - } - if (!(o instanceof RoleHostnamePair)) { - return false; - } - RoleHostnamePair that = (RoleHostnamePair) o; - return Objects.equals(roleId, that.roleId) && - Objects.equals(hostname, that.hostname); - } - - @Override - public int hashCode() { - return Objects.hash(roleId, hostname); - } - - @Override - public String toString() { - final StringBuilder sb = new StringBuilder( - "RoleHostnamePair{"); - sb.append("roleId=").append(roleId); - sb.append(", hostname='").append(hostname).append('\''); - sb.append('}'); - return sb.toString(); - } -} http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/51c2b92c/slider-core/src/main/java/org/apache/slider/server/appmaster/state/RoleInstance.java ---------------------------------------------------------------------- diff --git a/slider-core/src/main/java/org/apache/slider/server/appmaster/state/RoleInstance.java b/slider-core/src/main/java/org/apache/slider/server/appmaster/state/RoleInstance.java deleted file mode 100644 index 30cfec9..0000000 --- a/slider-core/src/main/java/org/apache/slider/server/appmaster/state/RoleInstance.java +++ /dev/null @@ -1,323 +0,0 @@ -/* - * 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.slider.server.appmaster.state; - -import com.google.common.base.Preconditions; -import org.apache.hadoop.registry.client.binding.RegistryTypeUtils; -import org.apache.hadoop.registry.client.types.Endpoint; -import org.apache.hadoop.registry.client.types.ProtocolTypes; -import org.apache.hadoop.yarn.api.records.Container; -import org.apache.hadoop.yarn.api.records.ContainerId; -import org.apache.hadoop.yarn.api.records.NodeId; -import org.apache.slider.api.ClusterNode; -import org.apache.slider.api.proto.Messages; -import org.apache.slider.api.types.ContainerInformation; -import org.apache.slider.common.tools.SliderUtils; - -import java.util.ArrayList; -import java.util.Arrays; -import java.util.List; - -/** - * Tracking information about a container - */ -public final class RoleInstance implements Cloneable { - - public Container container; - /** - * Container ID - */ - public final String id; - public long createTime; - public long startTime; - /** - * flag set when it is released, to know if it has - * already been targeted for termination - */ - public boolean released; - - /** - * Name of the role - */ - public String role; - public String group; - - /** - * Version of the app - */ - public String appVersion; - - /** - * Role Id; matches priority in resources.json - */ - public int roleId; - - /** - * state from StateValues - */ - public int state; - - /** - * Exit code: only valid if the state >= STOPPED - */ - public int exitCode; - - /** - * what was the command executed? - */ - public String command; - - /** - * Any diagnostics - */ - public String diagnostics; - - /** - * What is the tail output from the executed process (or [] if not started - * or the log cannot be picked up - */ - public String[] output; - - /** - * Any environment details - */ - public String[] environment; - - public String ip; - public String hostname; - public String host; - public String hostURL; - public ContainerAllocationOutcome placement; - - - /** - * A list of registered endpoints. - */ - private List<Endpoint> endpoints = - new ArrayList<>(2); - - public RoleInstance(ContainerAssignment assignment) { - this(assignment.container); - placement = assignment.placement; - } - /** - * Create an instance to track an allocated container - * @param container a container which must be non null, and have a non-null Id field. - */ - public RoleInstance(Container container) { - Preconditions.checkNotNull(container, "Null container"); - Preconditions.checkState(container.getId() != null, - "Null container ID"); - - this.container = container; - id = container.getId().toString(); - if (container.getNodeId() != null) { - host = container.getNodeId().getHost(); - } - if (container.getNodeHttpAddress() != null) { - hostURL = "http://"; + container.getNodeHttpAddress(); - } - } - - public ContainerId getId() { - return container.getId(); - } - - public NodeId getHost() { - return container.getNodeId(); - } - - @Override - public String toString() { - final StringBuilder sb = - new StringBuilder("RoleInstance{"); - sb.append("role='").append(role).append('\''); - sb.append(", id='").append(id).append('\''); - sb.append(", container=").append(SliderUtils.containerToString(container)); - sb.append(", createTime=").append(createTime); - sb.append(", startTime=").append(startTime); - sb.append(", released=").append(released); - sb.append(", roleId=").append(roleId); - sb.append(", host=").append(host); - sb.append(", hostURL=").append(hostURL); - sb.append(", state=").append(state); - sb.append(", placement=").append(placement); - sb.append(", exitCode=").append(exitCode); - sb.append(", command='").append(command).append('\''); - sb.append(", diagnostics='").append(diagnostics).append('\''); - sb.append(", output=").append(Arrays.toString(output)); - sb.append(", environment=").append(Arrays.toString(environment)); - sb.append('}'); - return sb.toString(); - } - - public ContainerId getContainerId() { - return container != null ? container.getId() : null; - } - - /** - * Generate the protobuf format of a request - * @return protobuf format. This excludes the Container info - */ - public Messages.RoleInstanceState toProtobuf() { - Messages.RoleInstanceState.Builder builder = - Messages.RoleInstanceState.newBuilder(); - if (container != null) { - builder.setName(container.getId().toString()); - } else { - builder.setName("unallocated instance"); - } - if (command != null) { - builder.setCommand(command); - } - if (environment != null) { - builder.addAllEnvironment(Arrays.asList(environment)); - } - if (diagnostics != null) { - builder.setDiagnostics(diagnostics); - } - builder.setExitCode(exitCode); - - if (output != null) { - builder.addAllOutput(Arrays.asList(output)); - } - if (role != null) { - builder.setRole(role); - } - builder.setRoleId(roleId); - builder.setState(state); - - builder.setReleased(released); - builder.setCreateTime(createTime); - builder.setStartTime(startTime); - builder.setHost(host); - builder.setHostURL(hostURL); - if (appVersion != null) { - builder.setAppVersion(appVersion); - } - return builder.build(); - } - - /** - * Build a serializable ClusterNode structure from this instance. - * This operation is unsynchronized. - * @return a serialized value. - */ - public ClusterNode toClusterNode() { - ClusterNode node; - if (container != null) { - node = new ClusterNode(container.getId()); - } else { - node = new ClusterNode(); - node.name = "unallocated instance"; - } - node.command = command; - node.createTime = createTime; - node.diagnostics = diagnostics; - if (environment != null) { - node.environment = Arrays.copyOf(environment, environment.length); - } - node.exitCode = exitCode; - node.ip = ip; - node.hostname = hostname; - node.host = host; - node.hostUrl = hostURL; - if (output != null) { - node.output = Arrays.copyOf(output, output.length); - } - node.released = released; - node.role = role; - node.roleId = roleId; - node.startTime = startTime ; - node.state = state; - - return node; - } - - /** - * Clone operation clones all the simple values but shares the - * Container object into the cloned copy -same with the output, - * diagnostics and env arrays. - * @return a clone of the object - * @throws CloneNotSupportedException - */ - @Override - public Object clone() throws CloneNotSupportedException { - RoleInstance cloned = (RoleInstance) super.clone(); - // clone the endpoint list, but not the values - cloned.endpoints = new ArrayList<Endpoint>(this.endpoints); - return cloned; - } - - /** - * Get the list of endpoints. - * @return the endpoint list. - */ - public List<Endpoint> getEndpoints() { - return endpoints; - } - - /** - * Add an endpoint registration - * @param endpoint endpoint (non-null) - */ - public void addEndpoint(Endpoint endpoint) { - Preconditions.checkArgument(endpoint != null); - endpoints.add(endpoint); - } - - /** - * Register a port endpoint as an inet-addr formatted endpoint, using the - * hostname as the first part of the address - * @param port port port - * @param api API API name - */ - public void registerPortEndpoint(int port, String api) { - Endpoint epr = - RegistryTypeUtils.inetAddrEndpoint(api, - ProtocolTypes.PROTOCOL_TCP, host, port); - addEndpoint(epr); - } - - /** - * Serialize. Some data structures (e.g output) - * may be shared - * @return a serialized form for marshalling as JSON - */ - public ContainerInformation serialize() { - ContainerInformation info = new ContainerInformation(); - info.containerId = id; - info.component = role; - info.appVersion = appVersion; - info.startTime = startTime; - info.createTime = createTime; - info.diagnostics = diagnostics; - info.state = state; - info.host = host; - info.hostURL = hostURL; - info.released = released ? Boolean.TRUE : null; - if (placement != null) { - info.placement = placement.toString(); - } - if (output != null) { - info.output = output; - } - return info; - } -} http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/51c2b92c/slider-core/src/main/java/org/apache/slider/server/appmaster/state/RoleStatus.java ---------------------------------------------------------------------- diff --git a/slider-core/src/main/java/org/apache/slider/server/appmaster/state/RoleStatus.java b/slider-core/src/main/java/org/apache/slider/server/appmaster/state/RoleStatus.java deleted file mode 100644 index 0a3a3c9..0000000 --- a/slider-core/src/main/java/org/apache/slider/server/appmaster/state/RoleStatus.java +++ /dev/null @@ -1,563 +0,0 @@ -/* - * 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.slider.server.appmaster.state; - -import com.codahale.metrics.Metric; -import com.codahale.metrics.MetricSet; -import com.google.common.base.Preconditions; -import org.apache.hadoop.yarn.api.records.Resource; -import org.apache.slider.api.types.ComponentInformation; -import org.apache.slider.api.types.RoleStatistics; -import org.apache.slider.providers.PlacementPolicy; -import org.apache.slider.providers.ProviderRole; -import org.apache.slider.server.appmaster.management.BoolMetricPredicate; -import org.apache.slider.server.appmaster.management.LongGauge; - -import java.io.Serializable; -import java.util.Comparator; -import java.util.HashMap; -import java.util.Map; - -/** - * Models the ongoing status of all nodes in an application. - * - * These structures are shared across the {@link AppState} and {@link RoleHistory} structures, - * and must be designed for synchronous access. Atomic counters are preferred to anything which - * requires synchronization. Where synchronized access is good is that it allows for - * the whole instance to be locked, for updating multiple entries. - */ -public final class RoleStatus implements Cloneable, MetricSet { - - private final String name; - private final String group; - - /** - * Role priority - */ - private final int key; - private final ProviderRole providerRole; - - private final LongGauge actual = new LongGauge(); - private final LongGauge completed = new LongGauge(); - private final LongGauge desired = new LongGauge(); - private final LongGauge failed = new LongGauge(); - private final LongGauge failedRecently = new LongGauge(0); - private final LongGauge limitsExceeded = new LongGauge(0); - private final LongGauge nodeFailed = new LongGauge(0); - /** Number of AA requests queued. */ - private final LongGauge pendingAntiAffineRequests = new LongGauge(0); - private final LongGauge preempted = new LongGauge(0); - private final LongGauge releasing = new LongGauge(); - private final LongGauge requested = new LongGauge(); - private final LongGauge started = new LongGauge(); - private final LongGauge startFailed = new LongGauge(); - private final LongGauge totalRequested = new LongGauge(); - - /** resource requirements */ - private Resource resourceRequirements; - - - /** any pending AA request */ - private volatile OutstandingRequest outstandingAArequest = null; - - - private String failureMessage = ""; - - public RoleStatus(ProviderRole providerRole) { - this.providerRole = providerRole; - this.name = providerRole.name; - this.group = providerRole.group; - this.key = providerRole.id; - } - - @Override - public Map<String, Metric> getMetrics() { - Map<String, Metric> metrics = new HashMap<>(15); - metrics.put("actual", actual); - metrics.put("completed", completed ); - metrics.put("desired", desired); - metrics.put("failed", failed); - metrics.put("limitsExceeded", limitsExceeded); - metrics.put("nodeFailed", nodeFailed); - metrics.put("preempted", preempted); - metrics.put("pendingAntiAffineRequests", pendingAntiAffineRequests); - metrics.put("releasing", releasing); - metrics.put("requested", requested); - metrics.put("preempted", preempted); - metrics.put("releasing", releasing ); - metrics.put("requested", requested); - metrics.put("started", started); - metrics.put("startFailed", startFailed); - metrics.put("totalRequested", totalRequested); - - metrics.put("outstandingAArequest", - new BoolMetricPredicate(new BoolMetricPredicate.Eval() { - @Override - public boolean eval() { - return isAARequestOutstanding(); - } - })); - return metrics; - } - - public String getName() { - return name; - } - - public String getGroup() { - return group; - } - - public int getKey() { - return key; - } - - public int getPriority() { - return getKey(); - } - - /** - * Get the placement policy enum, from the values in - * {@link PlacementPolicy} - * @return the placement policy for this role - */ - public int getPlacementPolicy() { - return providerRole.placementPolicy; - } - - public long getPlacementTimeoutSeconds() { - return providerRole.placementTimeoutSeconds; - } - - /** - * The number of failures on a specific node that can be tolerated - * before selecting a different node for placement - * @return - */ - public int getNodeFailureThreshold() { - return providerRole.nodeFailureThreshold; - } - - public boolean isExcludeFromFlexing() { - return hasPlacementPolicy(PlacementPolicy.EXCLUDE_FROM_FLEXING); - } - - public boolean isStrictPlacement() { - return hasPlacementPolicy(PlacementPolicy.STRICT); - } - - public boolean isAntiAffinePlacement() { - return hasPlacementPolicy(PlacementPolicy.ANTI_AFFINITY_REQUIRED); - } - - public boolean hasPlacementPolicy(int policy) { - return 0 != (getPlacementPolicy() & policy); - } - - public boolean isPlacementDesired() { - return !hasPlacementPolicy(PlacementPolicy.ANYWHERE); - } - - public long getDesired() { - return desired.get(); - } - - public void setDesired(long desired) { - this.desired.set(desired); - } - - public long getActual() { - return actual.get(); - } - - public long incActual() { - return actual.incrementAndGet(); - } - - public long decActual() { - return actual.decToFloor(1); - } - - /** - * Get the request count. - * @return a count of requested containers - */ - public long getRequested() { - return requested.get(); - } - - public long incRequested() { - totalRequested.incrementAndGet(); - return requested.incrementAndGet(); - } - - public void cancel(long count) { - requested.decToFloor(count); - } - - public void decRequested() { - cancel(1); - } - - public long getReleasing() { - return releasing.get(); - } - - public long incReleasing() { - return releasing.incrementAndGet(); - } - - public long decReleasing() { - return releasing.decToFloor(1); - } - - public long getFailed() { - return failed.get(); - } - - public long getFailedRecently() { - return failedRecently.get(); - } - - /** - * Reset the recent failure - * @return the number of failures in the "recent" window - */ - public long resetFailedRecently() { - return failedRecently.getAndSet(0); - } - - public long getLimitsExceeded() { - return limitsExceeded.get(); - } - - public long incPendingAntiAffineRequests(long v) { - return pendingAntiAffineRequests.addAndGet(v); - } - - /** - * Probe for an outstanding AA request being true - * @return true if there is an outstanding AA Request - */ - public boolean isAARequestOutstanding() { - return outstandingAArequest != null; - } - - /** - * expose the predicate {@link #isAARequestOutstanding()} as an integer, - * which is very convenient in tests - * @return 1 if there is an outstanding request; 0 if not - */ - public int getOutstandingAARequestCount() { - return isAARequestOutstanding()? 1: 0; - } - /** - * Note that a role failed, text will - * be used in any diagnostics if an exception - * is later raised. - * @param startupFailure flag to indicate this was a startup event - * @param text text about the failure - * @param outcome outcome of the container - */ - public synchronized void noteFailed(boolean startupFailure, String text, - ContainerOutcome outcome) { - if (text != null) { - failureMessage = text; - } - switch (outcome) { - case Preempted: - preempted.incrementAndGet(); - break; - - case Node_failure: - nodeFailed.incrementAndGet(); - failed.incrementAndGet(); - break; - - case Failed_limits_exceeded: // exceeded memory or CPU; app/configuration related - limitsExceeded.incrementAndGet(); - // fall through - case Failed: // application failure, possibly node related, possibly not - default: // anything else (future-proofing) - failed.incrementAndGet(); - failedRecently.incrementAndGet(); - //have a look to see if it short lived - if (startupFailure) { - incStartFailed(); - } - break; - } - } - - public long getStartFailed() { - return startFailed.get(); - } - - public synchronized void incStartFailed() { - startFailed.getAndIncrement(); - } - - public synchronized String getFailureMessage() { - return failureMessage; - } - - public long getCompleted() { - return completed.get(); - } - - public synchronized void setCompleted(int completed) { - this.completed.set(completed); - } - - public long incCompleted() { - return completed.incrementAndGet(); - } - public long getStarted() { - return started.get(); - } - - public synchronized void incStarted() { - started.incrementAndGet(); - } - - public long getTotalRequested() { - return totalRequested.get(); - } - - public long getPreempted() { - return preempted.get(); - } - - public long getNodeFailed() { - return nodeFailed.get(); - } - - public long getPendingAntiAffineRequests() { - return pendingAntiAffineRequests.get(); - } - - public void setPendingAntiAffineRequests(long pendingAntiAffineRequests) { - this.pendingAntiAffineRequests.set(pendingAntiAffineRequests); - } - - public long decPendingAntiAffineRequests() { - return pendingAntiAffineRequests.decToFloor(1); - } - - public OutstandingRequest getOutstandingAArequest() { - return outstandingAArequest; - } - - public void setOutstandingAArequest(OutstandingRequest outstandingAArequest) { - this.outstandingAArequest = outstandingAArequest; - } - - /** - * Complete the outstanding AA request (there's no check for one in progress, caller - * expected to have done that). - */ - public void completeOutstandingAARequest() { - setOutstandingAArequest(null); - } - - /** - * Cancel any outstanding AA request. Harmless if the role is non-AA, or - * if there are no outstanding requests. - */ - public void cancelOutstandingAARequest() { - if (outstandingAArequest != null) { - setOutstandingAArequest(null); - setPendingAntiAffineRequests(0); - decRequested(); - } - } - - /** - * Get the number of roles we are short of. - * nodes released are ignored. - * @return the positive or negative number of roles to add/release. - * 0 means "do nothing". - */ - public long getDelta() { - long inuse = getActualAndRequested(); - long delta = desired.get() - inuse; - if (delta < 0) { - //if we are releasing, remove the number that are already released. - delta += releasing.get(); - //but never switch to a positive - delta = Math.min(delta, 0); - } - return delta; - } - - /** - * Get count of actual and requested containers. This includes pending ones - * @return the size of the application when outstanding requests are included. - */ - public long getActualAndRequested() { - return actual.get() + requested.get(); - } - - @Override - public String toString() { - final StringBuilder sb = new StringBuilder("RoleStatus{"); - sb.append("name='").append(name).append('\''); - sb.append(", group=").append(group); - sb.append(", key=").append(key); - sb.append(", desired=").append(desired); - sb.append(", actual=").append(actual); - sb.append(", requested=").append(requested); - sb.append(", releasing=").append(releasing); - sb.append(", failed=").append(failed); - sb.append(", startFailed=").append(startFailed); - sb.append(", started=").append(started); - sb.append(", completed=").append(completed); - sb.append(", totalRequested=").append(totalRequested); - sb.append(", preempted=").append(preempted); - sb.append(", nodeFailed=").append(nodeFailed); - sb.append(", failedRecently=").append(failedRecently); - sb.append(", limitsExceeded=").append(limitsExceeded); - sb.append(", resourceRequirements=").append(resourceRequirements); - sb.append(", isAntiAffinePlacement=").append(isAntiAffinePlacement()); - if (isAntiAffinePlacement()) { - sb.append(", pendingAntiAffineRequests=").append(pendingAntiAffineRequests); - sb.append(", outstandingAArequest=").append(outstandingAArequest); - } - sb.append(", failureMessage='").append(failureMessage).append('\''); - sb.append(", providerRole=").append(providerRole); - sb.append('}'); - return sb.toString(); - } - - @Override - public synchronized Object clone() throws CloneNotSupportedException { - return super.clone(); - } - - /** - * Get the provider role - * @return the provider role - */ - public ProviderRole getProviderRole() { - return providerRole; - } - - /** - * Build the statistics map from the current data - * @return a map for use in statistics reports - */ - public Map<String, Integer> buildStatistics() { - ComponentInformation componentInformation = serialize(); - return componentInformation.buildStatistics(); - } - - /** - * Produced a serialized form which can be served up as JSON - * @return a summary of the current role status. - */ - public synchronized ComponentInformation serialize() { - ComponentInformation info = new ComponentInformation(); - info.name = name; - info.priority = getPriority(); - info.desired = desired.intValue(); - info.actual = actual.intValue(); - info.requested = requested.intValue(); - info.releasing = releasing.intValue(); - info.failed = failed.intValue(); - info.startFailed = startFailed.intValue(); - info.placementPolicy = getPlacementPolicy(); - info.failureMessage = failureMessage; - info.totalRequested = totalRequested.intValue(); - info.failedRecently = failedRecently.intValue(); - info.nodeFailed = nodeFailed.intValue(); - info.preempted = preempted.intValue(); - info.pendingAntiAffineRequestCount = pendingAntiAffineRequests.intValue(); - info.isAARequestOutstanding = isAARequestOutstanding(); - return info; - } - - /** - * Get the (possibly null) label expression for this role - * @return a string or null - */ - public String getLabelExpression() { - return providerRole.labelExpression; - } - - public Resource getResourceRequirements() { - return resourceRequirements; - } - - public void setResourceRequirements(Resource resourceRequirements) { - this.resourceRequirements = resourceRequirements; - } - - /** - * Compare two role status entries by name - */ - public static class CompareByName implements Comparator<RoleStatus>, - Serializable { - @Override - public int compare(RoleStatus o1, RoleStatus o2) { - return o1.getName().compareTo(o2.getName()); - } - } - - /** - * Compare two role status entries by key - */ - public static class CompareByKey implements Comparator<RoleStatus>, - Serializable { - @Override - public int compare(RoleStatus o1, RoleStatus o2) { - return (o1.getKey() < o2.getKey() ? -1 : (o1.getKey() == o2.getKey() ? 0 : 1)); - } - } - - /** - * Given a resource, set its requirements to those this role needs - * @param resource resource to configure - * @return the resource - */ - public Resource copyResourceRequirements(Resource resource) { - Preconditions.checkNotNull(resourceRequirements, - "Role resource requirements have not been set"); - resource.setMemory(resourceRequirements.getMemory()); - resource.setVirtualCores(resourceRequirements.getVirtualCores()); - return resource; - } - - public synchronized RoleStatistics getStatistics() { - RoleStatistics stats = new RoleStatistics(); - stats.activeAA = getOutstandingAARequestCount(); - stats.actual = actual.get(); - stats.desired = desired.get(); - stats.failed = failed.get(); - stats.limitsExceeded = limitsExceeded.get(); - stats.nodeFailed = nodeFailed.get(); - stats.preempted = preempted.get(); - stats.releasing = releasing.get(); - stats.requested = requested.get(); - stats.started = started.get(); - stats.startFailed = startFailed.get(); - stats.totalRequested = totalRequested.get(); - return stats; - } - -} http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/51c2b92c/slider-core/src/main/java/org/apache/slider/server/appmaster/state/SimpleReleaseSelector.java ---------------------------------------------------------------------- diff --git a/slider-core/src/main/java/org/apache/slider/server/appmaster/state/SimpleReleaseSelector.java b/slider-core/src/main/java/org/apache/slider/server/appmaster/state/SimpleReleaseSelector.java deleted file mode 100644 index b848096..0000000 --- a/slider-core/src/main/java/org/apache/slider/server/appmaster/state/SimpleReleaseSelector.java +++ /dev/null @@ -1,33 +0,0 @@ -/* - * 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.slider.server.appmaster.state; - -import java.util.List; - -/** - * Simplest release selector simply returns the list - */ -public class SimpleReleaseSelector implements ContainerReleaseSelector { - - @Override - public List<RoleInstance> sortCandidates(int roleId, - List<RoleInstance> candidates) { - return candidates; - } -}
