实践应用性能管理工具使用系列二：Pinpoint 支持采集性能数据输出到kafka

IT那活儿 2023-01-31

2012

点击上方“IT那活儿”公众号，关注后了解更多内容，不管IT什么活儿，干就完了！！！

Pinpoint应用性能管理工具将采集的大量的应用性能数据存入到hbase数据库，当我们面对一些实时业务场景需求，需要对采集的性能数据进行二次加工处理统计的时候，只能从hbase里面查询出对应的数据再做加工，这种数据处理方式存在实效性无法保障的问题。

从pinpoint 原理图可以看出Pinpoint Collector负责接收pinpoint agent 采集过来性能数据后，会将性能数据存入到hbase，我们可以对Collector进行二次开发改造让其在写hbase 同时将数据吐一份到kafka ，再利用大数据组件技术对接kafka的方式保障业务场景实效性。

我们以pinpoint 2.3.3 版本为例，介绍如何改造Collector 支持性能数据吐到kafka：

1. 先了解pinpoint collector采集了哪一些信息

协议	端口	采集内容
thrift	collector.receiver.base.port=9994	采集agent配置信息
grpc	collector.receiver.grpc.agent.bindaddress.port=9991
thrift	collector.receiver.stat.tcp.port=9995 collector.receiver.stat.udp.port=9995	采集agent jvm 性能数据
grpc	collector.receiver.grpc.stat.bindaddress.port=9992
thrift	agent: profiler.collector.span.port=9996 collector: collector.receiver.span.udp.port=9996 collector.receiver.span.tcp.port=9996	采集agent span报文
grpc	collector.receiver.grpc.span.bindaddress.port=9993

2. 源代码编译

下载pinpoint 源代码导入到idea开发具，并对源代码进行编译，需要解决相关依赖包报错问题后，才能对collector代码进行修改，可以参考之前发布公众号文章《应用性能管理工具使用系列一：Pinpoint代码编译安装》。

3. 加入kafka相关依赖包

collector 子工程pom.xml 文件加入配置依赖项。

<dependency>
  <groupId>org.springframework.kafka</groupId>
  <artifactId>spring-kafka</artifactId>
  <version>2.6.7</version>
</dependency>
     <dependency>
  <groupId>com.alibaba</groupId>
  <artifactId>fastjson</artifactId>
  <version>1.2.47</version>
</dependency>

4. 修改配置文件application.yml 支持不同采集内容入到不同kafka topic

kafka:
producer:
bootstrap-servers: 192.168.XXX.104:9098,192.168.XXX.102:9098
kafka:
agent-info:
topic: agent-info   #写入topic 名称
disabled: true #是否禁用 true 禁用 false 启用

5. 现有性能数据入hbase代码逻辑分析

通过查阅及debug collector 源代码，可以看出pinpoint 根据不同协议(thrift/grpc)将采集上来的性能数据入到对应的hbase表。

我们以thrift协议采集agent配置信息为例，看一下数据是如何流转写入到hbase 表，打开ThriftAgentInfoHandler 类，当接收到agent 端推送过来agent配置信息时，最终是通过如下方法将性能数据写入到hbase AgentInfo表。

6. agent配置信息加入支持写kafka

在现有写入到hbase 表方法下面，加入写kafka 代码，并且将接收到信息转为json字符串方式发送到kafka 。

修改代码后将collector 工程重新编译部署pinpoint-collector-boot-2.3.3.jar，可以将修改后的application.yml 放到pinpoint-collector-boot-2.3.3.jar同一级目录下，方便根据实际环境修改kafka 配置信息。

7. 验证agent配置信信息写入kafka

进入kafka/bin 下消费topic 数据.

kafka-console-consumer.sh --bootstrap-server 192.168.XXX.102:9098 --topic agent-info
{"agentId":"192.168.XXX.101-
8081","agentName":"","agentVersion":"2.3.3","applicationName":"snc-datacollection","container":false,"endStatus":0,"endTimeStam
p":0,"hostName":"bigdata-
02","ip":"192.168.XXX.101","jvmInfo":
{"gcTypeName":"PARALLEL","jvmVersion":"1.8.0_201","version":0},"pid":979789,"ports":"","serverMetaData":
{"serverInfo":"Apache Tomcat/9.0.30","serviceInfos":
[{"serviceLibs":[],"serviceName":"Tomcat/localhost/snc-datacollection"}],"vmArgs":["-Xms1g","-Xmx1g","-
XX:+HeapDumpOnOutOfMemoryError","-javaagent:/home/pinpoint/pinpoint-agent-2.3.3/pinpoint-
bootstrap-2.3.3.jar","-Dpinpoint.agentId=192.168.XXX.101-
8081","-Dpinpoint.applicationName=snc-datacollection","-Dpinpoint.licence=8310000"]},"serviceTypeCode":1000,"startTi
me":1672902590224,"vmVersion":"1.8.0_201"}