Cloudera Manager API使用

前言

接口介绍

使用实例

postman方式

java代码方式

前言

1. 需求：如何获取CDH集群的资源使用情况，包括cpu、磁盘、内存
2. 实现：调用CDH提供的Cloudera Manager API包，来实时访问CDH上的各种监控指标数据，如下图

接口介绍

1 获取集群信息

2 获取主机信息

3 获取服务信息

4 获取指标时序数据信息

5 详细信息看：官网文档，使用实例demo：后面

- cloudera官网文档：https://docs.cloudera.com/documentation/enterprise/6/6.1/topics/cm_intro_primer.html

- cloudera manager api接口文档：https://archive.cloudera.com/cm6/6.3.0/generic/jar/cm_api/apidocs/index.html

使用实例

postman方式

1. 举例：获取集群已使用内存和总内存

1）首先我得知道已使用内存和总内存对应的接口是哪个，官网接口文档，在timeseries时间序列接口里面

2）参数说明:from（开始时间）、to（截至时间）、query（查询语句）
3）query是查询指标语句，该语句怎么获取呢，入下图：

你也可以把该查询语句拷贝出来，在这个界面查询

4）好了，查询接口知道了，参数也知道了，示例如下：

get接口：http://utility:7180/api/v13/timeseries?query=select%20physical_memory_used/(1024*1024*1024),physical_memory_total/(1024*1024*1024)&from=2020-09-27&to=2020-09-28① 接口url  http://utility:7180/api/v13/timeseries②query=select%20physical_memory_used/(1024*1024*1024),physical_memory_total/(1024*1024*1024)③ from=2020-09-27④ to=2020-09-28# 指定查询开始时间和查询的结束时间(cdh的监控数据都是时间序列数据，构成的一个监控图表)
# 上述示范的是最近一天的监控数据，如果你只需要最近10分钟的，可以这样指定（他源码是用到LoacalDateTime的字符串格式）from=2020-09-28T14:30:30.100
to=2020-09-28T14:30:40.100⑤ 之后记得authorization信息填上cdh登陆的用户名和密码

java代码方式

代码github地址：https://github.com/yangxifi/cm-api-usage.git，如果觉得本文章有用的话，欢迎star、点赞

0 引入依赖包（cloudera-manager-api包）

        <dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><version>1.18.12</version></dependency><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>1.2.72</version></dependency><dependency><groupId>com.cloudera.api</groupId><artifactId>cloudera-manager-api</artifactId><version>7.x.0</version></dependency><dependency><groupId>org.apache.commons</groupId><artifactId>commons-lang3</artifactId><version>3.9</version></dependency><dependency><groupId>org.slf4j</groupId><artifactId>slf4j-log4j12</artifactId><version>1.7.25</version></dependency>

1 先登陆

import com.cloudera.api.ClouderaManagerClientBuilder;
import com.cloudera.api.v30.RootResourceV30;/*** @author : yangx* @date : 2020/9/28 20:29* @description :*/
public class CMClientUtil {private static String cdhApiPath = "utility";private static String cdhApiPort = "7180";private static String cdhApiUserName = "yourname";private static String cdhApiPassword = "yourpwd";private static RootResourceV30 apiRoot;public static RootResourceV30 getApiRoot(){if(apiRoot!=null){return apiRoot;}RootResourceV30 apiRoot1 =new ClouderaManagerClientBuilder().withHost(cdhApiPath).withPort(Integer.parseInt(cdhApiPort)).withUsernamePassword(cdhApiUserName, cdhApiPassword).build().getRootV30();apiRoot=apiRoot1;return apiRoot1;}
}

2 获取集群信息

import com.cloudera.api.DataView;
import com.cloudera.api.model.ApiCluster;
import com.cloudera.api.model.ApiClusterList;
import com.cloudera.api.v30.RootResourceV30;
import com.data.util.CMClientUtil;
import lombok.extern.slf4j.Slf4j;import java.time.LocalDateTime;@Slf4j
//获取cloudera-manager中所有集群的信息
public class CMCluster {static RootResourceV30 apiRoot= CMClientUtil.getApiRoot();public static void main(String[] args) {getAllCluster();}public static void getAllCluster() {log.info("开始测试的时间为{},**************开始测试获取ClouderaManager集群信息**************", LocalDateTime.now());ApiClusterList apiClusterList = apiRoot.getClustersResource().readClusters(DataView.FULL);log.info("ClouderaManager 共管理了{}个集群", apiClusterList.getClusters().size());System.out.println("ClouderaManager 共管理几个集群："+apiClusterList.getClusters().size());for (ApiCluster apiCluster : apiClusterList) {ApiCluster apiCluster1 = apiRoot.getClustersResource().readCluster(apiCluster.getName());log.info("集群名称 {}", apiCluster1.getName());log.info("集群显示名称 {}", apiCluster1.getDisplayName());log.info("CDH 版本：{}-{}", apiCluster1.getVersion(), apiCluster.getFullVersion());log.info("ClusterUrl {}", apiCluster1.getClusterUrl());log.info("HostUrl {}", apiCluster1.getHostsUrl());log.info("Cluster Uuid {}", apiCluster1.getUuid());log.info("集群运行状态 {}", apiCluster1.getEntityStatus());}log.info("结束测试的时间为{},**************结束测试获取ClouderaManager集群信息**************", LocalDateTime.now());}}

3 获取主机信息

import com.alibaba.fastjson.JSON;
import com.cloudera.api.DataView;
import com.cloudera.api.model.ApiHost;
import com.cloudera.api.model.ApiRoleRef;
import com.cloudera.api.v10.HostsResourceV10;
import com.cloudera.api.v18.RootResourceV18;
import com.data.util.CMClientUtil;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;@Slf4j
//获取cloudera-manager中每个集群的每个主机的详细信息：
public class CMHost {static RootResourceV18 apiRoot= CMClientUtil.getApiRoot();public static void main(String[] args) {getAllHost();}public static ArrayList<Host> getAllHost(){log.info("开始测试的时间为{},**************开始测试集群主机运行状态**************", LocalDateTime.now());HostsResourceV10 hostsResourceV10 = apiRoot.getHostsResource();List<ApiHost> hostList = hostsResourceV10.readHosts(DataView.SUMMARY).getHosts();log.info("总共有 {} 台主机组成集群",hostList.size());ArrayList<Host> hosts = new ArrayList<>();for(ApiHost apiHost:hostList){log.info("---------------------------------------------");Host host = formatHost(hostsResourceV10.readHost(apiHost.getHostId()));log.info("主机Id : {}",host.getHostId());log.info("主机名： {}",host.getHostName());log.info("主机IP： {}",host.getIpAddress());log.info("主机线程数：{}",host.getNumCores());log.info("上次上报心跳时间 ：{}",host.getLastHeart());log.info("核心数：{}",host.getNumPhysicalCores());log.info("机架：{}",host.getRack());log.info("内存（G）：{}",host.getTotalPhysMemBytes());log.info("进程：{}", JSON.toJSON(host.getServices()));log.info("---------------------------------------------");hosts.add(host);}log.info("结束测试的时间为{},**************结束测试集群主机运行状态**************",LocalDateTime.now());return hosts;}public static Host formatHost(ApiHost apiHost){Host host = new Host();List<String> services = new ArrayList<>();host.setHostId(apiHost.getHostId());host.setHostName(apiHost.getHostname());host.setIpAddress(apiHost.getIpAddress());host.setNumCores(apiHost.getNumCores());host.setNumPhysicalCores(apiHost.getNumPhysicalCores());host.setLastHeart(apiHost.getLastHeartbeat().toString());host.setRack(apiHost.getRackId());host.setTotalPhysMemBytes(apiHost.getTotalPhysMemBytes()/1073741824);for(ApiRoleRef apiRoleRef:apiHost.getRoleRefs()){services.add(apiRoleRef.getRoleName());}host.setServices(services);return host;}
}@Data
@AllArgsConstructor
@NoArgsConstructor
class Host {private String hostId;private String hostName;private String ipAddress;private String rack;private String lastHeart;private List<String> services = new ArrayList<>();private long numCores;private long numPhysicalCores;private long totalPhysMemBytes;
}

4 获取服务信息

package com.data.service;import com.cloudera.api.DataView;
import com.cloudera.api.model.*;
import com.cloudera.api.v10.ServicesResourceV10;
import com.cloudera.api.v11.RolesResourceV11;
import com.cloudera.api.v18.ServicesResourceV18;
import com.cloudera.api.v30.RootResourceV30;
import com.data.util.CMClientUtil;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;@Slf4j
//获取cloudera-manager上所有集群的所有服务的状态
public class CMService {static RootResourceV30 apiRoot= CMClientUtil.getApiRoot();public static void main(String[] args) {getAllService();}public static void getAllService(){log.info("开始测试的时间为{},**************开始测试集群服务运行状态**************", LocalDateTime.now());ApiClusterList apiClusterList = apiRoot.getClustersResource().readClusters(DataView.SUMMARY);for(ApiCluster apiCluster:apiClusterList){log.info("集群名称：{}",apiCluster.getDisplayName());log.info("CDH 版本：{}-{}",apiCluster.getVersion(),apiCluster.getFullVersion());ServicesResourceV10 servicesResourceV10 = apiRoot.getClustersResource().getServicesResource(apiCluster.getName());List<ApiService> apiServices = servicesResourceV10.readServices(DataView.FULL).getServices();log.info("集群总共有：{} 个service 在运行",apiServices.size());for(ApiService apiService:apiServices){Service service = formatService(apiService);log.info("***********************************");log.info("service 名称 {}",service.getName());log.info("service 类型 {}",service.getType());for(Agent agent:service.getAgentList()) {log.info("节点名称 {}", agent.getName());log.info("节点状态 {}", agent.getStatus());}log.info("***********************************");}}log.info("结束测试的时间为{},**************结束测试集群服务运行状态**************",LocalDateTime.now());}public static Service formatService(ApiService apiService){Service service = new Service();List<Agent> agents = new ArrayList<>();service.setName(apiService.getName());service.setType(apiService.getType());for(ApiHealthCheck apiHealthCheck:apiService.getHealthChecks()){Agent agent =new Agent();agent.setName(apiHealthCheck.getName());agent.setStatus(apiHealthCheck.getSummary());agents.add(agent);}service.setAgentList(agents);return service;}public static void getAllServiceRoles(){log.info("开始测试的时间为{},**************开始测试集群各个服务的roles运行状态**************",LocalDateTime.now());ApiClusterList apiClusterList = apiRoot.getClustersResource().readClusters(DataView.SUMMARY);for(ApiCluster apiCluster:apiClusterList){log.info("集群名称：{}",apiCluster.getDisplayName());log.info("CDH 版本：{}-{}",apiCluster.getVersion(),apiCluster.getFullVersion());ServicesResourceV18 servicesResourceV18 = apiRoot.getClustersResource().getServicesResource(apiCluster.getName());List<ApiService> apiServices = servicesResourceV18.readServices(DataView.FULL).getServices();log.info("集群总共有：{} 个service 在运行",apiServices.size());for(ApiService apiService:apiServices){RolesResourceV11 rolesResourceV11 = servicesResourceV18.getRolesResource(apiService.getName());log.info("---------------------服务名称是{}---------------------",apiService.getName());for(ApiRole apiRole :rolesResourceV11.readRoles()){log.info("***************************",apiRole.getName());log.info("role名称 {}",apiRole.getName());log.info("role类型 {}",apiRole.getType());log.info("所属集群 {}",apiRole.getServiceRef().getClusterName());log.info("所属服务 {}",apiRole.getServiceRef().getServiceName());log.info("主机ID {}",apiRole.getHostRef().getHostId());log.info("roleUrl {}",apiRole.getRoleUrl());log.info("role状态 {}",apiRole.getRoleState());log.info("运行状态总结 {}",apiRole.getHealthSummary());log.info("entityStatus {}",apiRole.getEntityStatus());log.info("roleConfigGroupName {}",apiRole.getRoleConfigGroupRef().getRoleConfigGroupName());log.info("configStalenessStatus {}",apiRole.getConfigStalenessStatus());log.info("haStatus {}",apiRole.getHaStatus());for(ApiHealthCheck apiHealthCheck:apiRole.getHealthChecks()){log.info("health check name {}",apiHealthCheck.getName());log.info("health check summary {}",apiHealthCheck.getSummary());log.info("health check suppressed {}",apiHealthCheck.getSuppressed());}log.info("***************************");}log.info("--------------------------------------------------------",apiService.getName());}}log.info("结束测试的时间为{},**************结束测试集群各个服务的roles运行状态**************",LocalDateTime.now());}}@Data
@AllArgsConstructor
@NoArgsConstructor
class Service {private String name;private String type;private List<Agent> agentList = new ArrayList<>();
}@Data
@AllArgsConstructor
@NoArgsConstructor
class Agent {private String name;private ApiHealthSummary status;
}

5 获取指标时序数据信息

package com.data.service;import com.alibaba.fastjson.JSON;
import com.cloudera.api.model.*;
import com.cloudera.api.v11.TimeSeriesResourceV11;
import com.cloudera.api.v30.RootResourceV30;
import com.data.bean.Resource;
import com.data.constant.Metrics;
import com.data.util.CMClientUtil;import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.util.*;
import java.util.stream.Collectors;@Slf4j
//获取cloudera-manager中所有监控图表的时序数据
public class CMTimeSeriesMetric {static RootResourceV30 apiRoot = CMClientUtil.getApiRoot();public static void main(String[] args) {System.out.println(getResources());}public static Map<String,List<Resource>> getResources(){HashMap<String, List<Resource>> map = new HashMap<>();map.put(Metrics.MEM,computeMemory());map.put(Metrics.DISK,computeDisk());map.put(Metrics.CPU,computeCpu());System.out.println(map);return map;}/*** @author : yangx* @date : 2020.09.29 20:16* @description : 计算cdh的disk：已用disk、总disk,单位:GB* @param :* @return :**/private static List<Resource> computeDisk(){String diskQuery=String.format("select %s,%s WHERE category=HOST",Metrics.DISK_USED_COMPUTE_RULE,Metrics.DISK_TOTAL_COMPUTE_RULE);List<Metric> allMetric = getServiceMetrics(diskQuery,LocalDateTime.now().toString(), "now");Map<String, List<Metric>> metricGroupBy = allMetric.stream().collect(Collectors.groupingBy(Metric::getMetricName));Double diskUsed = 0.0;Double diskTotal = 0.0;for (Map.Entry<String, List<Metric>> metrics : metricGroupBy.entrySet()) {String metricName = metrics.getKey();Double tempDisk = 0.0;for (Metric metric : metrics.getValue()) {//该dataList只有一个元素的Data data = metric.getData().get(0);tempDisk += data.getValue();}if (StringUtils.equals(metricName, Metrics.DISK_USED_COMPUTE_RULE)) {diskUsed = tempDisk;} else {diskTotal = tempDisk;}}diskUsed = new BigDecimal(diskUsed).setScale(2, BigDecimal.ROUND_HALF_UP).doubleValue();diskTotal = new BigDecimal(diskTotal).setScale(2, BigDecimal.ROUND_HALF_UP).doubleValue();Resource used = new Resource(Metrics.DISK_USED, diskUsed, "GB");Resource total = new Resource(Metrics.DISK_TOTAL,diskTotal, "GB");List<Resource> list = new ArrayList<>();list.add(used);list.add(total);return list;}/*** @author : yangx* @date : 2020.09.29 19:46* @description : 计算cdh的cpu：已用cpu核心数、总cpu核心数，单位:个* @param :* @return :**/private static List<Resource> computeCpu() {ArrayList<Host> allHost = CMHost.getAllHost();long cpuUsed = 0L;long cpuTotal = 0L;for (Host host : allHost) {long numCores = host.getNumCores();cpuTotal += numCores;String cpuQuery=String.format("select cpu_user_rate / getHostFact(numCores, 1) , cpu_system_rate / getHostFact(numCores, 1) , cpu_nice_rate / getHostFact(numCores, 1) , cpu_iowait_rate / getHostFact(numCores, 1) , cpu_irq_rate / getHostFact(numCores, 1) , cpu_soft_irq_rate / getHostFact(numCores, 1) , cpu_steal_rate / getHostFact(numCores, 1)  where hostId=%s",host.getHostId());List<Metric> allMetric = getServiceMetrics(cpuQuery,LocalDateTime.now().toString(), "now");Double temUsedRatio = 0.0;for (Metric metric : allMetric) {//该dataList只有一个元素的System.out.println(metric);Data data = metric.getData().get(0);temUsedRatio += data.getValue();}
//            System.out.println("hostName:"+hostName+",temUsedRatio:"+temUsedRatio);double numCoresUsed = temUsedRatio * numCores;cpuUsed += numCoresUsed;}Resource used = new Resource(Metrics.CPU_USED, cpuUsed, "num");Resource total = new Resource(Metrics.CPU_TOTAL, cpuTotal, "num");List<Resource> list = new ArrayList<>();list.add(used);list.add(total);return list;}/*** @param :* @return :* @author : yangx* @date : 2020.09.29 18:56* @description : 计算出此刻cdh集群：使用的内存和总内存,单位:GB**/private static List<Resource> computeMemory() {String memQuery = String.format("select %s,%s", Metrics.MEM_USED_COMPUTE_RULE, Metrics.MEM_TOTAL_COMPUTE_RULE);List<Metric> allMetric = getServiceMetrics(memQuery, LocalDateTime.now().toString(), "now");Double memUsed = 0.0;Double memTotal = 0.0;Map<String, List<Metric>> metricGroupBy = allMetric.stream().collect(Collectors.groupingBy(Metric::getMetricName));for (Map.Entry<String, List<Metric>> metrics : metricGroupBy.entrySet()) {String metricName = metrics.getKey();Double tempMem = 0.0;for (Metric metric : metrics.getValue()) {//该dataList只有一个元素的Data data = metric.getData().get(0);tempMem += data.getValue();}if (StringUtils.equals(metricName, Metrics.MEM_USED_COMPUTE_RULE)) {memUsed = tempMem;} else {memTotal = tempMem;}}memUsed = new BigDecimal(memUsed).setScale(2, BigDecimal.ROUND_HALF_UP).doubleValue();memTotal = new BigDecimal(memTotal).setScale(2, BigDecimal.ROUND_HALF_UP).doubleValue();Resource used = new Resource(Metrics.MEM_USED, memUsed, "GB");Resource total = new Resource(Metrics.MEM_TOTAL, memTotal, "GB");List<Resource> list = new ArrayList<>();list.add(used);list.add(total);return list;}private static List<Metric> formatApiTimeSeriesResponse(List<ApiTimeSeriesResponse> apiTimeSeriesResponseList) {List<Metric> metrics = new ArrayList<>();for (ApiTimeSeriesResponse apiTimeSeriesResponse : apiTimeSeriesResponseList) {List<ApiTimeSeries> apiTimeSeriesList = apiTimeSeriesResponse.getTimeSeries();for (ApiTimeSeries apiTimeSeries : apiTimeSeriesList) {Metric metric = new Metric();metric.setMetricName(apiTimeSeries.getMetadata().getMetricName());metric.setEntityName(apiTimeSeries.getMetadata().getEntityName());metric.setStartTime(apiTimeSeries.getMetadata().getStartTime().toString());metric.setEndTime(apiTimeSeries.getMetadata().getEndTime().toString());List<ApiTimeSeriesData> timeSeriesDataList = apiTimeSeries.getData();//他默认按照时间排序的，我只取最新的，即最后一条//ps：cdh最细的粒度是按照10分钟聚合的if (timeSeriesDataList.size() == 0) {//为空，不处理log.warn("metric has error:{}",metric);continue;}List<Data> dataList = new ArrayList<>();Data data = new Data();data.setTimestamp(apiTimeSeriesData.getTimestamp().toString());data.setType(apiTimeSeriesData.getType());data.setValue(apiTimeSeriesData.getValue());dataList.add(data);metric.setData(dataList);metrics.add(metric);}}return metrics;}private static List<Metric> getServiceMetrics(String query, String startTime, String endTime) {TimeSeriesResourceV11 timeSeriesResourceV11 = apiRoot.getTimeSeriesResource();String[] params = new String[]{query, startTime, endTime};log.info("query sql is {} ,startTime is {} ,endTime is now", params);log.info("开始测试的时间为{},**************开始查询某个服务点位状态**************", LocalDateTime.now());ApiTimeSeriesResponseList response = timeSeriesResourceV11.queryTimeSeries(query, startTime, endTime);List<ApiTimeSeriesResponse> apiTimeSeriesResponseList = response.getResponses();List<Metric> metrics = formatApiTimeSeriesResponse(apiTimeSeriesResponseList);log.info("查询时间序列点位:{}", JSON.toJSON(metrics));log.info("结束测试的时间为{},**************结束查询某个服务点位状态**************", LocalDateTime.now());return metrics;}}@lombok.Data
class Metric {private String metricName;private String entityName;private String startTime;private String endTime;List<Data> data = new ArrayList<>();
}@lombok.Data
class Data {private String timestamp;private Double value;private String type;
}

6 代码中其他相关的类

package com.data.bean;import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;@Data
@AllArgsConstructor
@NoArgsConstructor
public class Resource {private String key;//cpu mem diskprivate Object value;private String unit;//单位  num  GB  GB
}==============================================================
==============================================================package com.data.constant;public class Metrics {public static final String MEM_USED_COMPUTE_RULE="physical_memory_used/(1024*1024*1024)";public static final String MEM_TOTAL_COMPUTE_RULE="physical_memory_total/(1024*1024*1024)";public static final String DISK_USED_COMPUTE_RULE="total_capacity_used_across_filesystems/(1024*1024*1024)";public static final String DISK_TOTAL_COMPUTE_RULE="total_capacity_across_filesystems/(1024*1024*1024)";public static final String CPU_USER_COMPUTE_RULE="cpu_user_rate / getHostFact(numCores, 1) * 100";public static final String CPU_SYSTEM_COMPUTE_RULE="cpu_system_rate / getHostFact(numCores, 1) * 100";public static final String MEM="mem";public static final String DISK="disk";public static final String CPU="cpu";public static final String MEM_USED="memUSed";public static final String DISK_USED="diskUSed";public static final String CPU_USED="cpuUSed";public static final String MEM_TOTAL="memTotal";public static final String DISK_TOTAL="diskTotal";public static final String CPU_TOTAL="cpuTotal";}

7 执行CMTimeSeriesMetric的main方法，获取集群cpu、磁盘、内存使用情况

1. 执行CMTimeSeriesMetric的main方法，获取集群cpu、磁盘、内存使用情况2. 打印如下：{disk=[Resource(key=diskUSed, value=5226.12, unit=GB), Resource(key=diskTotal, value=11533.7, unit=GB)], mem=[Resource(key=memUSed, value=107.05, unit=GB), Resource(key=memTotal, value=306.4, unit=GB)], cpu=[Resource(key=cpuUSed, value=2, unit=num), Resource(key=cpuTotal, value=48, unit=num)]}

最后：文章如果有什么地方说的不对的，欢迎指正，有何问题的话，也可以联系我，欢迎交流