使用Thrift API监控Storm集群和Topology

拼搏现实的明天。 2022-06-02 23:07 217阅读 0赞

如要监控Storm集群和运行在其上的Topology，该如何做呢？

Storm已经为你考虑到了，Storm支持Thrift的C/S架构，在部署Nimbus组件的机器上启动一个Thrift Server进程来提供服务，我们可以通过编写一个Thrift Client来请求Thrift Server，来获取你想得到的集群和Topology的相关数据，来接入监控平台，如Zabbix等，我目前使用的就是Zabbix。

整体的流程已经清楚了，下面就来实践吧。

1 安装Thrift

由于我们要使用Thrift来编译Storm的源代码来获得Thrift Client相关的Java源代码，所以需要先安装Thrift，这里选取的版本为0.9.2。

到官网下载好安装包：http://thrift.apache.org/

编译安装：configure && make && make install

验证：thrift --version

如果打印出Thrift version 0.9.2，代表安装成功。

2 编译Thrift Client代码

首先下载Storm源代码，这里使用最新的0.9.3版本：http://mirrors.hust.edu.cn/apache/storm/apache-storm-0.9.3/apache-storm-0.9.3-src.tar.gz

解压后进行编译：thrift -gen java  apache-storm-0.9.3/storm-core/src/storm.thrift

在当前目录下出现gen-java文件夹，此文件夹下就是Thrift Client的Java源代码了。

3 使用Thrift Client API

然后创建一个Maven项目来进行执行监控数据的获取。

项目生成一个Jar文件，输入一些命令和自定义参数，然后输出结果。

以命令行的形式进行调用，这样可以方便的接入监控系统，当然使用形式可以根据自身情况施行。

创建好后，把gen-java生成的代码拷贝进来。

在pom.xml里引入Thrift对应版本的库：

\[html\]  [view plain][]  [copy][view plain]

1.  <dependency>  
2.      <groupId>org.apache.thrift</groupId>  
3.      <artifactId>libthrift</artifactId>  
4.      <version>0.9.2</version>  
5.  </dependency>

首先写一些Thrift相关的辅助类。

ClientInfo.java

\[java\]  [view plain][]  [copy][view plain]

1.  package com.damacheng009.storm.monitor.thrift;  
2.    
3.  import org.apache.thrift.protocol.TBinaryProtocol;  
4.  import org.apache.thrift.transport.TFramedTransport;  
5.  import org.apache.thrift.transport.TSocket;  
6.    
7.  import backtype.storm.generated.Nimbus;  
8.    
9.  /\*\* 
10.  \* 代表一个Thrift Client的信息 
11.  \* @author jb-xingchencheng 
12.  \* 
13.  \*/  
14. public class ClientInfo \{  
15.     private TSocket tsocket;  
16.     private TFramedTransport tTransport;  
17.     private TBinaryProtocol tBinaryProtocol;  
18.     private Nimbus.Client client;  
19.   
20.     public TSocket getTsocket() \{  
21.         return tsocket;  
22.     \}  
23.   
24.     public void setTsocket(TSocket tsocket) \{  
25.         this.tsocket = tsocket;  
26.     \}  
27.   
28.     public TFramedTransport gettTransport() \{  
29.         return tTransport;  
30.     \}  
31.   
32.     public void settTransport(TFramedTransport tTransport) \{  
33.         this.tTransport = tTransport;  
34.     \}  
35.   
36.     public TBinaryProtocol gettBinaryProtocol() \{  
37.         return tBinaryProtocol;  
38.     \}  
39.   
40.     public void settBinaryProtocol(TBinaryProtocol tBinaryProtocol) \{  
41.         this.tBinaryProtocol = tBinaryProtocol;  
42.     \}  
43.   
44.     public Nimbus.Client getClient() \{  
45.         return client;  
46.     \}  
47.   
48.     public void setClient(Nimbus.Client client) \{  
49.         this.client = client;  
50.     \}  
51. \}

ClientManager.java

\[java\]  [view plain][]  [copy][view plain]

1.  package com.damacheng009.storm.monitor.thrift;  
2.    
3.  import org.apache.thrift.protocol.TBinaryProtocol;  
4.  import org.apache.thrift.transport.TFramedTransport;  
5.  import org.apache.thrift.transport.TSocket;  
6.  import org.apache.thrift.transport.TTransportException;  
7.    
8.  import backtype.storm.generated.Nimbus;  
9.    
10. /\*\* 
11.  \* Thrift Client管理类 
12.  \* @author jb-xingchencheng 
13.  \* 
14.  \*/  
15. public class ClientManager \{  
16.     public static ClientInfo getClient(String nimbusHost, int nimbusPort) throws TTransportException \{  
17.         ClientInfo client = new ClientInfo();  
18.         TSocket tsocket = new TSocket(nimbusHost, nimbusPort);  
19.         TFramedTransport tTransport = new TFramedTransport(tsocket);  
20.         TBinaryProtocol tBinaryProtocol = new TBinaryProtocol(tTransport);  
21.         Nimbus.Client c = new Nimbus.Client(tBinaryProtocol);  
22.         tTransport.open();  
23.         client.setTsocket(tsocket);  
24.         client.settTransport(tTransport);  
25.         client.settBinaryProtocol(tBinaryProtocol);  
26.         client.setClient(c);  
27.           
28.         return client;    
29.     \}  
30.       
31.     public static void closeClient(ClientInfo client) \{  
32.         if (null == client) \{  
33.             return;  
34.         \}  
35.           
36.         if (null != client.gettTransport()) \{  
37.             client.gettTransport().close();  
38.         \}  
39.           
40.         if (null != client.getTsocket()) \{  
41.             client.getTsocket().close();  
42.         \}  
43.     \}  
44. \}

然后就可以写自己的逻辑去获取集群和拓扑的数据了，Storm提供的UI界面上展示的数据基本都可以获取到，这里只举出一个简单的例子，我们想获得某个拓扑发生异常的次数，和发生的异常的堆栈。剩下的项目你可以随意的定制。

下面是入口类：

Main.java

\[java\]  [view plain][]  [copy][view plain]

1.  package com.damacheng009.storm.monitor;  
2.    
3.  import com.damacheng009.storm.monitor.logic.Logic;  
4.    
5.  /\*\* 
6.   \* 入口类 
7.   \* @author jb-xingchencheng 
8.   \* 
9.   \*/  
10. public class Main \{  
11.     // NIMBUS的信息  
12.     public static String NIMBUS\_HOST = "192.168.180.36";  
13.     public static int NIMBUS\_PORT = 6627;  
14.   
15.     /\*\* 
16.      \* 命令格式 CMD（命令） \[ARG0\] \[ARG1\] ...（更多参数） 
17.      \* @param args 
18.      \*/  
19.     public static void main(String\[\] args) \{  
20.         if (args.length < 3) \{  
21.             return;  
22.         \}  
23.           
24.         NIMBUS\_HOST = args\[0\];  
25.         NIMBUS\_PORT = Integer.parseInt(args\[1\]);  
26.           
27.         String cmd = args\[2\];  
28.         String result = "-1";  
29.         if (cmd.equals("get\_topo\_exp\_size")) \{  
30.             String topoName = args\[3\];  
31.             result = Logic.getTopoExpSize(topoName);  
32.         \} else if (cmd.equals("get\_topo\_exp\_stack\_trace")) \{  
33.             String topoName = args\[3\];  
34.             result = Logic.getTopoExpStackTrace(topoName);  
35.         \}  
36.           
37.         System.out.println(result);  
38.     \}  
39. \}

测试的时候把具体的HOST和PORT改一下即可。

然后是具体的逻辑类。

Logic.java

\[java\]  [view plain][]  [copy][view plain]

1.  package com.damacheng009.storm.monitor.logic;  
2.    
3.  import java.util.Date;  
4.  import java.util.List;  
5.  import java.util.Set;  
6.    
7.  import com.damacheng009.storm.monitor.Main;  
8.  import com.damacheng009.storm.monitor.thrift.ClientInfo;  
9.  import com.damacheng009.storm.monitor.thrift.ClientManager;  
10.   
11. import backtype.storm.generated.ClusterSummary;  
12. import backtype.storm.generated.ErrorInfo;  
13. import backtype.storm.generated.TopologyInfo;  
14. import backtype.storm.generated.TopologySummary;  
15.   
16. public class Logic \{  
17.     /\*\* 
18.      \* 取得某个拓扑的异常个数 
19.      \* @param topoName 
20.      \* @return 
21.      \*/  
22.     public static String getTopoExpSize(String topoName) \{  
23.         ClientInfo client = null;  
24.         int errorTotal = 0;  
25.           
26.         try \{  
27.             client = ClientManager.getClient(Main.NIMBUS\_HOST, Main.NIMBUS\_PORT);  
28.               
29.             ClusterSummary clusterSummary = client.getClient().getClusterInfo();  
30.             List<TopologySummary> topoSummaryList = clusterSummary.getTopologies();  
31.             for (TopologySummary ts : topoSummaryList) \{  
32.                 if (ts.getName().equals(topoName)) \{  
33.                     TopologyInfo topologyInfo = client.getClient().getTopologyInfo(ts.getId());  
34.                     Set<String> errorKeySet = topologyInfo.getErrors().keySet();  
35.                     for (String errorKey : errorKeySet) \{  
36.                         List<ErrorInfo> listErrorInfo = topologyInfo.getErrors().get(errorKey);  
37.                         errorTotal += listErrorInfo.size();  
38.                     \}  
39.                     break;  
40.                 \}  
41.             \}  
42.               
43.             return String.valueOf(errorTotal);  
44.         \} catch (Exception e) \{  
45.             return "-1";  
46.         \} finally \{  
47.             ClientManager.closeClient(client);  
48.         \}     
49.     \}  
50.   
51.     /\*\* 
52.      \* 返回某个拓扑的异常堆栈 
53.      \* @param topoName 
54.      \* @return 
55.      \*/  
56.     public static String getTopoExpStackTrace(String topoName) \{  
57.         ClientInfo client = null;  
58.         StringBuilder error = new StringBuilder();  
59.           
60.         try \{  
61.             client = ClientManager.getClient(Main.NIMBUS\_HOST, Main.NIMBUS\_PORT);  
62.               
63.             ClusterSummary clusterSummary = client.getClient().getClusterInfo();  
64.             List<TopologySummary> topoSummaryList = clusterSummary.getTopologies();  
65.             for (TopologySummary ts : topoSummaryList) \{  
66.                 if (ts.getName().equals(topoName)) \{  
67.                     TopologyInfo topologyInfo = client.getClient().getTopologyInfo(ts.getId());  
68.                     // 得到错误信息  
69.                     Set<String> errorKeySet = topologyInfo.getErrors().keySet();  
70.                     for (String errorKey : errorKeySet) \{  
71.                         List<ErrorInfo> listErrorInfo = topologyInfo.getErrors().get(errorKey);  
72.                         for (ErrorInfo ei : listErrorInfo) \{  
73.                             // 发生异常的时间  
74.                             long expTime = (long) ei.getError\_time\_secs() \* 1000;  
75.                             // 现在的时间  
76.                             long now = System.currentTimeMillis();  
77.                               
78.                             // 由于获取的是全量的错误堆栈，我们可以设置一个范围来获取指定范围的错误，看情况而定  
79.                             // 如果超过5min，那么就不用记录了，因为5min检查一次  
80.                             if (now - expTime > 1000 \* 60 \* 5) \{  
81.                                 continue;  
82.                             \}  
83.                               
84.                             error.append(new Date(expTime) + "\\n");  
85.                             error.append(ei.getError() + "\\n");  
86.                         \}  
87.                     \}  
88.                       
89.                     break;  
90.                 \}  
91.             \}  
92.               
93.             return error.toString().isEmpty() ? "none" : error.toString();  
94.         \} catch (Exception e) \{  
95.             return "-1";  
96.         \} finally \{  
97.             ClientManager.closeClient(client);  
98.         \}  
99.     \}  
100. \}

最后打成一个Jar包，就可以跑起来接入监控系统了，如在Zabbix中，可以把各个监控项设置为自定义的item，在Zabbix Client中配置命令行来运行Jar取得数据。

接下来的测试过程先略过。

对于Storm监控的实践，目前就是这样了。

[view plain]: http://blog.csdn.net/damacheng/article/details/42706605#