Flink使用arthas在线诊断flink的那些事

Posted 2022-08-08 九师兄

1.概述

转载：使用arthas在线诊断flink的那些事

最近在使用arthas诊断工具, 诊断java服务的一些问题, 突然想到能不能使用arthas诊断flink的jobManager和taskManager呢? 答案是可以的.
采用javaagent, 在flink启动jobmanager和taskManager的时候, 注入我们的agent

官方文档

下载tunnel-server的jar包

使用的是Arthas Tunnel方式.

第一步: 先启动一个 tunnel-server服务, 
在服务器:10.0.16.1 执行 
java -jar -Dserver.port=4433 -Darthas.server.port=7766  arthas-tunnel-server-3.5.3-fatjar.jar

其中 -Dserver.port 指定web页面访问地址,-Darthas.server.port为监听端口, 远程的arthas-agent通过该端口连接

第二步,为flink程序在启动时注入arthas-agent,
我们需要使用javaagent, 
2.1 新建一个maven项目, 引入arthas-agent的依赖包
       <dependency>
            <groupId>com.taobao.arthas</groupId>
            <artifactId>arthas-agent-attach</artifactId>
            <version>3.5.3</version>
        </dependency>
        <dependency>
            <groupId>com.taobao.arthas</groupId>
            <artifactId>arthas-packaging</artifactId>
            <version>3.5.3</version>
        </dependency>

2.2 编写AgentPremain类,并且创建
   public static void premain(String agentArgs, Instrumentation inst)
        //arthasConfigMap为配置参数
        Map<String, String> arthasConfigMap = new HashMap<>();
        arthasConfigMap.put("arthas.tunnelServer","ws://10.0.16.1:7766/ws");
        arthasConfigMap.put("arthas.appName","flink:68480");
        arthasConfigMap.put("arthas.httpPort","-1"); //表示不使用httpPort
        arthasConfigMap.put("arthas.telnetPort","0"); //设置为0,表示自动获取,防止出现端口占用
        arthasConfigMap.put("arthas.agentId","xxxxxxxxxx"+uuidString);
        //上述可以通过-D传入, 比如下文的-Dagent.arthas-tunnelServer, 然后使用   
        //System.getProperty("agent.arthas-tunnelServer") 获取, 相对更灵活
		ArthasAgent arthasAgent = new ArthasAgent(arthasConfigMap, null, false, null);
        arthasAgent.init();
  

2.3 将编写javaagent项目打好jar包  customize-arthas-agent-1.0.0.jar, 
     并且将jar包放入到yarn集群下的所有节点上, 比如目录为/data/arthas/
2.4 在启动flink时, 注入这部分命令-yD yarn.container-start-command-template="" 源码中有或者看我以往的文章
flink run -m yarn-cluster -ys 1 -ynm flink-yarn-name \\
-yD state.checkpoints.dir=hdfs://clife/checkpoint/flink \\
-yD state.savepoints.dir=hdfs://clife/checkpoint/flink \\
-yjm 1G -ytm 2G -yqu default  \\
-yD execution.checkpointing.externalized-checkpoint-retention=RETAIN_ON_CANCELLATION   \\
-yD table.exec.source.idle-timeout=10s    \\
-yD yarn.tags=68480   \\
-yD state.backend=rocksdb  \\
-yD state.backend.incremental=true  \\
-yD yarn.container-start-command-template="%java% %jvmmem% %jvmopts% %logging% \\
-javaagent:/data/arthas/customize-arthas-agent-1.0.0.jar \\
-Dagent.arthasEnabled=true -Dagent.arthas-appName=flink:68480 \\
-Dagent.arthas-tunnelServer=ws://10.0.16.1:7766/ws  \\
 %class% %args% %redirects%"   \\
-yD env.java.opts="-Dfile.encoding=UTF-8"   \\
-c xx.Main  \\
/xxx.jar
  

最后, 即可在页面中连接arthas进行在线诊断了

我们改造了arthas部分代码, 目前可以筛选出指定flink任务的arthas
如果web-console出现不能复制粘贴时, 可以使用ctrl+insert 或者 ctrl+shift+c 来复制选中, shift+insert 或者 ctrl+shift+v 来粘贴