深入Ambari Metrics 机制分析

-Dflume.monitoring.type=org.apache.hadoop.metrics2.sink.flume.FlumeTimelineMetricsSink 
-Dflume.monitoring.node=<AMS_HOST>:6188

@Override
  public voidstart() {
    LOG.info("Starting Flume Metrics Sink");
    TimelineMetricsCollector timelineMetricsCollector = newTimelineMetricsCollector();
    if (scheduledExecutorService == null || scheduledExecutorService.isShutdown() ||scheduledExecutorService.isTerminated()) {
      scheduledExecutorService =Executors.newSingleThreadScheduledExecutor();
    }
    scheduledExecutorService.scheduleWithFixedDelay(timelineMetricsCollector, 0,
        pollFrequency, TimeUnit.MILLISECONDS);
  }  }

classTimelineMetricsCollector implements Runnable {
    @Override
    public voidrun() {
      try{
        Map<String, Map<String, String>> metricsMap =JMXPollUtil.getAllMBeans();
        long currentTimeMillis =System.currentTimeMillis();
        for(String component : metricsMap.keySet()) {
          Map<String, String> attributeMap = metricsMap.get(component);
          LOG.debug("Attributes for component " +component);
          processComponentAttributes(currentTimeMillis, component, attributeMap);
        }
      } 

protected voidemitMetrics(TimelineMetrics metrics) throws IOException {
    String connectUrl =getCollectorUri();
    try{
      String jsonData =mapper.writeValueAsString(metrics);
      StringRequestEntity requestEntity = new StringRequestEntity(jsonData, "application/json", "UTF-8");
      PostMethod postMethod = newPostMethod(connectUrl);
      postMethod.setRequestEntity(requestEntity);
      int statusCode = httpClient.executeMethod(postMethod);

curl -i -X POST -H "Content-Type: application/json" -d "${json}" ${url}

@Path("/metrics")
@POST
@Consumes({ MediaType.APPLICATION_JSON /*, MediaType.APPLICATION_XML */})
  publicTimelinePutResponse postMetrics(
    @Context HttpServletRequest req,
    @Context HttpServletResponse res,
    TimelineMetrics metrics) {
    init(res);
    if (metrics == null) {
      return newTimelinePutResponse();
    }
    try{
      //TODO: Check ACLs for MetricEntity using the TimelineACLManager.
      //TODO: Save owner of the MetricEntity.
      returntimelineMetricStore.putMetrics(metrics);
}

    （1）select hostname,max(metric_max) from metric_record where metric_name='mem_free' and server_time>=1471503845324 and server_time<1471504146520 group by hostname;------统计5分钟内，每台主机上该metric的最大值。
　　（2）select hostname,min(metric_min) from metric_record where metric_name='mem_free' and server_time>=1471503845324 and server_time<1471504146520 group by hostname; ------统计5分钟内，每台主机上该metric的最小值。
　　（3）select hostname,sum(metric_sum) from metric_record where metric_name='mem_free' and server_time>=1471503845324 and server_time<1471504146520 group by hostname; ------统计5分钟内，每台主机上该metric值总和。
　　（4）select hostname,sum(metric_count) from metric_record where metric_name='mem_free' and server_time>=1471503845324 and server_time<1471504146520 group by hostname; ------统计5分钟内，每台主机上统计的该metric数量和。

@Override
  public boolean doWork(long startTime, longendTime) {
    boolean success = true;
    Condition condition =prepareMetricQueryCondition(startTime, endTime);
    Connection conn = null;
    PreparedStatement stmt = null;
    ResultSet rs = null;
    try{
      conn =hBaseAccessor.getConnection();
      stmt =PhoenixTransactSQL.prepareGetMetricsSqlStmt(conn, condition);
      if(condition.doUpdate()) {
        int rows =stmt.executeUpdate();
        conn.commit();
      } else{
        rs =stmt.executeQuery();
      }
      aggregate(rs, startTime, endTime);
      }

http://<AMS_HOST>:6188/ws/v1/timeline/metrics?metricNames=<>&hostname=<>&appId=<>&startTime=<>&endTime=<>&precision=<>

@GET
@Path("/metrics")
@Produces({ MediaType.APPLICATION_JSON })
  publicTimelineMetrics getTimelineMetrics(
    @Context HttpServletRequest req,
    @Context HttpServletResponse res,
    @QueryParam("metricNames") String metricNames,
    @QueryParam("appId") String appId,
    @QueryParam("instanceId") String instanceId,
    @QueryParam("hostname") String hostname,
    @QueryParam("startTime") String startTime,
    @QueryParam("endTime") String endTime,
    @QueryParam("precision") String precision,
    @QueryParam("limit") String limit,
    @QueryParam("grouped") String grouped,
    @QueryParam("topN") String topN,
    @QueryParam("topNFunction") String topNFunction,
    @QueryParam("isBottomN") String isBottomN,
    @QueryParam("seriesAggregateFunction") String seriesAggregateFunction
 ) {
    init(res);
    try{
      returntimelineMetricStore.getTimelineMetrics(
        parseListStr(metricNames, ","), parseListStr(hostname, ","), appId, instanceId,
        parseLongStr(startTime), parseLongStr(endTime),
        Precision.getPrecision(precision), parseIntStr(limit),
        parseBoolean(grouped), parseTopNConfig(topN, topNFunction, isBottomN),
        seriesAggregateFunction);
}

@Override
  public TimelineMetrics getTimelineMetrics(List<String>metricNames,
      List<String>hostnames, String applicationId, String instanceId,
      Long startTime, Long endTime, Precision precision, Integer limit,
      boolean groupedByHosts, TopNConfig topNConfig, String seriesAggregateFunction) throws SQLException, IOException {
    TimelineMetricsSeriesAggregateFunction seriesAggrFunctionInstance = null;
    ……………………………………….
    Multimap<String, List<Function>> metricFunctions =
      parseMetricNamesToAggregationFunctions(metricNames);
    ConditionBuilder conditionBuilder = new ConditionBuilder(new ArrayList<String>(metricFunctions.keySet()))
      .hostnames(hostnames)
      .appId(applicationId)
      .instanceId(instanceId)
      .startTime(startTime)
      .endTime(endTime)
      .precision(precision)
      .limit(limit)
      .grouped(groupedByHosts);
………………………………………….
………………………………………
    Condition condition =conditionBuilder.build();
    TimelineMetrics metrics;
    if (hostnames == null ||hostnames.isEmpty()) {
      metrics =hBaseAccessor.getAggregateMetricRecords(condition, metricFunctions);
    } else{
      metrics =hBaseAccessor.getMetricRecords(condition, metricFunctions);
    }
    metrics =postProcessMetrics(metrics);
    ………………………………..
    returnseriesAggregateMetrics(seriesAggrFunctionInstance, metrics);
  }

http://<ambari-server>:8080/api/v1/clusters/<cluster-name>/hosts/<host-name>?fields=metrics/cpu/cpu_user[1430844610,1430848210,15]

http://<ambari-server>:8080/api/v1/clusters/<cluster-name>/services/HDFS/components/DATANODE?fields=metrics/dfs/datanode/DfsUsed[1430844610,1430848210,15]

public Collection<Resource>populateResources() throws SystemException {
if (!hostComponentHostMetrics.isEmpty()) {
        String hostComponentHostMetricParams = getSetString(processRegexps(hostComponentHostMetrics), -1);
         setQueryParams(hostComponentHostMetricParams, hostnames, true, componentName);
          TimelineMetrics metricsResponse = null;
          try{
            metricsResponse =getTimelineMetricsFromCache(
              getTimelineAppMetricCacheKey(hostComponentHostMetrics,
                componentName, uriBuilder.toString(),hostnames), componentName);
          } 

privateTimelineMetrics getTimelineMetricsFromCache(TimelineAppMetricCacheKey metricCacheKey,String componentName) throws IOException {
      //Cache only the component level metrics
      //No point in time metrics are cached
      if (metricCache != null
          && !StringUtils.isEmpty(componentName)
          && !componentName.equalsIgnoreCase("HOST")
          && metricCacheKey.getTemporalInfo() != null) {
        returnmetricCache.getAppTimelineMetricsFromCache(metricCacheKey);
      }
      returnrequestHelper.fetchTimelineMetrics(metricCacheKey.getSpec());
    }

publicTimelineMetrics fetchTimelineMetrics(String spec) throws IOException {
    LOG.debug("Metrics request url = " +spec);
    BufferedReader reader = null;
    TimelineMetrics timelineMetrics = null;
    try{
      reader = new BufferedReader(newInputStreamReader(streamProvider.readFrom(spec)));
      timelineMetrics =timelineObjectReader.readValue(reader);
}