import org.apache.hadoop.yarn.server.resourcemanager.nodelabels.RMNodeLabelsManager; //导入依赖的package包/类
private static void normalizeNodeLabelExpressionInRequest(
    ResourceRequest resReq, QueueInfo queueInfo) {

  String labelExp = resReq.getNodeLabelExpression();

  // if queue has default label expression, and RR doesn't have, use the
  // default label expression of queue
  if (labelExp == null && queueInfo != null && ResourceRequest.ANY
      .equals(resReq.getResourceName())) {
    labelExp = queueInfo.getDefaultNodeLabelExpression();
  }

  // If labelExp still equals to null, set it to be NO_LABEL
  if (labelExp == null) {
    labelExp = RMNodeLabelsManager.NO_LABEL;
  }
  resReq.setNodeLabelExpression(labelExp);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.nodelabels.RMNodeLabelsManager; //导入依赖的package包/类
public static boolean checkQueueAccessToNode(Set<String> queueLabels,
    Set<String> nodeLabels) {
  // if queue's label is *, it can access any node
  if (queueLabels != null && queueLabels.contains(RMNodeLabelsManager.ANY)) {
    return true;
  }
  // any queue can access to a node without label
  if (nodeLabels == null || nodeLabels.isEmpty()) {
    return true;
  }
  // a queue can access to a node only if it contains any label of the node
  if (queueLabels != null
      && Sets.intersection(queueLabels, nodeLabels).size() > 0) {
    return true;
  }
  // sorry, you cannot access
  return false;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.nodelabels.RMNodeLabelsManager; //导入依赖的package包/类
public static void checkIfLabelInClusterNodeLabels(RMNodeLabelsManager mgr,
    Set<String> labels) throws IOException {
  if (mgr == null) {
    if (labels != null && !labels.isEmpty()) {
      throw new IOException("NodeLabelManager is null, please check");
    }
    return;
  }

  if (labels != null) {
    for (String label : labels) {
      if (!label.equals(RMNodeLabelsManager.ANY)
          && !mgr.containsNodeLabel(label)) {
        throw new IOException("NodeLabelManager doesn't include label = "
            + label + ", please check.");
      }
    }
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.nodelabels.RMNodeLabelsManager; //导入依赖的package包/类
public float getCapacity(String label) {
  if (StringUtils.equals(label, RMNodeLabelsManager.NO_LABEL) && isRoot) {
    return 1f;
  }
  
  return _get(label, CapacityType.CAP);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.nodelabels.RMNodeLabelsManager; //导入依赖的package包/类
@Override
public synchronized void updateClusterResource(Resource clusterResource,
    ResourceLimits resourceLimits) {
  // Update all children
  for (CSQueue childQueue : childQueues) {
    // Get ResourceLimits of child queue before assign containers
    ResourceLimits childLimits =
        getResourceLimitsOfChild(childQueue, clusterResource, resourceLimits);     
    childQueue.updateClusterResource(clusterResource, childLimits);
  }
  
  // Update metrics
  CSQueueUtils.updateQueueStatistics(
      resourceCalculator, this, parent,
      labelManager.getResourceByLabel(RMNodeLabelsManager.NO_LABEL, clusterResource), minimumAllocation);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.nodelabels.RMNodeLabelsManager; //导入依赖的package包/类
synchronized void allocateResource(Resource clusterResource, 
    Resource resource, Set<String> nodeLabels) {
  
  // Update usedResources by labels
  if (nodeLabels == null || nodeLabels.isEmpty()) {
    queueUsage.incUsed(resource);
  } else {
    Set<String> anls = (accessibleLabels.contains(RMNodeLabelsManager.ANY))
        ? labelManager.getClusterNodeLabels() : accessibleLabels;
    for (String label : Sets.intersection(anls, nodeLabels)) {
      queueUsage.incUsed(label, resource);
    }
  }

  ++numContainers;
  if (null == nodeLabels || nodeLabels.isEmpty()) {
    CSQueueUtils.updateQueueStatistics(resourceCalculator, this, getParent(),
        labelManager.getResourceByLabel(RMNodeLabelsManager.NO_LABEL, clusterResource), minimumAllocation);
  }
}
 

import org.apache.hadoop.yarn.server.resourcemanager.nodelabels.RMNodeLabelsManager; //导入依赖的package包/类
protected synchronized void releaseResource(Resource clusterResource,
    Resource resource, Set<String> nodeLabels) {
  // Update usedResources by labels
  if (null == nodeLabels || nodeLabels.isEmpty()) {
    queueUsage.decUsed(resource);
  } else {
    Set<String> anls = (accessibleLabels.contains(RMNodeLabelsManager.ANY))
        ? labelManager.getClusterNodeLabels() : accessibleLabels;
    for (String label : Sets.intersection(anls, nodeLabels)) {
      queueUsage.decUsed(label, resource);
    }
  }

  if (null == nodeLabels || nodeLabels.isEmpty()) {
    CSQueueUtils.updateQueueStatistics(resourceCalculator, this, getParent(),
        labelManager.getResourceByLabel(RMNodeLabelsManager.NO_LABEL, clusterResource), minimumAllocation);
  }
  --numContainers;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.nodelabels.RMNodeLabelsManager; //导入依赖的package包/类
private Resource getCurrentLimitResource(String nodeLabel,
    Resource clusterResource, ResourceLimits currentResourceLimits) {
  /*
   * Current limit resource: For labeled resource: limit = queue-max-resource
   * (TODO, this part need update when we support labeled-limit) For
   * non-labeled resource: limit = min(queue-max-resource,
   * limit-set-by-parent)
   */
  Resource queueMaxResource =
      Resources.multiplyAndNormalizeDown(resourceCalculator,
          labelManager.getResourceByLabel(nodeLabel, clusterResource),
          queueCapacities.getAbsoluteMaximumCapacity(nodeLabel), minimumAllocation);
  if (nodeLabel.equals(RMNodeLabelsManager.NO_LABEL)) {
    return Resources.min(resourceCalculator, clusterResource,
        queueMaxResource, currentResourceLimits.getLimit());
  }
  return queueMaxResource;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.nodelabels.RMNodeLabelsManager; //导入依赖的package包/类
@Override
public synchronized void reinitialize(CSQueue newlyParsedQueue,
    Resource clusterResource) throws IOException {
  // Sanity check
  if (!(newlyParsedQueue instanceof ReservationQueue)
      || !newlyParsedQueue.getQueuePath().equals(getQueuePath())) {
    throw new IOException("Trying to reinitialize " + getQueuePath()
        + " from " + newlyParsedQueue.getQueuePath());
  }
  super.reinitialize(newlyParsedQueue, clusterResource);
  CSQueueUtils.updateQueueStatistics(
      parent.schedulerContext.getResourceCalculator(), newlyParsedQueue,
      parent, labelManager.getResourceByLabel(RMNodeLabelsManager.NO_LABEL, clusterResource),
      parent.schedulerContext.getMinimumResourceCapability());
  updateQuotas(parent.getUserLimitForReservation(),
      parent.getUserLimitFactor(),
      parent.getMaxApplicationsForReservations(),
      parent.getMaxApplicationsPerUserForReservation());
}
 

import org.apache.hadoop.yarn.server.resourcemanager.nodelabels.RMNodeLabelsManager; //导入依赖的package包/类
private void updateCurrentResourceLimits(
    ResourceLimits currentResourceLimits, Resource clusterResource) {
  // TODO: need consider non-empty node labels when resource limits supports
  // node labels
  // Even if ParentQueue will set limits respect child's max queue capacity,
  // but when allocating reserved container, CapacityScheduler doesn't do
  // this. So need cap limits by queue's max capacity here.
  this.cachedResourceLimitsForHeadroom = new ResourceLimits(currentResourceLimits.getLimit());
  Resource queueMaxResource =
      Resources.multiplyAndNormalizeDown(resourceCalculator, labelManager
          .getResourceByLabel(RMNodeLabelsManager.NO_LABEL, clusterResource),
          queueCapacities
              .getAbsoluteMaximumCapacity(RMNodeLabelsManager.NO_LABEL),
          minimumAllocation);
  this.cachedResourceLimitsForHeadroom.setLimit(Resources.min(resourceCalculator,
      clusterResource, queueMaxResource, currentResourceLimits.getLimit()));
}
 

import org.apache.hadoop.yarn.server.resourcemanager.nodelabels.RMNodeLabelsManager; //导入依赖的package包/类
@Before
public void setUp() throws Exception {
  resourceManager = new ResourceManager() {
    @Override
    protected RMNodeLabelsManager createNodeLabelManager() {
      RMNodeLabelsManager mgr = new NullRMNodeLabelsManager();
      mgr.init(getConfig());
      return mgr;
    }
  };
  CapacitySchedulerConfiguration csConf 
     = new CapacitySchedulerConfiguration();
  setupQueueConfiguration(csConf);
  YarnConfiguration conf = new YarnConfiguration(csConf);
  conf.setClass(YarnConfiguration.RM_SCHEDULER, 
      CapacityScheduler.class, ResourceScheduler.class);
  resourceManager.init(conf);
  resourceManager.getRMContext().getContainerTokenSecretManager().rollMasterKey();
  resourceManager.getRMContext().getNMTokenSecretManager().rollMasterKey();
  ((AsyncDispatcher)resourceManager.getRMContext().getDispatcher()).start();
  mockContext = mock(RMContext.class);
  when(mockContext.getConfigurationProvider()).thenReturn(
      new LocalConfigurationProvider());
}
 

import org.apache.hadoop.yarn.server.resourcemanager.nodelabels.RMNodeLabelsManager; //导入依赖的package包/类
@Test
public void testQueueParsingWhenLabelsNotExist() throws IOException {
  YarnConfiguration conf = new YarnConfiguration();
  CapacitySchedulerConfiguration csConf =
      new CapacitySchedulerConfiguration(conf);
  setupQueueConfigurationWithLabels(csConf);

  CapacityScheduler capacityScheduler = new CapacityScheduler();
  RMContextImpl rmContext =
      new RMContextImpl(null, null, null, null, null, null,
          new RMContainerTokenSecretManager(csConf),
          new NMTokenSecretManagerInRM(csConf),
          new ClientToAMTokenSecretManagerInRM(), null);
  
  RMNodeLabelsManager nodeLabelsManager = new NullRMNodeLabelsManager();
  nodeLabelsManager.init(conf);
  nodeLabelsManager.start();
  
  rmContext.setNodeLabelManager(nodeLabelsManager);
  capacityScheduler.setConf(csConf);
  capacityScheduler.setRMContext(rmContext);
  capacityScheduler.init(csConf);
  capacityScheduler.start();
  ServiceOperations.stopQuietly(capacityScheduler);
  ServiceOperations.stopQuietly(nodeLabelsManager);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.nodelabels.RMNodeLabelsManager; //导入依赖的package包/类
public static boolean checkResourceRequestMatchingNodePartition(
    String requestedPartition, String nodePartition,
    SchedulingMode schedulingMode) {
  // We will only look at node label = nodeLabelToLookAt according to
  // schedulingMode and partition of node.
  String nodePartitionToLookAt = null;
  if (schedulingMode == SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY) {
    nodePartitionToLookAt = nodePartition;
  } else {
    nodePartitionToLookAt = RMNodeLabelsManager.NO_LABEL;
  }

  if (null == requestedPartition) {
    requestedPartition = RMNodeLabelsManager.NO_LABEL;
  }
  return requestedPartition.equals(nodePartitionToLookAt);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.nodelabels.RMNodeLabelsManager; //导入依赖的package包/类
private boolean checkHeadroom(Resource clusterResource,
    ResourceLimits currentResourceLimits, Resource required,
    FiCaSchedulerNode node) {
  // If headroom + currentReservation < required, we cannot allocate this
  // require
  Resource resourceCouldBeUnReserved = application.getCurrentReservation();
  if (!application.getCSLeafQueue().getReservationContinueLooking()
      || !node.getPartition().equals(RMNodeLabelsManager.NO_LABEL)) {
    // If we don't allow reservation continuous looking, OR we're looking at
    // non-default node partition, we won't allow to unreserve before
    // allocation.
    resourceCouldBeUnReserved = Resources.none();
  }
  return Resources.greaterThanOrEqual(rc, clusterResource, Resources.add(
      currentResourceLimits.getHeadroom(), resourceCouldBeUnReserved),
      required);
}
 

import org.apache.hadoop.yarn.server.resourcemanager.nodelabels.RMNodeLabelsManager; //导入依赖的package包/类
public boolean accessibleToPartition(String nodePartition) {
  // if queue's label is *, it can access any node
  if (accessibleLabels != null
      && accessibleLabels.contains(RMNodeLabelsManager.ANY)) {
    return true;
  }
  // any queue can access to a node without label
  if (nodePartition == null
      || nodePartition.equals(RMNodeLabelsManager.NO_LABEL)) {
    return true;
  }
  // a queue can access to a node only if it contains any label of the node
  if (accessibleLabels != null && accessibleLabels.contains(nodePartition)) {
    return true;
  }
  // sorry, you cannot access
  return false;
}
 

import org.apache.hadoop.yarn.server.resourcemanager.nodelabels.RMNodeLabelsManager; //导入依赖的package包/类
private void updateCurrentResourceLimits(
    ResourceLimits currentResourceLimits, Resource clusterResource) {
  // TODO: need consider non-empty node labels when resource limits supports
  // node labels
  // Even if ParentQueue will set limits respect child's max queue capacity,
  // but when allocating reserved container, CapacityScheduler doesn't do
  // this. So need cap limits by queue's max capacity here.
  this.cachedResourceLimitsForHeadroom =
      new ResourceLimits(currentResourceLimits.getLimit());
  Resource queueMaxResource =
      Resources.multiplyAndNormalizeDown(resourceCalculator, labelManager
          .getResourceByLabel(RMNodeLabelsManager.NO_LABEL, clusterResource),
          queueCapacities
              .getAbsoluteMaximumCapacity(RMNodeLabelsManager.NO_LABEL),
          minimumAllocation);
  this.cachedResourceLimitsForHeadroom.setLimit(Resources.min(
      resourceCalculator, clusterResource, queueMaxResource,
      currentResourceLimits.getLimit()));
}
 

import org.apache.hadoop.yarn.server.resourcemanager.nodelabels.RMNodeLabelsManager; //导入依赖的package包/类
@Test
public void testRemoveClusterNodeLabelsWithCentralizedConfigurationDisabled()
    throws IOException, YarnException {
  // create RM and set it's ACTIVE
  MockRM rm = new MockRM();
  ((RMContextImpl) rm.getRMContext())
      .setHAServiceState(HAServiceState.ACTIVE);
  RMNodeLabelsManager labelMgr = rm.rmContext.getNodeLabelManager();
  rm.adminService.isCentralizedNodeLabelConfiguration = false;

  // by default, distributed configuration for node label is disabled, this
  // should pass
  labelMgr.addToCluserNodeLabelsWithDefaultExclusivity(ImmutableSet.of("x", "y"));
  rm.adminService
      .removeFromClusterNodeLabels(RemoveFromClusterNodeLabelsRequest
          .newInstance((Set<String>) ImmutableSet.of("x")));

  Set<String> clusterNodeLabels = labelMgr.getClusterNodeLabelNames();
  assertEquals(1,clusterNodeLabels.size());
  rm.close();
}
 

import org.apache.hadoop.yarn.server.resourcemanager.nodelabels.RMNodeLabelsManager; //导入依赖的package包/类
public void setup() {
  resourceManager = new ResourceManager() {
    @Override
    protected RMNodeLabelsManager createNodeLabelManager() {
      RMNodeLabelsManager mgr = new NullRMNodeLabelsManager();
      mgr.init(getConfig());
      return mgr;
    }
  };

  YarnConfiguration conf = new YarnConfiguration();
  conf.setClass(YarnConfiguration.RM_SCHEDULER, CapacityScheduler.class,
    ResourceScheduler.class);
  resourceManager.init(conf);
  resourceManager.getRMContext().getContainerTokenSecretManager()
    .rollMasterKey();
  resourceManager.getRMContext().getNMTokenSecretManager().rollMasterKey();
  ((AsyncDispatcher) resourceManager.getRMContext().getDispatcher()).start();
}
 

import org.apache.hadoop.yarn.server.resourcemanager.nodelabels.RMNodeLabelsManager; //导入依赖的package包/类
@Inject QueuesBlock(ResourceManager rm, CSQInfo info) {
  cs = (CapacityScheduler) rm.getResourceScheduler();
  csqinfo = info;
  RMNodeLabelsManager nodeLabelManager =
      rm.getRMContext().getNodeLabelManager();
  nodeLabelsInfo = nodeLabelManager.pullRMNodeLabelsInfo();      
}
 

import org.apache.hadoop.yarn.server.resourcemanager.nodelabels.RMNodeLabelsManager; //导入依赖的package包/类
@Override
protected void render(Block html) {
  TBODY<TABLE<Hamlet>> tbody = html.table("#nodelabels").
      thead().
      tr().
      th(".name", "Label Name").
      th(".numOfActiveNMs", "Num Of Active NMs").
      th(".totalResource", "Total Resource").
      _()._().
      tbody();
  
  RMNodeLabelsManager nlm = rm.getRMContext().getNodeLabelManager();
  for (NodeLabel info : nlm.pullRMNodeLabelsInfo()) {
    TR<TBODY<TABLE<Hamlet>>> row =
        tbody.tr().td(
            info.getLabelName().isEmpty() ? "<NO_LABEL>" : info
                .getLabelName());
    int nActiveNMs = info.getNumActiveNMs();
    if (nActiveNMs > 0) {
      row = row.td()
      .a(url("nodes",
          "?" + YarnWebParams.NODE_LABEL + "=" + info.getLabelName()),
          String.valueOf(nActiveNMs))
       ._();
    } else {
      row = row.td(String.valueOf(nActiveNMs));
    }
    row.td(info.getResource().toString())._();
  }
  tbody._()._();
}