Hadoop3.2.1 【 YARN 】源码分析 : ResourceManager # Application管理

短命女 2022-12-28 04:11 192阅读 0赞

# 一.前言 #

在YARN中， Application是指应用程序， 它可能启动多个运行实例， 每个运行实例由一个ApplicationMaster与一组该ApplicationMaster启动的任务组成， 它拥有名称、 队列名、优先级等属性， 是一个比较宽泛的概念， 可以是一个MapReduce作业、 一个DAG应用程序， 甚至可以是一个Storm集群实例。  
YARN中Application管理涉及应用程序的权限管理、启动与关闭、 生命周期管理等…  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70]

# 二. ApplicationACLsManager #

ApplicationACLsManager负责管理应用程序的访问权限， 包含两部分权限： **查看权限和修改权限**。 其中， **查看权限主要用于查看应用程序基本信息， 比如运行时间、 优先级等信息； 而修改权限则主要用于修改应用程序优先级、 杀死应用程序等**。 **默认情况下， 任意一个普通用户可以查看所有其他用户的应用程序**。 用户可以为自己的应用程序设置具有访问权限的用户列表， 具体方法是在客户端使用ContainerLaunchContext\#newInstance构造ContainerLaunchContext实例时将其作为参数传入。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 1]

## 2.1. 属性 ##

有以下几个属性,代码如下:

// 访问控制列表
      private static AccessControlList DEFAULT_YARN_APP_ACL   = new AccessControlList(YarnConfiguration.DEFAULT_YARN_APP_ACL);
    
      // 配置文件
      private final Configuration conf;
    
      // AdminACLsManager
      private final AdminACLsManager adminAclsManager;
    
      // ApplicationId访问控制清单
      private final ConcurrentMap<ApplicationId, Map<ApplicationAccessType, AccessControlList>> applicationACLS  = new ConcurrentHashMap<ApplicationId, Map<ApplicationAccessType, AccessControlList>>();

## 2.2. 构造方法 ##

没啥可说的,主要是构造了一个AdminACLsManager

public ApplicationACLsManager(Configuration conf) {
        this.conf = conf;
        this.adminAclsManager = new AdminACLsManager(this.conf);
      }

## 2.4. ApplicationAccessType 访问权限类型 ##

application访问权限类型, 一共就两种: 查看权限\[VIEW\_APP\]和修改权限\[MODIFY\_APP\] .

**查看权限\[VIEW\_APP\]** : 主要用于查看应用程序基本信息， 比如运行时间、 优先级等信息； 而修改权限则主要用于修改应用程序优先级、 杀死应用程序等  
\*\*修改权限\[MODIFY\_APP\] \*\*: 主要用于修改应用程序优先级、 杀死应用程序等

## 2.3. AdminACLsManager admin权限管理类型 ##

admin用户的访问控制列表管理, ApplicationACLsManager 的构造方法中进行实例化.  
我们就看看一下构造方法就可以了, 其实就是构造一个admin用户的访问控制列表: AccessControlList  
其他的就是获取当前用户,以及判断当前用户是否是admin用户.

/**
       * Constructs and initializes this AdminACLsManager
       *
       * @param conf configuration for this object to use
       */
      public AdminACLsManager(Configuration conf) {
    
        // yarn.admin.acl : *
        this.adminAcl = new AccessControlList(conf.get(
              YarnConfiguration.YARN_ADMIN_ACL,
              YarnConfiguration.DEFAULT_YARN_ADMIN_ACL));
        try {
          // 获取当前用户
          owner = UserGroupInformation.getCurrentUser();
          // 设置用户 adminAcl
          adminAcl.addUser(owner.getShortUserName());
        } catch (IOException e){
          LOG.warn("Could not add current user to admin:" + e);
          throw new YarnRuntimeException(e);
        }
        // 是否开启权限控制:  yarn.acl.enable : false
        aclsEnabled = conf.getBoolean(YarnConfiguration.YARN_ACL_ENABLE,
            YarnConfiguration.DEFAULT_YARN_ACL_ENABLE);
      }

## 2.4. areACLsEnabled() ##

返回 是否开启acl权限控制,默认false

public boolean areACLsEnabled() {
        return adminAclsManager.areACLsEnabled();
      }

## 2.5. addApplication ##

为application添加访问控制列表

public void addApplication(ApplicationId appId,
          Map<ApplicationAccessType, String> acls) {
        Map<ApplicationAccessType, AccessControlList> finalMap
            = new HashMap<ApplicationAccessType, AccessControlList>(acls.size());
        for (Entry<ApplicationAccessType, String> acl : acls.entrySet()) {
          finalMap.put(acl.getKey(), new AccessControlList(acl.getValue()));
        }
        this.applicationACLS.put(appId, finalMap);
      }
    
      // 移除application权限控制.
      public void removeApplication(ApplicationId appId) {
        this.applicationACLS.remove(appId);
      }

## 2.6. checkAccess 检查权限 ##

*  如果鉴权被启用, 检查用户是否有指定的application权限
 *  admin用户, application的所有者,授权用户, 可以拥有application上的所有权限
 *  非application所有者,需要检查权限.

/**
       * 如果鉴权被启用, 检查用户是否有指定的application权限
       * application的所有者可以拥有application上的所有权限
       * 非application所有者,检查权限
       *
       * If authorization is enabled, checks whether the user (in the callerUGI) is
       * authorized to perform the access specified by 'applicationAccessType' on
       * the application by checking if the user is applicationOwner or part of
       * application ACL for the specific access-type.
       * <ul>
       * <li>The owner of the application can have all access-types on the
       * application</li>
       * <li>For all other users/groups application-acls are checked</li>
       * </ul>
       * 
       * @param callerUGI
       * @param applicationAccessType
       * @param applicationOwner
       * @param applicationId
       */
      public boolean checkAccess(UserGroupInformation callerUGI,
          ApplicationAccessType applicationAccessType, String applicationOwner,
          ApplicationId applicationId) {
    
        if (LOG.isDebugEnabled()) {
          LOG.debug("Verifying access-type " + applicationAccessType + " for "
              + callerUGI + " on application " + applicationId + " owned by "
              + applicationOwner);
        }
    
        // 获取用户
        String user = callerUGI.getShortUserName();
    
        // 是否开启权限控制
        if (!areACLsEnabled()) {
          return true;
        }
    
        AccessControlList applicationACL = DEFAULT_YARN_APP_ACL;
        // 获取application的权限
        Map<ApplicationAccessType, AccessControlList> acls = this.applicationACLS.get(applicationId);
        if (acls == null) {
          if (LOG.isDebugEnabled()) {
            LOG.debug("ACL not found for application "
                + applicationId + " owned by "
                + applicationOwner + ". Using default ["
                + YarnConfiguration.DEFAULT_YARN_APP_ACL + "]");
          }
        } else {
          // 获取授权列表
          AccessControlList applicationACLInMap = acls.get(applicationAccessType);
          if (applicationACLInMap != null) {
            // 如果权限不为空
            applicationACL = applicationACLInMap;
          } else if (LOG.isDebugEnabled()) {
            LOG.debug("ACL not found for access-type " + applicationAccessType
                + " for application " + applicationId + " owned by "
                + applicationOwner + ". Using default ["
                + YarnConfiguration.DEFAULT_YARN_APP_ACL + "]");
          }
        }
        // 如果是管理员或者application的创建或者已经分配权限,直接返回true
        // Allow application-owner for any type of access on the application
        if (this.adminAclsManager.isAdmin(callerUGI)
            || user.equals(applicationOwner)
            || applicationACL.isUserAllowed(callerUGI)) {
          return true;
        }
        return false;
      }

## 2.7. AccessControlList 访问控制列表 ##

### .2.1. 属性 ###

这里比较重要的是 users, groups 和缓存 groupsMapping .

// 所有用户的 ACL权限控制 : * 
      // Indicates an ACL string that represents access to all users
      public static final String WILDCARD_ACL_VALUE = "*";
    
      // 初始容量 256
      private static final int INITIAL_CAPACITY = 256;
    
      // Set of users who are granted access.
      private Collection<String> users;
      // Set of groups which are granted access
      private Collection<String> groups;
      // Whether all users are granted access.
      private boolean allAllowed;
    
      //  缓存 group 映射信息
      private Groups groupsMapping = Groups.getUserToGroupsMappingService(new Configuration());

### .2.2. 构造方法 ###

入参为 acl字符串, acl字符串默认以空格进行拆分, 然后再以逗号\[,\]分别区分用户名和用户组. 示例 : “user1,user2 group1,group2”

/**
       * Construct a new ACL from a String representation of the same.
       * 
       * The String is a a comma separated list of users and groups.
       * The user list comes first and is separated by a space followed 
       * by the group list. For e.g. "user1,user2 group1,group2"
       * 
       * @param aclString String representation of the ACL
       */
      public AccessControlList(String aclString) {
        buildACL(aclString.split(" ", 2));
      }
      
      /**
       * Construct a new ACL from String representation of users and groups
       * 
       * The arguments are comma separated lists
       * 
       * @param users comma separated list of users
       * @param groups comma separated list of groups
       */
      public AccessControlList(String users, String groups) {
        buildACL(new String[] {users, groups});
      }

### .2.3. buildACL 构建ACL ###

根据给定的字符串构建ACL . 该string包含以逗号分隔的值.

/**
       * 根据给定的字符串构建ACL . 该string包含以逗号分隔的值.
       * Build ACL from the given array of strings.
       * The strings contain comma separated values.
       *
       * @param userGroupStrings build ACL from array of Strings
       */
      private void buildACL(String[] userGroupStrings) {
        users = new HashSet<String>();
        groups = new HashSet<String>();
        for (String aclPart : userGroupStrings) {
          // 验证是否具备所有权限.
          if (aclPart != null && isWildCardACLValue(aclPart)) {
            allAllowed = true;
            break;
          }
        }
        if (!allAllowed) {
          // 验证权限
          if (userGroupStrings.length >= 1 && userGroupStrings[0] != null) {
            // 获取用户 [多个]
            users = StringUtils.getTrimmedStringCollection(userGroupStrings[0]);
          } 
          
          if (userGroupStrings.length == 2 && userGroupStrings[1] != null) {
            // 获取用户组 [多个]
            groups = StringUtils.getTrimmedStringCollection(userGroupStrings[1]);
            // 构建缓存组
            groupsMapping.cacheGroupsAdd(new LinkedList<String>(groups));
          }
        }
      }

### .2.3.4. isUserAllowed 验证访问权限 ###

根据提供的UserGroupInformation中的用户,检查该用户是否存在访问控制列表中.

public boolean isUserAllowed(UserGroupInformation ugi) {
        return isUserInList(ugi);
      }
    
      /**
       * 根据提供的UserGroupInformation中的用户,检查该用户是否存在访问控制列表中.
       * Checks if a user represented by the provided {@link UserGroupInformation}
       * is a member of the Access Control List
       *
       *             要检查ACL中是否包含的UserGroupInformation 如果ture 在访问控制列表中.
       * @param ugi  UserGroupInformation to check if contained in the ACL
       * @return true if ugi is member of the list
       */
      public final boolean isUserInList(UserGroupInformation ugi) {
        //如果全员访问,或者用户包含访问控制清单中 返回true
        if (allAllowed || users.contains(ugi.getShortUserName())) {
          return true;
        } else if (!groups.isEmpty()) {
          // 验证是否在group 中...
          for (String group : ugi.getGroups()) {
            if (groups.contains(group)) {
              // 有权限,直接返回
              return true;
            }
          }
        }
        return false;
      }

# 三. RMAppManager #

**RMAppManager负责应用程序的启动和关闭。**  
ClientRMService收到来自客户端的提交应用程序请求后， 将调用函数RMAppManager\#submitApplication创建一个RMApp对象，它将维护这个应用程序的整个生命周期， 从开始运行到最终结束 ；  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 2]

## 3.1. 属性 ##

// application内存最大存储数量
      // yarn.resourcemanager.max-completed-applications : 1000
      private int maxCompletedAppsInMemory;
      // 最大存储容量
      // yarn.resourcemanager.state-store.max-completed-applications : maxCompletedAppsInMemory [1000]
      private int maxCompletedAppsInStateStore;
    
      // 完成application存储的数量
      protected int completedAppsInStateStore = 0;
    
      // 已完成的application
      protected LinkedList<ApplicationId> completedApps = new LinkedList<>();
    
      // RM的 Context
      private final RMContext rmContext;
    
      // 负责与ApplicationMaster通讯
      private final ApplicationMasterService masterService;
      // 调度器
      private final YarnScheduler scheduler;
      // 访问控制清单
      private final ApplicationACLsManager applicationACLsManager;
      //配置
      private Configuration conf;
      // 权限相关
      private YarnAuthorizationProvider authorizer;
      // 是否启动timelineServer
      private boolean timelineServiceV2Enabled;
      // 是否启动node label
      private boolean nodeLabelsEnabled;

## 3.2. 构造方法 ##

public RMAppManager(RMContext context,
          YarnScheduler scheduler, ApplicationMasterService masterService,
          ApplicationACLsManager applicationACLsManager, Configuration conf) {
        this.rmContext = context;
        this.scheduler = scheduler;
        this.masterService = masterService;
        this.applicationACLsManager = applicationACLsManager;
        this.conf = conf;
    
        // yarn.resourcemanager.max-completed-applications : 1000
        this.maxCompletedAppsInMemory = conf.getInt(
            YarnConfiguration.RM_MAX_COMPLETED_APPLICATIONS,
            YarnConfiguration.DEFAULT_RM_MAX_COMPLETED_APPLICATIONS);
        // yarn.resourcemanager.state-store.max-completed-applications : maxCompletedAppsInMemory [1000]
        this.maxCompletedAppsInStateStore =
            conf.getInt(
              YarnConfiguration.RM_STATE_STORE_MAX_COMPLETED_APPLICATIONS,
              this.maxCompletedAppsInMemory);
    
        // 如果最大存储大于内存中的存储, 将最大存储设置为内存存储的数量
        if (this.maxCompletedAppsInStateStore > this.maxCompletedAppsInMemory) {
          this.maxCompletedAppsInStateStore = this.maxCompletedAppsInMemory;
        }
        // 构建权限
        this.authorizer = YarnAuthorizationProvider.getInstance(conf);
    
        // timelineService 是否启用 : yarn.timeline-service.versions == 2 启用
        this.timelineServiceV2Enabled = YarnConfiguration.
            timelineServiceV2Enabled(conf);
        // nodeLabel 是否启用 : yarn.node-labels.enabled : false
        this.nodeLabelsEnabled = YarnConfiguration
            .areNodeLabelsEnabled(rmContext.getYarnConfiguration());
      }

## 3.3. submitApplication ##

提交application. 构建一个RMAppImpl, 然后交由Dispatcher执行.

@SuppressWarnings("unchecked")
      protected void submitApplication(
          ApplicationSubmissionContext submissionContext, long submitTime,
          String user) throws YarnException {
        // 获取applicationId
        ApplicationId applicationId = submissionContext.getApplicationId();
    
    
        // 构建 RMAppImpl
        // 设置startTime为-1,在构造方法中会初始化
        // Passing start time as -1. It will be eventually set in RMAppImpl constructor.
        RMAppImpl application = createAndPopulateNewRMApp( submissionContext, submitTime, user, false, -1, null);
    
        try {
          // 验证权限
          if (UserGroupInformation.isSecurityEnabled()) {
            this.rmContext.getDelegationTokenRenewer()
                .addApplicationAsync(applicationId,
                    BuilderUtils.parseCredentials(submissionContext),
                    submissionContext.getCancelTokensWhenComplete(),
                    application.getUser(),
                    BuilderUtils.parseTokensConf(submissionContext));
          } else {
    
            // 处理任务 RMAppEvent 启动
            // Dispatcher这时候还没有启动,所以dispatcher应该在启动的时候处理队列中的event
    
            // Dispatcher is not yet started at this time, so these START events
            // enqueued should be guaranteed to be first processed when dispatcher
            // gets started.
            this.rmContext.getDispatcher().getEventHandler()
                .handle(new RMAppEvent(applicationId, RMAppEventType.START));
          }
        } catch (Exception e) {
          LOG.warn("Unable to parse credentials for " + applicationId, e);
          // Sending APP_REJECTED is fine, since we assume that the
          // RMApp is in NEW state and thus we haven't yet informed the
          // scheduler about the existence of the application
          this.rmContext.getDispatcher().getEventHandler()
              .handle(new RMAppEvent(applicationId,
                  RMAppEventType.APP_REJECTED, e.getMessage()));
          throw RPCUtil.getRemoteException(e);
        }
      }

## 3.4. moveApplicationAcrossQueue ##

/**
       * moveToQueue will invoke scheduler api to perform move queue operation.
       *
       * @param applicationId
       *          Application Id.
       * @param targetQueue
       *          Target queue to which this app has to be moved.
       * @throws YarnException
       *           Handle exceptions.
       */
      public void moveApplicationAcrossQueue(ApplicationId applicationId, String targetQueue)
          throws YarnException {
    
        // 获取 application
        RMApp app = this.rmContext.getRMApps().get(applicationId);
    
    
        // Capacity 调度器将会遵循以下规则
        // 1. 检查变更是正确的.
        // 2. 更新存储信息
        // 3. 执行实际操作,并且更新内存结构
    
        // Capacity scheduler will directly follow below approach.
        // 1. Do a pre-validate check to ensure that changes are fine.
        // 2. Update this information to state-store
        // 3. Perform real move operation and update in-memory data structures.
        synchronized (applicationId) {
    
          // 验证app是否为null 或者已经执行完
          if (app == null || app.isAppInCompletedStates()) {
            return;
          }
          // 获取 源 queue
          String sourceQueue = app.getQueue();
    
          // 1. 验证移动application请求是否有访问权限或者其他的错误,如果验证失败,抛出YarnException
    
          // 1. pre-validate move application request to check for any access
          // violations or other errors. If there are any violations, YarnException
          // will be thrown.
          //
          rmContext.getScheduler().preValidateMoveApplication(applicationId,
              targetQueue);
          // 2. 使用新的队列, 更新存储信息
          // 2. Update to state store with new queue and throw exception is failed.
          updateAppDataToStateStore(targetQueue, app, false);
    
          // 3. 执行真实操作 , 根据调度器的不同,执行 moveApplication 方法
          // 3. Perform the real move application
          String queue = "";
          try {
            queue = rmContext.getScheduler().moveApplication(applicationId,
                targetQueue);
          } catch (YarnException e) {
            // Revert to source queue since in-memory move has failed. Chances
            // of this is very rare as we have already done the pre-validation.
            updateAppDataToStateStore(sourceQueue, app, true);
            throw e;
          }
    
          // 更新内存信息   update in-memory
          if (queue != null && !queue.isEmpty()) {
            app.setQueue(queue);
          }
        }
    
        rmContext.getSystemMetricsPublisher().appUpdated(app,
            System.currentTimeMillis());
      }

## 3.5. finishApplication ##

当RMApp运行结束后， 将向RMAppManager发送一个RMAppManagerEventType.APP\_COMPLETED事件， 它收到该事件后将调用RMAppManager\#finishApplication进行收尾工作，  
包括：

❑将该应用程序放入已完成应用程序列表中， 以便用户查询历史应用程序运行信息。需要注意的是， 该列表的大小是有限的， 默认是1000（管理员可通过参数yarn.resourcemanager.max-completed-applications修改） ， 当已完成应用程序数目超过该值时， 将从内存数据结构中移除（移除的应用程序可称为“过期的应用程序”） ， 这样用户只能通过History Server获取过期的应用程序信息， History Server是从磁盘文件中获取这些信息的（应用程序会将运行日志和基本信息写到磁盘上） 。

❑将应用程序从RMStateStore中移除。 RMStateStore记录了运行中的应用程序的运行日志， 当集群故障重启后， ResourceManager可通过这些日志恢复应用程序运行状态， 从而避免全部重新运行， 一旦应用程序运行结束后， 这些日志便失去了意义， 故可以对其进行删除。 这属于ResourceManager容错机制的范畴

protected synchronized void finishApplication(ApplicationId applicationId) {
        if (applicationId == null) {
          LOG.error("RMAppManager received completed appId of null, skipping");
        } else {
          //验证权限
          // Inform the DelegationTokenRenewer
          if (UserGroupInformation.isSecurityEnabled()) {
            rmContext.getDelegationTokenRenewer().applicationFinished(applicationId);
          }
          // 设置状态
          completedApps.add(applicationId);
          completedAppsInStateStore++;
          writeAuditLog(applicationId);
        }
      }

# 四. ContainerAllocationExpirer #

当一个AM获得一个Container后， YARN不允许AM长时间不对其使用， 因为这会降低整个集群的利用率。 当AM收到RM新分配的一个Container后， 必须在一定的时间（默认为10min， 管理员可通过参数yarn.resourcemanager.rm.container-allocation.expiry-interval-ms修改） 内在对应的NM上启动该Container， 否则RM将强制回收该Container。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 3]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70]: /images/20221120/3cc305ab1117454387fbc17106a8c8de.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 1]: /images/20221120/7264f7dcb3d544898eb9a717421ca80b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20201215171406129.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20201215171347534.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0,size_16,color_FFFFFF,t_70