【Hadoop】【Yarn】Yarn的web 接口实现

Yarn中为用户提供了是三中类型的交互模式：
1-命令行
2-客户端API
3-WebService

其中命令行是基于客户端API实现的。两者都是基于RPC的。而WebService是基于HTTP协议。
本文将详解介绍Yarn中提供的WebService接口。
在hadoop代码中搜索“@Path”.得到的路径就是所有的hadoop WebService接口。
基本分为下面几类：

Yarn & JHS
HsWebServices：jhs webservice
AHSWebServices: resourceManager Application Historyservice 
AMWebServices: Mapreduce AM 的webservice
RMWebServices: RM的webservice
ResourceEstimatorService：ResourceEstimator webservice
NMWebServices：NM的webservice
TimelineWebServices：timeline的webservice
TimelineCollectorWebService：timelineserver collector的webservice
TimelineReaderWebServices：timeline read的webservice
LogWebService：Yarn中日志的webservice

KMS & HDFS
HttpsFSserver
RouterWebHdfsMethods：hdfs 基于router的联邦的webservice，通过router访问hdfs的api
NameNodeWebHdfsMethods：hdfs的webservice
RouterWebServices：router的webservice。管理router本身的api

上面这些就接口基本涵盖了一个hadoop集群的方方面面。由于篇幅有限，这里仅仅对RMWebServices为例进行说明。例如RM常见的请求：/ws/v1/cluster/apps

@GET
@Path(RMWSConsts.APPS)
@Produces({ MediaType.APPLICATION_JSON + "; " + JettyUtils.UTF_8,
    MediaType.APPLICATION_XML + "; " + JettyUtils.UTF_8 })
@Override
public AppsInfo getApps(@Context HttpServletRequest hsr,
    @QueryParam(RMWSConsts.STATE) String stateQuery,
    @QueryParam(RMWSConsts.STATES) Set<String> statesQuery,
    @QueryParam(RMWSConsts.FINAL_STATUS) String finalStatusQuery,
    @QueryParam(RMWSConsts.USER) String userQuery,
    @QueryParam(RMWSConsts.QUEUE) String queueQuery,
    @QueryParam(RMWSConsts.LIMIT) String limit,
    @QueryParam(RMWSConsts.STARTED_TIME_BEGIN) String startedBegin,
    @QueryParam(RMWSConsts.STARTED_TIME_END) String startedEnd,
    @QueryParam(RMWSConsts.FINISHED_TIME_BEGIN) String finishBegin,
    @QueryParam(RMWSConsts.FINISHED_TIME_END) String finishEnd,
    @QueryParam(RMWSConsts.APPLICATION_TYPES) Set<String> applicationTypes,
    @QueryParam(RMWSConsts.APPLICATION_TAGS) Set<String> applicationTags,
    @QueryParam(RMWSConsts.DESELECTS) Set<String> unselectedFields) {

  initForReadableEndpoints();

  GetApplicationsRequest request =
          ApplicationsRequestBuilder.create()
                  .withStateQuery(stateQuery)
                  .withStatesQuery(statesQuery)
                  .withUserQuery(userQuery)
                  .withQueueQuery(rm, queueQuery)
                  .withLimit(limit)
                  .withStartedTimeBegin(startedBegin)
                  .withStartedTimeEnd(startedEnd)
                  .withFinishTimeBegin(finishBegin)
                  .withFinishTimeEnd(finishEnd)
                  .withApplicationTypes(applicationTypes)
                  .withApplicationTags(applicationTags)
          .build();

  List<ApplicationReport> appReports;
  try {
    //clientRMService专门用来处理客户端请求。getApplications()返回的是经过一定条件过滤之后的application列表。
    appReports = rm.getClientRMService().getApplications(request).getApplicationList();
  } catch (YarnException e) {
    LOG.error("Unable to retrieve apps from ClientRMService", e);
    throw new YarnRuntimeException(
        "Unable to retrieve apps from ClientRMService", e);
  }

  final ConcurrentMap<ApplicationId, RMApp> apps = rm.getRMContext().getRMApps();
  AppsInfo allApps = new AppsInfo();
  for (ApplicationReport report : appReports) {
    RMApp rmapp = apps.get(report.getApplicationId());
    if (rmapp == null) {
      continue;
    }

    if (finalStatusQuery != null && !finalStatusQuery.isEmpty()) {
      FinalApplicationStatus.valueOf(finalStatusQuery);
      if (!rmapp.getFinalApplicationStatus().toString()
          .equalsIgnoreCase(finalStatusQuery)) {
        continue;
      }
    }

    DeSelectFields deSelectFields = new DeSelectFields();
    deSelectFields.initFields(unselectedFields);
    //当前登录的用户需要有这个application的查看权限或者这个application所在队列的管理权限。即便是查看操作，也只能查看这个用户有权限的application列表。
    boolean allowAccess = hasAccess(rmapp, hsr);
    // Given RM is configured to display apps per user, skip apps to which
    // this caller doesn't have access to view.
    if (filterAppsByUser && !allowAccess) {
      continue;
    }

    AppInfo app = new AppInfo(rm, rmapp, allowAccess,
        WebAppUtils.getHttpSchemePrefix(conf), deSelectFields);
    allApps.add(app);
  }

  if (filterInvalidXMLChars) {
    final String format = hsr.getHeader(HttpHeaders.ACCEPT);
    if (format != null &&
        format.toLowerCase().contains(MediaType.APPLICATION_XML)) {
      for (AppInfo appInfo : allApps.getApps()) {
        //如果请求的accept类型为application/xml,那么需要转移response中的特殊字符。防止xml注入。
        appInfo.setNote(escapeInvalidXMLCharacters(appInfo.getNote()));
      }
    }
  }

  return allApps;
}