YARN SLS-AM请求资源的协议解析

在slsrunner.java中：

startAM->startAMFromSLSTrace->createAMForJob->runNewAM

在runNewAM函数中，最后两个参数很重要：

1. 
   
2. private void createAMForJob(Map jsonJob) throws YarnException {
3. ......
4. for (int i = 0; i < jobCount; i++) {
5. runNewAM(amType, user, queue, oldAppId, jobStartTime, jobFinishTime,
6. getTaskContainers(jsonJob), getAMContainerResource(jsonJob));
7. }
8. }

最后两个参数分别是getTaskContainers和getAMContainerResource，前者是获取task所需的container的需求，后者是获得AM运行的container的需求。

先看后者：

getAMContainerResource，函数返回对象是Resource类

1. 
   
2. private Resource getAMContainerResource(Map jsonJob) {
3. Resource amContainerResource =
4. SLSConfiguration.getAMContainerResource(getConf());
6. if (jsonJob == null) {
7. return amContainerResource;
8. }
10. ResourceInformation[] infors = ResourceUtils.getResourceTypesArray();
11. for (ResourceInformation info : infors) {
12. String key = SLSConfiguration.JOB_AM_PREFIX + info.getName();
13. if (jsonJob.containsKey(key)) {
14. long value = Long.parseLong(jsonJob.get(key).toString());
15. amContainerResource.setResourceValue(info.getName(), value);
16. }
17. }
19. return amContainerResource;
20. }

Resource类在org.apache.hadoop.yarn.api.records.Resource.java中实现的。

是<mem, vcores>的形式，也就是单个container的resource量。

在函数的开始，通过SLSConfiguration.getAMContainerResource(getConf())获得SLSconfiguration对于AM的resource初始量：

getAMContainerResource.getAMContainerResource:

1. 
   
2. public static Resource getAMContainerResource(Configuration conf) {
3. return Resource.newInstance(
4. conf.getLong(AM_CONTAINER_MEMORY, AM_CONTAINER_MEMORY_DEFAULT),
5. conf.getInt(AM_CONTAINER_VCORES, AM_CONTAINER_VCORES_DEFAULT));
6. }

可以看到AM的mem default是1024，vcores default是1

1. 
   
2. public static final int AM_CONTAINER_MEMORY_DEFAULT = 1024;
4. public static final String AM_CONTAINER_VCORES = AM_PREFIX +
5. "container.vcores";
6. public static final int AM_CONTAINER_VCORES_DEFAULT = 1;

这就将AM的container的需求量得到了。

然后再看getTaskContainers：

1. 
   
2. private List<ContainerSimulator> getTaskContainers(Map jsonJob)
3. throws YarnException {
4. List<ContainerSimulator> containers = new ArrayList<>();
5. List tasks = (List) jsonJob.get(SLSConfiguration.JOB_TASKS);
6. if (tasks == null || tasks.size() == 0) {
7. throw new YarnException("No task for the job!");
8. }
10. for (Object o : tasks) {
11. Map jsonTask = (Map) o;
13. String hostname = (String) jsonTask.get(SLSConfiguration.TASK_HOST);
15. long duration = 0;
16. if (jsonTask.containsKey(SLSConfiguration.TASK_DURATION_MS)) {
17. duration = Integer.parseInt(
18. jsonTask.get(SLSConfiguration.TASK_DURATION_MS).toString());
19. } else if (jsonTask.containsKey(SLSConfiguration.DURATION_MS)) {
20. // Also support "duration.ms" for backward compatibility
21. duration = Integer.parseInt(
22. jsonTask.get(SLSConfiguration.DURATION_MS).toString());
23. } else if (jsonTask.containsKey(SLSConfiguration.TASK_START_MS) &&
24. jsonTask.containsKey(SLSConfiguration.TASK_END_MS)) {
25. long taskStart = Long.parseLong(
26. jsonTask.get(SLSConfiguration.TASK_START_MS).toString());
27. long taskFinish = Long.parseLong(
28. jsonTask.get(SLSConfiguration.TASK_END_MS).toString());
29. duration = taskFinish - taskStart;
30. }
31. if (duration <= 0) {
32. throw new YarnException("Duration of a task shouldn't be less or equal"
33. + " to 0!");
34. }
36. Resource res = getResourceForContainer(jsonTask);
38. int priority = DEFAULT_MAPPER_PRIORITY;
39. if (jsonTask.containsKey(SLSConfiguration.TASK_PRIORITY)) {
40. priority = Integer.parseInt(
41. jsonTask.get(SLSConfiguration.TASK_PRIORITY).toString());
42. }
44. String type = "map";
45. if (jsonTask.containsKey(SLSConfiguration.TASK_TYPE)) {
46. type = jsonTask.get(SLSConfiguration.TASK_TYPE).toString();
47. }
49. int count = 1;
50. if (jsonTask.containsKey(SLSConfiguration.COUNT)) {
51. count = Integer.parseInt(
52. jsonTask.get(SLSConfiguration.COUNT).toString());
53. }
54. count = Math.max(count, 1);
56. ExecutionType executionType = ExecutionType.GUARANTEED;
57. if (jsonTask.containsKey(SLSConfiguration.TASK_EXECUTION_TYPE)) {
58. executionType = ExecutionType.valueOf(
59. jsonTask.get(SLSConfiguration.TASK_EXECUTION_TYPE).toString());
60. }
61. for (int i = 0; i < count; i++) {
62. containers.add(
63. new ContainerSimulator(res, duration, hostname, priority, type,
64. executionType));
65. }
66. }
68. return containers;
69. }

函数返回的对象是List<ContainerSimulator>

该类的实现在org.apache.hadoop.yarn.sls.scheduler.ContainerSimulator:

1. 
   
2. public class ContainerSimulator implements Delayed {
3. // id
4. private ContainerId id;
5. // resource allocated
6. private Resource resource;
7. // end time
8. private long endTime;
9. // life time (ms)
10. private long lifeTime;
11. // host name
12. private String hostname;
13. // priority
14. private int priority;
15. // type
16. private String type;
17. // execution type
18. private ExecutionType executionType = ExecutionType.GUARANTEED;

可以看到包括了Resource类，还有一些其他属性，比如id，endtime，hostname，priority等。

也就是说给task分配的container的属性，要比给AM分配的container属性要多。

回到getTaskContainers，对了循环了sls-jobs.json中每个task，然后获得task在-jobs.json中描述的属性，特别关注下resource的获得：

Resource res = getResourceForContainer(jsonTask);

1. 
   
2. private Resource getResourceForContainer(Map jsonTask) {
3. Resource res = getDefaultContainerResource();
4. ResourceInformation[] infors = ResourceUtils.getResourceTypesArray();
5. for (ResourceInformation info : infors) {
6. if (jsonTask.containsKey(SLSConfiguration.TASK_PREFIX + info.getName())) {
7. long value = Long.parseLong(
8. jsonTask.get(SLSConfiguration.TASK_PREFIX + info.getName())
9. .toString());
10. res.setResourceValue(info.getName(), value);
11. }
12. }
14. return res;
15. }

可以看到是通过getDefaultContainerResource获得的，看这个函数的实现：

getDefaultContainerResource：

1. 
   
2. private Resource getDefaultContainerResource() {
3. int containerMemory = getConf().getInt(SLSConfiguration.CONTAINER_MEMORY_MB,
4. SLSConfiguration.CONTAINER_MEMORY_MB_DEFAULT);
5. int containerVCores = getConf().getInt(SLSConfiguration.CONTAINER_VCORES,
6. SLSConfiguration.CONTAINER_VCORES_DEFAULT);
7. return Resources.createResource(containerMemory, containerVCores);
8. }

可以看到不出所料，还是获得SLSConfiguration中的default mem：1024和default vcore：1

现在可以总结了：AM的container和task的container是分开分配的。AM的container分配的资源类型是resource：<mem, core>，而task的container分配的资源类型是ContainerSimulator，resource是其中的一个成员，还有其他的成员，比如hostname，id等。

再回到最初的runNewAM：

将AM获得的container资源和task获得的container资源作为参数传入后。

1. 
   
2. private void runNewAM(String jobType, String user,
3. String jobQueue, String oldJobId, long jobStartTimeMS,
4. long jobFinishTimeMS, List<ContainerSimulator> containerList,
5. Resource amContainerResource) {
6. runNewAM(jobType, user, jobQueue, oldJobId, jobStartTimeMS,
7. jobFinishTimeMS, containerList, null, -1,
8. amContainerResource, null);
9. }

1. 
   
2. private void runNewAM(String jobType, String user,
3. String jobQueue, String oldJobId, long jobStartTimeMS,
4. long jobFinishTimeMS, List<ContainerSimulator> containerList,
5. ReservationId reservationId, long deadline, Resource amContainerResource,
6. Map<String, String> params) {
8. AMSimulator amSim = (AMSimulator) ReflectionUtils.newInstance(
9. amClassMap.get(jobType), new Configuration());
11. if (amSim != null) {
12. int heartbeatInterval = getConf().getInt(
13. SLSConfiguration.AM_HEARTBEAT_INTERVAL_MS,
14. SLSConfiguration.AM_HEARTBEAT_INTERVAL_MS_DEFAULT);
15. boolean isTracked = trackedApps.contains(oldJobId);
17. if (oldJobId == null) {
18. oldJobId = Integer.toString(AM_ID);
19. }
20. AM_ID++;
21. amSim.init(heartbeatInterval, containerList, rm, this, jobStartTimeMS,
22. jobFinishTimeMS, user, jobQueue, isTracked, oldJobId,
23. runner.getStartTimeMS(), amContainerResource, params);
24. if(reservationId != null) {
25. // if we have a ReservationId, delegate reservation creation to
26. // AMSim (reservation shape is impl specific)
27. UTCClock clock = new UTCClock();
28. amSim.initReservation(reservationId, deadline, clock.getTime());
29. }
30. runner.schedule(amSim);
31. maxRuntime = Math.max(maxRuntime, jobFinishTimeMS);
32. numTasks += containerList.size();
33. amMap.put(oldJobId, amSim);
34. }
35. }

可以看到作为amSim.init进行初始化了。

amSim.init：

1. 
   
2. public void init(int heartbeatInterval,
3. List<ContainerSimulator> containerList, ResourceManager resourceManager,
4. SLSRunner slsRunnner, long startTime, long finishTime, String simUser,
5. String simQueue, boolean tracked, String oldApp, long baseTimeMS,
6. Resource amResource, Map<String, String> params) {
7. super.init(startTime, startTime + 1000000L * heartbeatInterval,
8. heartbeatInterval);
9. this.user = simUser;
10. this.rm = resourceManager;
11. this.se = slsRunnner;
12. this.queue = simQueue;
13. this.oldAppId = oldApp;
14. this.isTracked = tracked;
15. this.baselineTimeMS = baseTimeMS;
16. this.traceStartTimeMS = startTime;
17. this.traceFinishTimeMS = finishTime;
18. this.amContainerResource = amResource;
19. }

可以看到，amResource赋给了this.amContainerResource。作为AM的一个成员了。

接下来看看this.amContainerResource在哪被调用了：

1. 
   
2. private void submitApp(ReservationId reservationId)
3. throws YarnException, InterruptedException, IOException {
4. // ask for new application
5. GetNewApplicationRequest newAppRequest =
6. Records.newRecord(GetNewApplicationRequest.class);
7. GetNewApplicationResponse newAppResponse =
8. rm.getClientRMService().getNewApplication(newAppRequest);
9. appId = newAppResponse.getApplicationId();
11. // submit the application
12. final SubmitApplicationRequest subAppRequest =
13. Records.newRecord(SubmitApplicationRequest.class);
14. ApplicationSubmissionContext appSubContext =
15. Records.newRecord(ApplicationSubmissionContext.class);
16. appSubContext.setApplicationId(appId);
17. appSubContext.setMaxAppAttempts(1);
18. appSubContext.setQueue(queue);
19. appSubContext.setPriority(Priority.newInstance(0));
20. ContainerLaunchContext conLauContext =
21. Records.newRecord(ContainerLaunchContext.class);
22. conLauContext.setApplicationACLs(new HashMap<>());
23. conLauContext.setCommands(new ArrayList<>());
24. conLauContext.setEnvironment(new HashMap<>());
25. conLauContext.setLocalResources(new HashMap<>());
26. conLauContext.setServiceData(new HashMap<>());
27. appSubContext.setAMContainerSpec(conLauContext);
28. appSubContext.setResource(amContainerResource);

可以看到，在submitApp中，通过appSubContext.setResource(amContainerResource)，amContainerResource作为appSubContext的一部分。

关于appSubContext结构的分析，见另一篇博文：YARN-SLS AM，RM协议解析

ApplicationSubmissionContext的协议结构如下：

1. 
   
2. message ApplicationSubmissionContextProto {
3. optional ApplicationIdProto application_id = 1;
4. optional string application_name = 2 [default = "N/A"];
5. optional string queue = 3 [default = "default"];
6. optional PriorityProto priority = 4;
7. optional ContainerLaunchContextProto am_container_spec = 5;
8. optional bool cancel_tokens_when_complete = 6 [default = true];
9. optional bool unmanaged_am = 7 [default = false];
10. optional int32 maxAppAttempts = 8 [default = 0];
11. optional ResourceProto resource = 9;
12. optional string applicationType = 10 [default = "YARN"];
13. optional bool keep_containers_across_application_attempts = 11 [default = false];
14. repeated string applicationTags = 12;
15. optional int64 attempt_failures_validity_interval = 13 [default = -1];
16. optional LogAggregationContextProto log_aggregation_context = 14;
17. optional ReservationIdProto reservation_id = 15;
18. optional string node_label_expression = 16;
19. repeated ResourceRequestProto am_container_resource_request = 17;
20. repeated ApplicationTimeoutMapProto application_timeouts = 18;
21. repeated StringStringMapProto application_scheduling_properties = 19;
22. }

其中：optional ResourceProto resource就是通过appSubContext.setResource(amContainerResource)赋值过去的。

然后将appSubContext作为subAppRequest的一部分，最后通过：

rm.getClientRMService().submitApplication(subAppRequest);传递给RM。

以上就是AM的container请求传递给RM的过程。

那么AM中task的container请求传递给RM是什么样的呢？

在runNewAM中，ask的container请求的形式是List<ContainerSimulator> containerList,传入到amSim.init，与上面相同，但奇怪的是在amSim.init中，并未将containerList参数传给任何对象。

1. 
   
2. public void init(int heartbeatInterval,
3. List<ContainerSimulator> containerList, ResourceManager resourceManager,
4. SLSRunner slsRunnner, long startTime, long finishTime, String simUser,
5. String simQueue, boolean tracked, String oldApp, long baseTimeMS,
6. Resource amResource, Map<String, String> params) {
7. super.init(startTime, startTime + 1000000L * heartbeatInterval,
8. heartbeatInterval);
9. this.user = simUser;
10. this.rm = resourceManager;
11. this.se = slsRunnner;
12. this.queue = simQueue;
13. this.oldAppId = oldApp;
14. this.isTracked = tracked;
15. this.baselineTimeMS = baseTimeMS;
16. this.traceStartTimeMS = startTime;
17. this.traceFinishTimeMS = finishTime;
18. this.amContainerResource = amResource;
19. }

可以看到参数List<ContainerSimulator> containerList并没有传给其他对象。

在middlestep的sendContainerRequest中，调用了packageRequests方法来打包task的container request，应该是在某个地方进行了参数的传递。

转到sendContainerRequest：

1. 
   
2. @Override
3. protected void sendContainerRequest()
4. throws YarnException, IOException, InterruptedException {
5. if (isFinished) {
6. return;
7. }
9. // send out request
10. List<ResourceRequest> ask = null;
11. if (mapFinished != mapTotal) {
12. // map phase
13. if (!pendingMaps.isEmpty()) {
14. ask = packageRequests(mergeLists(pendingMaps, scheduledMaps),
15. PRIORITY_MAP);
16. LOG.debug("Application {} sends out request for {} mappers.",

看到packageRequests的传入参数是mergeLists(pendingMaps, scheduledMaps)。

1. 
   
2. protected List<ResourceRequest> packageRequests(
3. List<ContainerSimulator> csList, int priority) {

对比可发现，mergeLists(pendingMaps, scheduledMaps)就是List<ContainerSimulator>类型，猜测也就是List<ContainerSimulator> containerList。

在packageRequests中的一些结构变化和传递见博文：YARN-SLS AM，RM协议解析

1. 
   
2. protected List<ResourceRequest> packageRequests(
3. List<ContainerSimulator> csList, int priority) {
4. // create requests
5. Map<String, ResourceRequest> rackLocalRequestMap = new HashMap<String, ResourceRequest>();
6. Map<String, ResourceRequest> nodeLocalRequestMap = new HashMap<String, ResourceRequest>();
7. ResourceRequest anyRequest = null;
8. for (ContainerSimulator cs : csList) {
9. if (cs.getHostname() != null) {
10. String[] rackHostNames = SLSUtils.getRackHostName(cs.getHostname());
11. // check rack local
12. String rackname = "/" + rackHostNames[0];
13. if (rackLocalRequestMap.containsKey(rackname)) {
14. rackLocalRequestMap.get(rackname).setNumContainers(
15. rackLocalRequestMap.get(rackname).getNumContainers() + 1);
16. } else {
17. ResourceRequest request = createResourceRequest(cs.getResource(),
18. cs.getExecutionType(), rackname, priority, 1);
19. rackLocalRequestMap.put(rackname, request);
20. }
21. // check node local
22. String hostname = rackHostNames[1];
23. if (nodeLocalRequestMap.containsKey(hostname)) {
24. nodeLocalRequestMap.get(hostname).setNumContainers(
25. nodeLocalRequestMap.get(hostname).getNumContainers() + 1);
26. } else {
27. ResourceRequest request = createResourceRequest(cs.getResource(),
28. cs.getExecutionType(), hostname, priority, 1);
29. nodeLocalRequestMap.put(hostname, request);
30. }
31. }
32. // any
33. if (anyRequest == null) {
34. anyRequest = createResourceRequest(cs.getResource(),
35. cs.getExecutionType(), ResourceRequest.ANY, priority, 1);
36. } else {
37. anyRequest.setNumContainers(anyRequest.getNumContainers() + 1);
38. }
39. }
40. List<ResourceRequest> ask = new ArrayList<ResourceRequest>();
41. ask.addAll(nodeLocalRequestMap.values());
42. ask.addAll(rackLocalRequestMap.values());
43. if (anyRequest != null) {
44. ask.add(anyRequest);
45. }
46. return ask;
47. }

在packageRequests中，通过

1. 
   
2. ResourceRequest request = createResourceRequest(cs.getResource(),
3. cs.getExecutionType(), hostname, priority, 1);

将List<ContainerSimulator> containerList的资源（resource）以及其他一些资源特性封装为ResourceRequest格式的request。

看下该函数：

createResourceRequest：

1. 
   
2. protected ResourceRequest createResourceRequest(Resource resource,
3. ExecutionType executionType, String host, int priority, int
4. numContainers) {
5. ResourceRequest request = recordFactory
6. .newRecordInstance(ResourceRequest.class);
7. request.setCapability(resource);
8. request.setResourceName(host);
9. request.setNumContainers(numContainers);
10. request.setExecutionTypeRequest(
11. ExecutionTypeRequest.newInstance(executionType));
12. Priority prio = recordFactory.newRecordInstance(Priority.class);
13. prio.setPriority(priority);
14. request.setPriority(prio);
15. return request;
16. }

该函数就是将这些分散的资源（container resource，host，pri，numcontainer）封装成ResourceRequest类。

其中通过request.setCapability(resource); 将List<ContainerSimulator> containerList的资源(<mem, vcores>)封装进去

看下ResourceRequest的格式：

1. 
   
2. message ResourceRequestProto {
3. optional PriorityProto priority = 1;
4. optional string resource_name = 2;
5. optional ResourceProto capability = 3;
6. optional int32 num_containers = 4;
7. optional bool relax_locality = 5 [default = true];
8. optional string node_label_expression = 6;
9. optional ExecutionTypeRequestProto execution_type_request = 7;
10. optional int64 allocation_request_id = 8 [default = -1];
11. }

其中ResourceProto capability就是List<ContainerSimulator> containerList的资源（resource）。

然后回到sendContainerRequest函数：

ask = packageRequests(mergeLists(pendingMaps, scheduledMaps),

PRIORITY_MAP);

ask就是List<ResourceRequest> 格式。sendContainerRequest接着往下走：

1. 
   
3. if (ask == null) {
4. ask = new ArrayList<>();
5. }
7. final AllocateRequest request = createAllocateRequest(ask);
8. if (totalContainers == 0) {
9. request.setProgress(1.0f);
10. } else {
11. request.setProgress((float) finishedContainers / totalContainers);
12. }
14. UserGroupInformation ugi =
15. UserGroupInformation.createRemoteUser(appAttemptId.toString());
16. Token<AMRMTokenIdentifier> token = rm.getRMContext().getRMApps()
17. .get(appAttemptId.getApplicationId())
18. .getRMAppAttempt(appAttemptId).getAMRMToken();
19. ugi.addTokenIdentifier(token.decodeIdentifier());
20. AllocateResponse response = ugi.doAs(
21. new PrivilegedExceptionAction<AllocateResponse>() {
22. @Override
23. public AllocateResponse run() throws Exception {
24. return rm.getApplicationMasterService().allocate(request);
25. }
26. });
27. if (response != null) {
28. responseQueue.put(response);
29. }

可以看到，通过createAllocateRequest(ask)函数将ask（List<ResourceRequest> ）转换为request（AllocateRequest）。

1. 
   
2. protected AllocateRequest createAllocateRequest(List<ResourceRequest> ask,
3. List<ContainerId> toRelease) {
4. AllocateRequest allocateRequest =
5. recordFactory.newRecordInstance(AllocateRequest.class);
6. allocateRequest.setResponseId(responseId++);
7. allocateRequest.setAskList(ask);
8. allocateRequest.setReleaseList(toRelease);
9. return allocateRequest;
10. }

ask被最终包装成AllocateRequest类型的request，AllocateRequest就是AM向RM申请资源最终的格式。在真正的YARN中，AM与RM通信也是AllocateRequest格式。

以上就是AM向RM请求资源的协议解析。包括AM自身的container以及AM的task的container。