Yarn資源隔離

YARN · 發表 2018-10-16 21:49:10

摘要： 1 概述 1.1 基本概念 ** A.ResourceManager ** ResourceManager 是一個全域性的資源管理器，負責整個叢集的資源管理和分配。它主要由兩個元件構成：排程器(Scheduler)和應用程式管理器(Applications Manager，AppM...

1 概述

1.1 基本概念

** A.ResourceManager **

ResourceManager 是一個全域性的資源管理器，負責整個叢集的資源管理和分配。它主要由兩個元件構成：排程器(Scheduler)和應用程式管理器(Applications Manager，AppManager)。這裡指進行基本介紹，部分模組單獨進行深入。

排程器

該排程器是一個 "純排程器"，不再參與任何與具體應用程式邏輯相關的工作，而僅根據各個應用程式的資源需求進行分配，資源分配的單位用一個資源抽象概念 "Container" 來表示。Container 封裝了記憶體和 CPU。此外，排程器是一個可插拔的元件，使用者可根據自己的需求設計新的排程器，YARN 自身提供了 Fair Scheduler 和 Capacity Scheduler。

應用程式管理器

應用程式管理器負責管理整個系統中所有應用程式，包括應用程式的提交、與排程器協商資源以啟動 ApplicationMaster、監控 ApplicationMaster 執行狀態並在失敗時重新啟動它等。

** B.ApplicationMaster **

使用者提交的每個 Application 都要包含一個 ApplicatioNMaster，主要功能包括：

向 RM 排程器申請資源(用 Container 表示)
將從 RM 分配的資源分配給 Applcation 內部的任務
與 NM 通訊請求啟動/停止任務
監控所有任務的執行狀態，並在失敗時重新為任務申請資源以重啟任務

** C.NodeManager **

NM 是每個節點上的資源和工作管理員，一方面，它會定時地向 RM 彙報本節點上的資源使用情況和各個 Container 的執行狀態;另一方面，它接收並處理來自 AM 的 Container 啟動/停止等各種命令。

** D.Container **

Container 是 YARN 中資源抽象，它封裝了某個節點上的記憶體和 CPU，當 AM 向 RM 申請資源時，RM 為 AM 返回的資源便是用 Container 表示的。YARN 是使用輕量級資源隔離機制 Cgroups 進行資源隔離的。

1.2 YARN通訊協議

在 YARN 中，任何兩個需要相互通訊的元件之間僅有一個RPC協議，而對於任何一個RPC協議，通訊雙方有一端是Client，另一端是Server，且Client總是主動連線Server。YARN 中有以下幾個主要的RPC協議：

1.JobClient與RM之間的協議：ApplicationClientProtocol，JobClient 通過該 RPC 協議提交應用程式、查詢應用程式狀態等
2.Admin(管理員)與RM之間的協議：ResourceManagerAdministrationProtocol，Admin通過該RPC協議更新系統配置檔案，比如節點黑白名單、使用者佇列、許可權等
3.AM與RM之間的協議：ApplicationMasterProtocol，AM 通過該RPC協議向RM註冊並撤銷自己，併為各個人物申請資源
4.NM與RM之間的協議：ResourceTracker，NM通過該協議向RM註冊，並定時傳送心跳資訊彙報當前節點的資源使用情況和Container執行情況，並接收來自AM的命令
5.AM與NM 之間的協議：ContainerManagermentProtocol，AM通過該RPC協議要求NM啟動或者停止Container

Yarn工作流程如下：

yarn工作流程.png

說明：

1.Client 向 YARN 提交應用程式，其中包括 ApplicationMaster 程式及啟動 ApplicationMaster 的命令
2.ResourceManager 為該 ApplicationMaster 分配第一個 Container，並與對應的 NodeManager 通訊，要求它在這個 Container 中啟動應用程式的 ApplicationMaster
3.ApplicationMaster 首先向 ResourceManager 註冊
4.ApplicationMaster 為 Application 的任務申請並領取資源
5.領取到資源後，要求對應的 NodeManager 在 Container 中啟動任務
6.NodeManager收到ApplicationMaster的請求後，為任務設定好執行環境(包括環境變數、JAR 包、二進位制程式等)，將任務啟動指令碼寫到一個指令碼中，並通過執行該指令碼啟動任務
7.各個任務通過 RPC 協議向ApplicationMaster彙報自己的狀態和進度，以讓ApplicationMaster隨時掌握各個任務的執行狀態，從而可以在失敗時重啟任務
8.應用程式完成後，ApplicationMaster向ResourceManager登出並關閉自己

2 關鍵程式碼流程

客戶端與RM進行交換（Client<-->ResourceManager），首先需要建立一個YarnClient

YarnClient yarnClient = YarnClient.createYarnClient();
 yarnClient.init(conf);
 yarnClient.start();

完整如下：

package com.hortonworks.simpleyarnapp;

import java.io.File;
import java.io.IOException;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.yarn.api.ApplicationConstants;
import org.apache.hadoop.yarn.api.ApplicationConstants.Environment;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.ApplicationReport;
import org.apache.hadoop.yarn.api.records.ApplicationSubmissionContext;
import org.apache.hadoop.yarn.api.records.ContainerLaunchContext;
import org.apache.hadoop.yarn.api.records.LocalResource;
import org.apache.hadoop.yarn.api.records.LocalResourceType;
import org.apache.hadoop.yarn.api.records.LocalResourceVisibility;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.api.records.YarnApplicationState;
import org.apache.hadoop.yarn.client.api.YarnClient;
import org.apache.hadoop.yarn.client.api.YarnClientApplication;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.util.Apps;
import org.apache.hadoop.yarn.util.ConverterUtils;
import org.apache.hadoop.yarn.util.Records;


public class Client {

Configuration conf = new YarnConfiguration();

public void run(String[] args) throws Exception {
final String command = args[0];
final int n = Integer.valueOf(args[1]);
final Path jarPath = new Path(args[2]);

// Create yarnClient
YarnConfiguration conf = new YarnConfiguration();
YarnClient yarnClient = YarnClient.createYarnClient();
yarnClient.init(conf);
yarnClient.start();

// Create application via yarnClient
YarnClientApplication app = yarnClient.createApplication();

// Set up the container launch context for the application master
ContainerLaunchContext amContainer = 
Records.newRecord(ContainerLaunchContext.class);
amContainer.setCommands(
Collections.singletonList(
"$JAVA_HOME/bin/java" +
" -Xmx256M" +
" com.hortonworks.simpleyarnapp.ApplicationMaster" +
" " + command +
" " + String.valueOf(n) +
" 1>" + ApplicationConstants.LOG_DIR_EXPANSION_VAR + "/stdout" + 
" 2>" + ApplicationConstants.LOG_DIR_EXPANSION_VAR + "/stderr" 
)
);

// Setup jar for ApplicationMaster
LocalResource appMasterJar = Records.newRecord(LocalResource.class);
setupAppMasterJar(jarPath, appMasterJar);
amContainer.setLocalResources(
Collections.singletonMap("simpleapp.jar", appMasterJar));

// Setup CLASSPATH for ApplicationMaster
Map<String, String> appMasterEnv = new HashMap<String, String>();
setupAppMasterEnv(appMasterEnv);
amContainer.setEnvironment(appMasterEnv);

// Set up resource type requirements for ApplicationMaster
Resource capability = Records.newRecord(Resource.class);
capability.setMemory(256);
capability.setVirtualCores(1);

// Finally, set-up ApplicationSubmissionContext for the application
ApplicationSubmissionContext appContext = 
app.getApplicationSubmissionContext();
appContext.setApplicationName("simple-yarn-app"); // application name
appContext.setAMContainerSpec(amContainer);
appContext.setResource(capability);
appContext.setQueue("default"); // queue 

// Submit application
ApplicationId appId = appContext.getApplicationId();
System.out.println("Submitting application " + appId);
yarnClient.submitApplication(appContext);

ApplicationReport appReport = yarnClient.getApplicationReport(appId);
YarnApplicationState appState = appReport.getYarnApplicationState();
while (appState != YarnApplicationState.FINISHED && 
appState != YarnApplicationState.KILLED && 
appState != YarnApplicationState.FAILED) {
Thread.sleep(100);
appReport = yarnClient.getApplicationReport(appId);
appState = appReport.getYarnApplicationState();
}

System.out.println(
"Application " + appId + " finished with" +
" state " + appState + 
" at " + appReport.getFinishTime());

}

private void setupAppMasterJar(Path jarPath, LocalResource appMasterJar) throws IOException {
FileStatus jarStat = FileSystem.get(conf).getFileStatus(jarPath);
appMasterJar.setResource(ConverterUtils.getYarnUrlFromPath(jarPath));
appMasterJar.setSize(jarStat.getLen());
appMasterJar.setTimestamp(jarStat.getModificationTime());
appMasterJar.setType(LocalResourceType.FILE);
appMasterJar.setVisibility(LocalResourceVisibility.PUBLIC);
}

private void setupAppMasterEnv(Map<String, String> appMasterEnv) {
for (String c : conf.getStrings(
YarnConfiguration.YARN_APPLICATION_CLASSPATH,
YarnConfiguration.DEFAULT_YARN_APPLICATION_CLASSPATH)) {
Apps.addToEnvironment(appMasterEnv, Environment.CLASSPATH.name(),
c.trim());
}
Apps.addToEnvironment(appMasterEnv,
Environment.CLASSPATH.name(),
Environment.PWD.$() + File.separator + "*");
}

public static void main(String[] args) throws Exception {
Client c = new Client();
c.run(args);
}
}

ApplicatioNMaster與ResourceManager的交換（ApplicationMater<-->ResourceManager），需要建立AMRMClientAsync/AMRMClient,對應處理事件為AMRMClientAsync.CallbackHandler/AMRMClient.CallbackHandler

AMRMClient<ContainerRequest> rmClient = AMRMClient.createAMRMClient();
rmClient.init(conf);
rmClient.start();

ApplicationMater在啟動任務執行時的container，需要直接與NodeManager互動（ApplicationMaster<-->NodeManager），需要建立NMClientAsync/NMClient物件，處理事件通過NMClientAsync.CallbackHandler/NMClient.CallbackHandler

完整原始碼如下：

package com.hortonworks.simpleyarnapp;

import java.util.Collections;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.yarn.api.ApplicationConstants;
import org.apache.hadoop.yarn.api.protocolrecords.AllocateResponse;
import org.apache.hadoop.yarn.api.records.Container;
import org.apache.hadoop.yarn.api.records.ContainerLaunchContext;
import org.apache.hadoop.yarn.api.records.ContainerStatus;
import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
import org.apache.hadoop.yarn.api.records.Priority;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.client.api.AMRMClient;
import org.apache.hadoop.yarn.client.api.AMRMClient.ContainerRequest;
import org.apache.hadoop.yarn.client.api.NMClient;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.util.Records;

public class ApplicationMaster {

public static void main(String[] args) throws Exception {

final String command = args[0];
final int n = Integer.valueOf(args[1]);

// Initialize clients to ResourceManager and NodeManagers
Configuration conf = new YarnConfiguration();

AMRMClient<ContainerRequest> rmClient = AMRMClient.createAMRMClient();
rmClient.init(conf);
rmClient.start();

NMClient nmClient = NMClient.createNMClient();
nmClient.init(conf);
nmClient.start();

// Register with ResourceManager
System.out.println("registerApplicationMaster 0");
rmClient.registerApplicationMaster("", 0, "");
System.out.println("registerApplicationMaster 1");

// Priority for worker containers - priorities are intra-application
Priority priority = Records.newRecord(Priority.class);
priority.setPriority(0);

// Resource requirements for worker containers
Resource capability = Records.newRecord(Resource.class);
capability.setMemory(128);
capability.setVirtualCores(1);

// Make container requests to ResourceManager
for (int i = 0; i < n; ++i) {
ContainerRequest containerAsk = new ContainerRequest(capability, null, null, priority);
System.out.println("Making res-req " + i);
rmClient.addContainerRequest(containerAsk);
}

// Obtain allocated containers, launch and check for responses
int responseId = 0;
int completedContainers = 0;
while (completedContainers < n) {
AllocateResponse response = rmClient.allocate(responseId++);
for (Container container : response.getAllocatedContainers()) {
// Launch container by create ContainerLaunchContext
ContainerLaunchContext ctx =
Records.newRecord(ContainerLaunchContext.class);
ctx.setCommands(
Collections.singletonList(
//實際執行的內容
command +
" 1>" + ApplicationConstants.LOG_DIR_EXPANSION_VAR + "/stdout" +
" 2>" + ApplicationConstants.LOG_DIR_EXPANSION_VAR + "/stderr"
));
System.out.println("Launching container " + container.getId());
nmClient.startContainer(container, ctx);
}
for (ContainerStatus status : response.getCompletedContainersStatuses()) {
++completedContainers;
System.out.println("Completed container " + status.getContainerId());
}
Thread.sleep(100);
}

// Un-register with ResourceManager
rmClient.unregisterApplicationMaster(
FinalApplicationStatus.SUCCEEDED, "", "");
}
}

3 Container實現

當NodeManager收到AM的啟動Container事件時，執行ContainerLauncher，然後在其中執行具體的ContainerExecutor（DefaultContainerExecutor/LinuxContainerExecutor），針對Container中資源監控時傳入了Executor物件獲取相關PID。

public ContainersMonitorImpl(ContainerExecutor exec,
AsyncDispatcher dispatcher, Context context) {
super("containers-monitor");

this.containerExecutor = exec;
this.eventDispatcher = dispatcher;
this.context = context;

this.containersToBeAdded = new HashMap<ContainerId, ProcessTreeInfo>();
this.containersToBeRemoved = new ArrayList<ContainerId>();
this.monitoringThread = new MonitoringThread();
}

3.1 資源隔離

預設情況下，YARN採用了執行緒監控的方法判斷任務是否超量使用記憶體，一旦發生超量，則直接將其殺死。由於Cgroups對記憶體的控制缺乏靈活性（即任務任何時刻不能超過記憶體上限，如果超過，則直接將其殺死或者報OOM）,而Java程序在建立瞬間記憶體將翻倍，之後驟降到正常值，這種情況下，採用執行緒監控的方式更加靈活（當發現程序樹記憶體瞬間翻倍超過設定值時，可認為是正常現象，不會將任務殺死）。

構造程序樹

在YARN中每個Container可以認為是一個獨立的程序，同時，Container也可能會建立子程序（可能會建立多個子程序，這些子程序可能也會建立子程序），因此Container程序的記憶體（實體記憶體、虛擬記憶體）使用量應該表示為：以Container程序為根的程序樹中所有程序的記憶體（實體記憶體、虛擬記憶體）使用總量。

在Linux系統中的 /proc目錄下，有大量以整數命名的目錄，這些整數是某個正在執行的程序的PID，如下圖所示：

程序資訊.png

而目錄/proc/<PID>下面的那些檔案分別表示著程序執行時的各方面資訊

PID內容.png

YARN只關心/proc/<PID>/stat檔案。

開啟一個/proc/<PID>/stat檔案你會發現它僅有一行（多列）文字：

60717 (rpc.statd) S 1 60717 60717 0 -1 4202816 1262 0 0 0 3 0 0 0 20 0 1 0 412627997 75718656 12987 18446744073709551615 139780868575232 139780868643356 140737412651392 140737412650712 139780856985091 0 0 16846848 18947 18446744071580587289 0 0 17 6 0 0 0 0 0

裡面記錄了該程序相關的資訊，可以通過正則表示式

^([\\d-]+)\\s\\(([^)]+)\\)\\s[^\\s]\\s([\\d-]+)\\s([\\d-]+)\\s+([\\d-]+)\\s([\\d-]+\\s){7}(\\d+)\\s(\\d+)\\s([\\d-]+\\s){7}(\\d+)\\s +(\\d+)(\\s[\\d-]+){15}

從中抽取程序的執行時資訊，包括：程序名稱、父程序PID、父程序使用者組ID、Session ID在使用者態執行的時間（單位：jiffies）、核心態執行的時間（單位：jiffies）、佔用虛擬記憶體大小（單位：page）和佔用實體記憶體大小（單位：page）等。

檔案/proc/<PID>/stat中包含的記憶體大小單位為page，為了獲取以位元組為單位的記憶體資訊，可通過執行以下Shell指令碼獲取每個page對應的記憶體量（單位：B）：

[root@lark001 60717]# getconfPAGESIZE
4096

通過對以上資訊的計算可以得到當前每個執行的Container使用的記憶體總量。在ContainersMonitorImpl內部維護著每個Container程序的PID，通過遍歷/proc下各個程序的stat檔案內容（父程序PID、佔用虛擬記憶體大小和佔用實體記憶體大小），YARN構建出每個Container的程序樹，從而得出每個程序樹的虛擬記憶體、實體記憶體使用總量。後期針對該資訊會定時重新整理。

判斷記憶體

在獲取到每一個Container之後已經可以獲得各個Container程序樹的記憶體（實體記憶體、虛擬記憶體）使用量，YARN並不會僅憑程序樹的記憶體使用量（實體記憶體、虛擬記憶體）是否超過上限值就決定是否“殺死”一個Container，因為“子程序”的記憶體使用量是有“波動”的(Java中建立子程序採用了fork()+exec()的方案，子程序啟動瞬間，它使用的記憶體量與父程序一致，從外面看來，一個程序使用記憶體量可能瞬間翻倍，然後又降下來)，為了避免“誤殺”的情況出現，Hadoop賦予每個程序“年齡”屬性，並規定剛啟動程序的年齡是1，MonitoringThread執行緒每更新一次，各個程序的年齡加一，在此基礎上，選擇被“殺死”的Container的標準如下：如果一個Contaier對應的程序樹中所有程序（年齡大於0）總記憶體（實體記憶體或虛擬記憶體）使用量超過上限值的兩倍；或者所有年齡大於1的程序總記憶體（實體記憶體或虛擬記憶體）使用量超過上限值，則認為該Container使用記憶體超量，則Container傳送ContainerEventType.KILL_CONTAINER事件將其KILL掉。

CPU資源隔離

預設情況下，NodeManager未啟用任何對CPU資源的隔離機制，如果需要啟用該機制需使用LinuxContainerExecutor，它能夠以應用程式提交者的身份建立檔案、執行Container和銷燬Container。相比於DefaultContainerExecutor採用NodeManager啟動者的身份執行這些操作，LinuxContainerExecutor具有更高的安全性。

LinuxContainerExecutor的核心設計思想是賦予NodeManager啟動者以root許可權，進而使其擁有足夠的許可權以任意使用者身份執行一些操作，從而使得NodeManager執行者可以將Container使用的目錄和檔案的擁有者修改為應用程式的提交者，並以應用程式提交者的身份執行Container，防止所有Container以NodeManager執行者身份執行進而帶來的各種風險。

上述機制的實現在YARN的NodeManager採用C語言實現了一個setuid功能的工具container-executor，其位於如下目錄：

nm監控指令碼.png

該指令碼擁有root許可權，可以完成任意操作，其可執行指令碼位於：/opt/yarn/hadoop/bin/container-executor。YARN正是通過該指令碼創建出Cgroups層級樹以及完成Cgroups屬性設定等操作。具體cgroups詳細講解參考引用部分文章。

References

1. ofollow,noindex">Hadoop: Writing YARN Applications
2. Yarn Application Demo
3. Simple YARN application
4. YARN ApplicationMaster與ResourceManager之間基於applicationmaster_protocol.proto協議的allocate()介面原始碼解析
5. Docker容器使用cgroups限制資源使用