【Hadoop】【Yarn】ResourceManager启动流程源码分析

ResourceManager.start() ---> AbstractService.start()
    ResourceManager.serviceStart()
        ResourceManager.startWebApp()
            ResourceManager.serviceStart()
                CompositeService.serviceStart()//启动RM在初始化Service过程中注册的所有Service


            ResourceManager.transitionToActive()
                ResourceManager.startActiveServices()//启动集群升主的ActiveService，该过程同上面的过程

protected void serviceInit(Configuration conf) throws Exception {
  this.conf = conf;
 /**
 * RMContextImpl class holds two services context.
 * serviceContext : These services called as <b>Always On</b> services.
 * Services that need to run always irrespective of the HA state of the RM.
 * activeServiceCotext : Active services context. Services that need to run
 * only on the Active RM.
 * Note: If any new service to be added to context, add it to a right
 * context as per above description.
 * 上面的注释说的很清楚了，RMContextImpl其实就是用来记录RM中service的信息，分为两种，其中activeServiceContext中记录了仅仅在主RM上面运行的Service
 */
  this.rmContext = new RMContextImpl();
  rmContext.setResourceManager(this);


  this.configurationProvider =
      ConfigurationProviderFactory.getConfigurationProvider(conf);
  this.configurationProvider.init(this.conf);
  rmContext.setConfigurationProvider(configurationProvider);//设置配置项解析类


  // load core-site.xml
  loadConfigurationXml(YarnConfiguration.CORE_SITE_CONFIGURATION_FILE);//加载core-site.xml配置文件中的配置项值configuration对象中


  // Do refreshSuperUserGroupsConfiguration with loaded core-site.xml
  // Or use RM specific configurations to overwrite the common ones first
  // if they exist
  RMServerUtils.processRMProxyUsersConf(conf);
  ProxyUsers.refreshSuperUserGroupsConfiguration(this.conf);


  // load yarn-site.xml
  loadConfigurationXml(YarnConfiguration.YARN_SITE_CONFIGURATION_FILE);//加载yarn-site.xml配置文件中的配置项到configuration对象中


  validateConfigs(this.conf);//验证配置项
  
  // Set HA configuration should be done before login
  this.rmContext.setHAEnabled(HAUtil.isHAEnabled(this.conf));
  if (this.rmContext.isHAEnabled()) {
    HAUtil.verifyAndSetConfiguration(this.conf);//校验HA模式下特有的必要配置项
  }


  // Set UGI and do login
  // If security is enabled, use login user
  // If security is not enabled, use current user
  this.rmLoginUGI = UserGroupInformation.getCurrentUser();
  try {
    doSecureLogin();
  } catch(IOException ie) {
    throw new YarnRuntimeException("Failed to login", ie);
  }


  // register the handlers for all AlwaysOn services using setupDispatcher().
  //创建一个名为：RM Event dispatcher的ResourceManager的事件处理器，所有的常驻service都是使用该事件处理器进行处理。
  rmDispatcher = setupDispatcher();
  addIfService(rmDispatcher);
  rmContext.setDispatcher(rmDispatcher);


  // The order of services below should not be changed as services will be
  // started in same order
  // As elector service needs admin service to be initialized and started,
  // first we add admin service then elector service


  adminService = createAdminService();
  addService(adminService);
  rmContext.setRMAdminService(adminService);


  // elector must be added post adminservice
  if (this.rmContext.isHAEnabled()) {
    // If the RM is configured to use an embedded leader elector,
    // initialize the leader elector.
    if (HAUtil.isAutomaticFailoverEnabled(conf)
        && HAUtil.isAutomaticFailoverEmbedded(conf)) {
      EmbeddedElector elector = createEmbeddedElector();
      addIfService(elector);
      rmContext.setLeaderElectorService(elector);
    }
  }


  rmContext.setYarnConfiguration(conf);
  //创建并初始化ActiveService，即上面rmContext中设置的service全部都是AlwaysOn service，即和当前ResourceManager HA状态没有关系的Service。而这里设置的都是猪ResourceManager中独有的service，下面将对createAndInitActiveServices（）方法进行详细介绍
  createAndInitActiveServices(false);


  webAppAddress = WebAppUtils.getWebAppBindURL(this.conf,
                    YarnConfiguration.RM_BIND_HOST,
                    WebAppUtils.getRMWebAppURLWithoutScheme(this.conf));//ResourceManager web服务器的地址，用来接收并处理rest请求以及Yarn原生页面的请求。


  RMApplicationHistoryWriter rmApplicationHistoryWriter =
      createRMApplicationHistoryWriter();
  addService(rmApplicationHistoryWriter);
  rmContext.setRMApplicationHistoryWriter(rmApplicationHistoryWriter);//这里也是一个alwayson service，用来调度一个app历史写事件。然后上面提到的dispatcher会在使用特定的线程处理这个事件。


  // initialize the RM timeline collector first so that the system metrics
  // publisher can bind to it
  if (YarnConfiguration.timelineServiceV2Enabled(this.conf)) {
    RMTimelineCollectorManager timelineCollectorManager =
        createRMTimelineCollectorManager();
    addService(timelineCollectorManager);
    rmContext.setRMTimelineCollectorManager(timelineCollectorManager);
  }


  SystemMetricsPublisher systemMetricsPublisher =
      createSystemMetricsPublisher();
  addIfService(systemMetricsPublisher);
  rmContext.setSystemMetricsPublisher(systemMetricsPublisher);


  super.serviceInit(this.conf);
}

protected void createAndInitActiveServices(boolean fromActive) {
  activeServices = new RMActiveServices(this);
  activeServices.fromActive = fromActive;
  activeServices.init(conf);
}

@Override
protected void serviceInit(Configuration configuration) throws Exception {
  standByTransitionRunnable = new StandByTransitionRunnable();


  rmSecretManagerService = createRMSecretManagerService();
  addService(rmSecretManagerService);


  EventMetricsManager eventMetricsManager = createEventMetricsManager();
  rmContext.setEventMetricsManager(eventMetricsManager);


  RMAsyncService rmAsyncService = createRMAsyncService();
  addService(rmAsyncService);
  rmContext.setRmAsyncService(rmAsyncService);


  containerAllocationExpirer = new ContainerAllocationExpirer(rmDispatcher);
  addService(containerAllocationExpirer);
  rmContext.setContainerAllocationExpirer(containerAllocationExpirer);


  AMLivelinessMonitor amLivelinessMonitor = createAMLivelinessMonitor();
  addService(amLivelinessMonitor);
  rmContext.setAMLivelinessMonitor(amLivelinessMonitor);


  AMLivelinessMonitor amFinishingMonitor = createAMLivelinessMonitor();
  addService(amFinishingMonitor);
  rmContext.setAMFinishingMonitor(amFinishingMonitor);
  
  RMAppLifetimeMonitor rmAppLifetimeMonitor = createRMAppLifetimeMonitor();
  addService(rmAppLifetimeMonitor);
  rmContext.setRMAppLifetimeMonitor(rmAppLifetimeMonitor);


  RMNodeLabelsManager nlm = createNodeLabelManager();
  nlm.setRMContext(rmContext);
  addService(nlm);
  rmContext.setNodeLabelManager(nlm);


  AllocationTagsManager allocationTagsManager =
      createAllocationTagsManager();
  rmContext.setAllocationTagsManager(allocationTagsManager);


  PlacementConstraintManagerService placementConstraintManager =
      createPlacementConstraintManager();
  addService(placementConstraintManager);
  rmContext.setPlacementConstraintManager(placementConstraintManager);


  // add resource profiles here because it's used by AbstractYarnScheduler
  ResourceProfilesManager resourceProfilesManager =
      createResourceProfileManager();
  resourceProfilesManager.init(conf);
  rmContext.setResourceProfilesManager(resourceProfilesManager);


  RMDelegatedNodeLabelsUpdater delegatedNodeLabelsUpdater =
      createRMDelegatedNodeLabelsUpdater();
  if (delegatedNodeLabelsUpdater != null) {
    addService(delegatedNodeLabelsUpdater);
    rmContext.setRMDelegatedNodeLabelsUpdater(delegatedNodeLabelsUpdater);
  }


  recoveryEnabled = conf.getBoolean(YarnConfiguration.RECOVERY_ENABLED,
      YarnConfiguration.DEFAULT_RM_RECOVERY_ENABLED);


  RMStateStore rmStore = null;
  if (recoveryEnabled) {
    rmStore = RMStateStoreFactory.getStore(conf);
    boolean isWorkPreservingRecoveryEnabled =
        conf.getBoolean(
          YarnConfiguration.RM_WORK_PRESERVING_RECOVERY_ENABLED,
          YarnConfiguration.DEFAULT_RM_WORK_PRESERVING_RECOVERY_ENABLED);
    rmContext
        .setWorkPreservingRecoveryEnabled(isWorkPreservingRecoveryEnabled);
  } else {
    rmStore = new NullRMStateStore();
  }


  try {
    rmStore.setResourceManager(rm);
    rmStore.init(conf);
    rmStore.setRMDispatcher(rmDispatcher);
  } catch (Exception e) {
    // the Exception from stateStore.init() needs to be handled for
    // HA and we need to give up master status if we got fenced
    LOG.error("Failed to init state store", e);
    throw e;
  }
  rmContext.setStateStore(rmStore);


  if (UserGroupInformation.isSecurityEnabled()) {
    delegationTokenRenewer = createDelegationTokenRenewer();
    rmContext.setDelegationTokenRenewer(delegationTokenRenewer);
  }


  // Register event handler for NodesListManager
  nodesListManager = new NodesListManager(rmContext);
  rmDispatcher.register(NodesListManagerEventType.class, nodesListManager,
      rmContext.getEventMetricsManager()
          .getNodesListManagerEventTypeMetrics());
  addService(nodesListManager);
  rmContext.setNodesListManager(nodesListManager);


  // Initialize the scheduler
  scheduler = createScheduler();
  scheduler.setRMContext(rmContext);
  addIfService(scheduler);
  rmContext.setScheduler(scheduler);


  schedulerDispatcher = createSchedulerEventDispatcher();
  addIfService(schedulerDispatcher);


  rmDispatcher.register(SchedulerEventType.class, schedulerDispatcher,
      rmContext.getEventMetricsManager().getRMSchedulerTypeMetrics());


  // Register event handler for RmAppEvents
  rmDispatcher.register(RMAppEventType.class,
      new ApplicationEventDispatcher(rmContext),
      rmContext.getEventMetricsManager().getAppEventTypeMetrics());


  // Register event handler for RmAppAttemptEvents
  rmDispatcher.register(RMAppAttemptEventType.class,
      new ApplicationAttemptEventDispatcher(rmContext),
      rmContext.getEventMetricsManager().getAttemptMetrics());


  // Register event handler for RmNodes
  rmDispatcher.register(RMNodeEventType.class,
      new NodeEventDispatcher(rmContext),
      rmContext.getEventMetricsManager().getNodeEventTypeMetrics());


  nmLivelinessMonitor = createNMLivelinessMonitor();
  addService(nmLivelinessMonitor);


  resourceTracker = createResourceTrackerService();
  addService(resourceTracker);
  rmContext.setResourceTrackerService(resourceTracker);


  MetricsSystem ms = DefaultMetricsSystem.initialize("ResourceManager");
  if (fromActive) {
    JvmMetrics.reattach(ms, jvmMetrics);
    UserGroupInformation.reattachMetrics();
  } else {
    jvmMetrics = JvmMetrics.initSingleton("ResourceManager", null);
  }


  JvmPauseMonitor pauseMonitor = new JvmPauseMonitor();
  addService(pauseMonitor);
  jvmMetrics.setPauseMonitor(pauseMonitor);


  // Initialize the Reservation system
  if (conf.getBoolean(YarnConfiguration.RM_RESERVATION_SYSTEM_ENABLE,
      YarnConfiguration.DEFAULT_RM_RESERVATION_SYSTEM_ENABLE)) {
    reservationSystem = createReservationSystem();
    if (reservationSystem != null) {
      reservationSystem.setRMContext(rmContext);
      addIfService(reservationSystem);
      rmContext.setReservationSystem(reservationSystem);
      LOG.info("Initialized Reservation system");
    }
  }


  masterService = createApplicationMasterService();
  createAndRegisterOpportunisticDispatcher(masterService);
  addService(masterService) ;
  rmContext.setApplicationMasterService(masterService);




  applicationACLsManager = new ApplicationACLsManager(conf);


  queueACLsManager = createQueueACLsManager(scheduler, conf);
  appRecoveryHandler = RMAppsHandler.getRMAppsHandler(rm);
  rmContext.setRMAppsHandler(appRecoveryHandler);
  rmAppManager = createRMAppManager();
  // Register event handler for RMAppManagerEvents
  rmDispatcher.register(RMAppManagerEventType.class, rmAppManager,
      rmContext.getEventMetricsManager().getAppManagerEventTypeMetrics());


  clientRM = createClientRMService();
  addService(clientRM);
  rmContext.setClientRMService(clientRM);


  ClusterMetrics.getMetrics();
  applicationMasterLauncher = createAMLauncher();
  rmDispatcher.register(AMLauncherEventType.class,
      applicationMasterLauncher);


  addService(applicationMasterLauncher);
  if (UserGroupInformation.isSecurityEnabled()) {
    addService(delegationTokenRenewer);
    delegationTokenRenewer.setRMContext(rmContext);
  }


  if(HAUtil.isFederationEnabled(conf)) {
    String cId = YarnConfiguration.getClusterId(conf);
    if (cId.isEmpty()) {
      String errMsg =
          "Cannot initialize RM as Federation is enabled"
              + " but cluster id is not configured.";
      LOG.error(errMsg);
      throw new YarnRuntimeException(errMsg);
    }
    federationStateStoreService = createFederationStateStoreService();
    addIfService(federationStateStoreService);
    LOG.info("Initialized Federation membership.");
  }


  rmnmInfo = new RMNMInfo(rmContext, scheduler);


  if (conf.getBoolean(YarnConfiguration.YARN_API_SERVICES_ENABLE,
      false)) {
    SystemServiceManager systemServiceManager = createServiceManager();
    addIfService(systemServiceManager);
  }


  super.serviceInit(conf);
}

protected void serviceStart() throws Exception {
  if (this.rmContext.isHAEnabled()) {
    //如果将当前的ResourceManager的状态转移为Standby状态，如果之前是Active状态，依然转移为Standby状态。
    transitionToStandby(false);
  }
  //启动Yarn Webapp
  startWepApp();
  //如果是最小化集群
  if (getConfig().getBoolean(YarnConfiguration.IS_MINI_YARN_CLUSTER,
      false)) {
    int port = webApp.port();
    WebAppUtils.setRMWebAppPort(conf, port);
  }
  //
  super.serviceStart();


  // Non HA case, start after RM services are started.
  //如果是非HA集群，那么将该Resourcemanager节点的状态转移到Active状态
  if (!this.rmContext.isHAEnabled()) {
    transitionToActive();
  }
}

protected void serviceStart() throws Exception {
  List<Service> services = getServices();
  if (LOG.isDebugEnabled()) {
    LOG.debug(getName() + ": starting services, size=" + services.size());
  }
  for (Service service : services) {
    // start the service. If this fails that service
    // will be stopped and an exception raised
    service.start();
  }
  super.serviceStart();
}

@Override
public void serviceInit(Configuration conf) throws Exception {
  Configuration configuration = new Configuration(conf);
  super.serviceInit(conf);
  initScheduler(configuration);
  // Initialize SchedulingMonitorManager
  schedulingMonitorManager.initialize(rmContext, conf);
}

public void serviceStart() throws Exception {
  //启动调度器相关线程
  startSchedulerThreads();
  super.serviceStart();
}

private void startSchedulerThreads() {
  try {
    writeLock.lock();


    //ActivityManager用来用来存储节点和application申请资源的信息，主要包含了申请资源的start,add,update,finish操作。
    activitiesManager.start();
    if (scheduleAsynchronously) {
      Preconditions.checkNotNull(asyncSchedulerThreads,
          "asyncSchedulerThreads is null");


      //AsyncSchedulerThreads是用来分配container的线程。默认是一个，可以参数指定数量。
      for (Thread t : asyncSchedulerThreads) {
        t.start();
      }


      //资源申请提交服务线程
      resourceCommitterService.start();
    }
  } finally {
    writeLock.unlock();
  }
}

<property>
 
   <name>yarn.scheduler.capacity.schedule-asynchronously.enable</name>
 
   <value>true</value>
 
</property>




In addition to that, the 2.9.0/3.0.0 Yarn support specify multiple thread (by 
d
default is 1) to allocate containers.








<property>
 
   <name>yarn.scheduler.capacity.schedule-asynchronously.maximum-threads</name>
 
   <value>4</value>
 
</property>

protected void serviceStart() throws Exception {
  //更新线程，
  if (updateThread != null) {
    updateThread.start();
  }
  schedulingMonitorManager.startAll();
  super.serviceStart();
}

AbstractYarnScheduler.UpdateThread.runI()代码如下：
/**
 * Thread which calls {@link #update()} every
 * <code>updateInterval</code> milliseconds.
 */
private class UpdateThread extends Thread {
  @Override
  public void run() {
    while (!Thread.currentThread().isInterrupted()) {
      try {
        synchronized (updateThreadMonitor) {
          //其中updateInterval的意思是该定时线程调度的时间间隔，每间隔updateInterval时间执行下面的update方法
          updateThreadMonitor.wait(updateInterval);
        }
        //锁定调度器和重新计算资源分配和需求，检查是否有抢占。默认的实现是空的，FairScheduler对其进行了重写。CapacityScheduler没有重写该方法。
        update();
      } catch (InterruptedException ie) {
        LOG.warn("Scheduler UpdateThread interrupted. Exiting.");
        return;
      } catch (Exception e) {
        LOG.error("Exception in scheduler UpdateThread", e);
      }
    }
  }
}


@VisibleForTesting
public void update() {
  // do nothing by default
}


SchedulingMonitorManager.startAll();
public synchronized void startAll() {
  for (SchedulingMonitor schedulingMonitor : runningSchedulingMonitors
      .values()) {
    schedulingMonitor.start();
  }
}




 Map<String, SchedulingMonitor> runningSchedulingMonitors
runningSchedulingMonitors.put(configuredPolicy, new SchedulingMonitor(rmContext, policyInstance));
@Override
public void serviceStart() throws Exception {
  LOG.info("Starting SchedulingMonitor=" + getName());
  assert !stopped : "starting when already stopped";
  ses = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
    public Thread newThread(Runnable r) {
      Thread t = new Thread(r);
      t.setName(getName());
      return t;
    }
  });
  schedulePreemptionChecker();
  super.serviceStart();
}


private void schedulePreemptionChecker() {
  //启一个定时周期线程任务：PolicyInvoker
  handler = ses.scheduleAtFixedRate(new PolicyInvoker(),
      0, monitorInterval, TimeUnit.MILLISECONDS);
}

private class PolicyInvoker implements Runnable {
  @Override
  public void run() {
    try {
      if (monitorInterval != scheduleEditPolicy.getMonitoringInterval()) {
        handler.cancel(true);
        monitorInterval = scheduleEditPolicy.getMonitoringInterval();
        schedulePreemptionChecker();
      } else {
        invokePolicy();
      }
    } catch (Throwable t) {
      // The preemption monitor does not alter structures nor do structures
      // persist across invocations. Therefore, log, skip, and retry.
      LOG.error("Exception raised while executing preemption"
          + " checker, skip this run..., exception=", t);
    }
  }
}

org.apache.hadoop.yarn.server.resourcemanager.monitor.capacity.ProportionalCapacityPreemptionPolicy
private final SchedulingEditPolicy scheduleEditPolicy;


public void invokePolicy(){
   scheduleEditPolicy.editSchedule();
}

public synchronized void editSchedule() {
  updateConfigIfNeeded();


  long startTs = clock.getTime();


  CSQueue root = scheduler.getRootQueue();
  Resource clusterResources = Resources.clone(scheduler.getClusterResource());
  //重点就是这一行，用来选出并跟踪需要抢占的所有候选container，等待这些container运行完成，如果超时未完成，则强制杀死。
  containerBasedPreemptOrKill(root, clusterResources);


  if (LOG.isDebugEnabled()) {
    LOG.debug("Total time used=" + (clock.getTime() - startTs) + " ms.");
  }
}

private void containerBasedPreemptOrKill(CSQueue root,
    Resource clusterResources) {
  // Sync killable containers from scheduler when lazy preemption enabled
  if (lazyPreempionEnabled) {
    syncKillableContainersFromScheduler();
  }


  // All partitions to look at
  Set<String> partitions = new HashSet<>();
  partitions.addAll(scheduler.getRMContext()
      .getNodeLabelManager().getClusterNodeLabelNames());
  partitions.add(RMNodeLabelsManager.NO_LABEL);
  this.allPartitions = ImmutableSet.copyOf(partitions);


  // extract a summary of the queues from scheduler
  synchronized (scheduler) {
    queueToPartitions.clear();
    //----------------------------------------第一步--------------------------------------------------------
    //从Root队列开始递归拷贝子队列信息，该步骤的目的是为了生成所有队列的快照，防止集群队列信息变化导致资源抢占过程中出现问题。
    for (String partitionToLookAt : allPartitions) {
      cloneQueues(root, Resources
              .clone(nlm.getResourceByLabel(partitionToLookAt, clusterResources)),
          partitionToLookAt);
    }


    // Update effective priority of queues
  }


  this.leafQueueNames = ImmutableSet.copyOf(getLeafQueueNames(
      getQueueByPartition(CapacitySchedulerConfiguration.ROOT,
          RMNodeLabelsManager.NO_LABEL)));
  // compute total preemption allowed
  Resource totalPreemptionAllowed = Resources.multiply(clusterResources,
      percentageClusterPreemptionAllowed);


  //clear under served queues for every run
  partitionToUnderServedQueues.clear();


  // based on ideal allocation select containers to be preemptionCandidates from each
  // queue and each application
  Map<ApplicationAttemptId, Set<RMContainer>> toPreempt =
      new HashMap<>();
  Map<PreemptionCandidatesSelector, Map<ApplicationAttemptId,
      Set<RMContainer>>> toPreemptPerSelector =  new HashMap<>();;
  for (PreemptionCandidatesSelector selector :
      candidatesSelectionPolicies) {
    long startTime = 0;
    if (LOG.isDebugEnabled()) {
      LOG.debug(MessageFormat
          .format("Trying to use {0} to select preemption candidates",
              selector.getClass().getName()));
      startTime = clock.getTime();
    }
    Map<ApplicationAttemptId, Set<RMContainer>> curCandidates =
        selector.selectCandidates(toPreempt, clusterResources,
            totalPreemptionAllowed);
    toPreemptPerSelector.putIfAbsent(selector, curCandidates);


    if (LOG.isDebugEnabled()) {
      LOG.debug(MessageFormat
          .format("{0} uses {1} millisecond to run",
              selector.getClass().getName(), clock.getTime() - startTime));
      int totalSelected = 0;
      int curSelected = 0;
      for (Set<RMContainer> set : toPreempt.values()) {
        totalSelected += set.size();
      }
      for (Set<RMContainer> set : curCandidates.values()) {
        curSelected += set.size();
      }
      LOG.debug(MessageFormat
          .format("So far, total {0} containers selected to be preempted, {1}"
                  + " containers selected this round\n",
              totalSelected, curSelected));
    }
  }


  if (LOG.isDebugEnabled()) {
    logToCSV(new ArrayList<>(leafQueueNames));
  }


  // if we are in observeOnly mode return before any action is taken
  if (observeOnly) {
    return;
  }


  // TODO: need consider revert killable containers when no more demandings.
  // Since we could have several selectors to make decisions concurrently.
  // So computed ideal-allocation varies between different selectors.
  //
  // We may need to "score" killable containers and revert the most preferred
  // containers. The bottom line is, we shouldn't preempt a queue which is already
  // below its guaranteed resource.


  long currentTime = clock.getTime();


  pcsMap = toPreemptPerSelector;


  // preempt (or kill) the selected containers
  preemptOrkillSelectedContainerAfterWait(toPreemptPerSelector, currentTime);


  // cleanup staled preemption candidates
  cleanupStaledPreemptionCandidates(currentTime);
}