怎样实现Dolphinscheduler YARN Task状态跟踪？

背景

Dolphinscheduler针对YARN使命，好比说MR、Spark、Flink，甚至是Shell使命，最初都是会判定假如有YARN使命，剖析到applicationId。这样就会不但单以判定客户端进程为单一判定依据，还要根据YARN状态举行最终的Dolphinscheduler使命状态判定。后期，社区对此举行了重构(确实是好的向往，如今已经是半成品)，但是导致了一些问题，好比说针对Flink Stream Application模式，这种客户端分离模式会让客户端Shell直接退出，所以如今Dolphinscheduler里面的使命就直接乐成了。YARN上的使命还在运行呢，但Dolphinscheduler已经不能追踪到YARN上使命的状态了。
那么，想要实现对于YARN上使命的状态跟踪，可以怎么做呢？

注：以3.2.1版本为例。

Worker Task关系图

起首，让我们来看下DolphinScheduler中Worker Task的关系原理。

• AbstractTask: 主要界说了Task的根本生命周期接口，好比说init、handle和cancel
  
• AbstractRemoteTask: 主要对handle方法做了实现，表现了模版方法设计模式，提取了submitApplication、trackApplicationStatus以及cancelApplication三个核心接口方法
  
• AbstractYarnTask: 好比说YARN使命，就抽象了AbstractYarnTask，此中submitApplication、trackApplicationStatus以及cancelApplication可以直接是对YARN API的访问
  

AbstractYarnTask实现YARN状态跟踪

AbstractYarnTask可以实现YARN状态跟踪，参考org.apache.dolphinscheduler.plugin.task.api.AbstractYarnTask，完整代码如下 :

1. public abstract class AbstractYarnTask extends AbstractRemoteTask {
3.     private static final int MAX_RETRY_ATTEMPTS = 3;
5.     private ShellCommandExecutor shellCommandExecutor;
7.     public AbstractYarnTask(TaskExecutionContext taskRequest) {
8.         super(taskRequest);
9.         this.shellCommandExecutor = new ShellCommandExecutor(this::logHandle, taskRequest);
10.     }
12.     @Override
13.     public void submitApplication() throws TaskException {
14.         try {
15.             IShellInterceptorBuilder shellActuatorBuilder =
16.                     ShellInterceptorBuilderFactory.newBuilder()
17.                             .properties(getProperties())
18.                             // todo: do we need to move the replace to subclass?
19.                             .appendScript(getScript().replaceAll("\\r\\n", System.lineSeparator()));
20.             // SHELL task exit code
21.             TaskResponse response = shellCommandExecutor.run(shellActuatorBuilder, null);
22.             setExitStatusCode(response.getExitStatusCode());
23.             setAppIds(String.join(TaskConstants.COMMA, getApplicationIds()));
24.             setProcessId(response.getProcessId());
25.         } catch (InterruptedException ex) {
26.             Thread.currentThread().interrupt();
27.             log.info("The current yarn task has been interrupted", ex);
28.             setExitStatusCode(TaskConstants.EXIT_CODE_FAILURE);
29.             throw new TaskException("The current yarn task has been interrupted", ex);
30.         } catch (Exception e) {
31.             log.error("yarn process failure", e);
32.             exitStatusCode = -1;
33.             throw new TaskException("Execute task failed", e);
34.         }
35.     }
37.     @Override
38.     public void trackApplicationStatus() throws TaskException {
39.         if (StringUtils.isEmpty(appIds)) {
40.             return;
41.         }
44.         List<String> appIdList = Arrays.asList(appIds.split(","));
45.         boolean continueTracking = true;
47.         while (continueTracking) {
48.             Map<String, YarnState> yarnStateMap = new HashMap<>();
49.             for (String appId : appIdList) {
50.                 if (StringUtils.isEmpty(appId)) {
51.                     continue;
52.                 }
54.                 boolean hadoopSecurityAuthStartupState =
55.                         PropertyUtils.getBoolean(HADOOP_SECURITY_AUTHENTICATION_STARTUP_STATE, false);
56.                 String yarnStateJson = fetchYarnStateJsonWithRetry(appId, hadoopSecurityAuthStartupState);
57.                 if (StringUtils.isNotEmpty(yarnStateJson)) {
58.                     String appJson = JSONUtils.getNodeString(yarnStateJson, "app");
59.                     YarnTask yarnTask = JSONUtils.parseObject(appJson, YarnTask.class);
60.                     log.info("yarnTask : {}", yarnTask);
61.                     yarnStateMap.put(yarnTask.getId(), YarnState.of(yarnTask.getState()));
62.                 }
63.             }
65.             YarnState yarnTaskOverallStatus = YarnTaskStatusChecker.getYarnTaskOverallStatus(yarnStateMap);
66.             if (yarnTaskOverallStatus.isFinalState()) {
67.                 handleFinalState(yarnTaskOverallStatus);
68.                 continueTracking = false;
69.             } else {
70.                 try {
71.                     TimeUnit.MILLISECONDS.sleep(SLEEP_TIME_MILLIS * 10);
72.                 } catch (InterruptedException e) {
73.                     Thread.currentThread().interrupt();
74.                     throw new RuntimeException(e);
75.                 }
76.             }
77.         }
78.     }
80.     private String fetchYarnStateJsonWithRetry(String appId,
81.                                                boolean hadoopSecurityAuthStartupState) throws TaskException {
82.         int retryCount = 0;
83.         while (retryCount < MAX_RETRY_ATTEMPTS) {
84.             try {
85.                 return fetchYarnStateJson(appId, hadoopSecurityAuthStartupState);
86.             } catch (Exception e) {
87.                 retryCount++;
88.                 log.error("Failed to fetch or parse Yarn state for appId: {}. Attempt: {}/{}",
89.                         appId, retryCount, MAX_RETRY_ATTEMPTS, e);
91.                 if (retryCount >= MAX_RETRY_ATTEMPTS) {
92.                     throw new TaskException("Failed to fetch Yarn state after "
93.                             + MAX_RETRY_ATTEMPTS + " attempts for appId: " + appId, e);
94.                 }
96.                 try {
97.                     TimeUnit.MILLISECONDS.sleep(SLEEP_TIME_MILLIS);
98.                 } catch (InterruptedException ie) {
99.                     Thread.currentThread().interrupt();
100.                     throw new RuntimeException(ie);
101.                 }
102.             }
103.         }
104.         return null;
105.     }
107.     private void handleFinalState(YarnState yarnState) {
108.         switch (yarnState) {
109.             case FINISHED:
110.                 setExitStatusCode(EXIT_CODE_SUCCESS);
111.                 break;
112.             case KILLED:
113.                 setExitStatusCode(EXIT_CODE_KILL);
114.                 break;
115.             default:
116.                 setExitStatusCode(EXIT_CODE_FAILURE);
117.                 break;
118.         }
119.     }
121.     private String fetchYarnStateJson(String appId, boolean hadoopSecurityAuthStartupState) throws Exception {
122.         return hadoopSecurityAuthStartupState
123.                 ? KerberosHttpClient.get(getApplicationUrl(appId))
124.                 : HttpUtils.get(getApplicationUrl(appId));
125.     }
128.     static class YarnTaskStatusChecker {
130.         public static YarnState getYarnTaskOverallStatus(Map<String, YarnState> yarnTaskMap) {
131.             // 检查是否有任何任务处于 FAILED 或 KILLED 状态
132.             boolean hasKilled = yarnTaskMap.values().stream()
133.                     .anyMatch(state -> state == YarnState.KILLED);
135.             if (hasKilled) {
136.                 return YarnState.KILLED;
137.             }
139.             // 检查是否有任何任务处于 FAILED 或 KILLED 状态
140.             boolean hasFailed = yarnTaskMap.values().stream()
141.                     .anyMatch(state -> state == YarnState.FAILED);
143.             if (hasFailed) {
144.                 return YarnState.FAILED;
145.             }
148.             // 检查是否所有任务都处于 FINISHED 状态
149.             boolean allFINISHED = yarnTaskMap.values().stream()
150.                     .allMatch(state -> state == YarnState.FINISHED);
152.             if (allFINISHED) {
153.                 return YarnState.FINISHED;
154.             }
156.             // 检查是否有任何任务处于 RUNNING 状态
157.             boolean hasRunning = yarnTaskMap.values().stream()
158.                     .anyMatch(state -> state == YarnState.RUNNING);
160.             if (hasRunning) {
161.                 return YarnState.RUNNING;
162.             }
164.             // 检查是否有任何任务处于提交中状态
165.             boolean hasSubmitting = yarnTaskMap.values().stream()
166.                     .anyMatch(state -> state == YarnState.NEW || state == YarnState.NEW_SAVING
167.                             || state == YarnState.SUBMITTED || state == YarnState.ACCEPTED);
169.             if (hasSubmitting) {
170.                 return YarnState.SUBMITTING;
171.             }
173.             // 如果都不匹配，返回未知状态
174.             return YarnState.UNKNOWN;
175.         }
176.     }
179.     /**
180.      * cancel application
181.      *
182.      * @throws TaskException exception
183.      */
184.     @Override
185.     public void cancelApplication() throws TaskException {
186.         // cancel process
187.         try {
188.             shellCommandExecutor.cancelApplication();
189.         } catch (Exception e) {
190.             throw new TaskException("cancel application error", e);
191.         }
192.     }
194.     /**
195.      * get application ids
196.      *
197.      * @return
198.      * @throws TaskException
199.      */
200.     @Override
201.     public List<String> getApplicationIds() throws TaskException {
202.         // TODO 这里看common.properties中是否配置 appId.collect了，如果配置了走aop，否则走log
203.         return LogUtils.getAppIds(
204.                 taskRequest.getLogPath(),
205.                 taskRequest.getAppInfoPath(),
206.                 PropertyUtils.getString(APPID_COLLECT, DEFAULT_COLLECT_WAY));
207.     }
209.     /** Get the script used to bootstrap the task */
210.     protected abstract String getScript();
212.     /** Get the properties of the task used to replace the placeholders in the script. */
213.     protected abstract Map<String, String> getProperties();
215.     @Data
216.     static class YarnTask {
217.         private String id;
218.         private String state;
219.     }
221.     private String getApplicationUrl(String applicationId) throws BaseException {
223.         String yarnResourceRmIds = PropertyUtils.getString(YARN_RESOURCEMANAGER_HA_RM_IDS);
224.         String yarnAppStatusAddress = PropertyUtils.getString(YARN_APPLICATION_STATUS_ADDRESS);
225.         String hadoopResourceManagerHttpAddressPort =
226.                 PropertyUtils.getString(HADOOP_RESOURCE_MANAGER_HTTPADDRESS_PORT);
228.         String appUrl = StringUtils.isEmpty(yarnResourceRmIds) ?
229.                 yarnAppStatusAddress :
230.                 getAppAddress(yarnAppStatusAddress, yarnResourceRmIds);
232.         if (StringUtils.isBlank(appUrl)) {
233.             throw new BaseException("yarn application url generation failed");
234.         }
236.         log.info("yarn application url:{}", String.format(appUrl, hadoopResourceManagerHttpAddressPort, applicationId));
237.         return String.format(appUrl, hadoopResourceManagerHttpAddressPort, applicationId);
238.     }
240.     private static String getAppAddress(String appAddress, String rmHa) {
242.         String[] appAddressArr = appAddress.split(Constants.DOUBLE_SLASH);
244.         if (appAddressArr.length != 2) {
245.             return null;
246.         }
248.         String protocol = appAddressArr[0] + Constants.DOUBLE_SLASH;
249.         String[] pathSegments = appAddressArr[1].split(Constants.COLON);
251.         if (pathSegments.length != 2) {
252.             return null;
253.         }
255.         String end = Constants.COLON + pathSegments[1];
257.         // get active ResourceManager
258.         String activeRM = YarnHAAdminUtils.getActiveRMName(protocol, rmHa);
260.         if (StringUtils.isEmpty(activeRM)) {
261.             return null;
262.         }
264.         return protocol + activeRM + end;
265.     }
267.     /** yarn ha admin utils */
268.     private static final class YarnHAAdminUtils {
270.         /**
271.          * get active resourcemanager node
272.          *
273.          * @param protocol http protocol
274.          * @param rmIds yarn ha ids
275.          * @return yarn active node
276.          */
277.         public static String getActiveRMName(String protocol, String rmIds) {
279.             String hadoopResourceManagerHttpAddressPort =
280.                     PropertyUtils.getString(HADOOP_RESOURCE_MANAGER_HTTPADDRESS_PORT);
282.             String[] rmIdArr = rmIds.split(Constants.COMMA);
284.             String yarnUrl = protocol
285.                     + "%s:"
286.                     + hadoopResourceManagerHttpAddressPort
287.                     + "/ws/v1/cluster/info";
288.             try {
289.                 /** send http get request to rm */
290.                 for (String rmId : rmIdArr) {
291.                     String state = getRMState(String.format(yarnUrl, rmId));
292.                     if (Constants.HADOOP_RM_STATE_ACTIVE.equals(state)) {
293.                         return rmId;
294.                     }
295.                 }
297.             } catch (Exception e) {
298.                 log.error("get yarn ha application url failed", e);
299.             }
300.             return null;
301.         }
303.         /** get ResourceManager state */
304.         public static String getRMState(String url) {
305.             boolean hadoopSecurityAuthStartupState =
306.                     PropertyUtils.getBoolean(HADOOP_SECURITY_AUTHENTICATION_STARTUP_STATE, false);
307.             String retStr = Boolean.TRUE.equals(hadoopSecurityAuthStartupState)
308.                     ? KerberosHttpClient.get(url)
309.                     : HttpUtils.get(url);
311.             if (StringUtils.isEmpty(retStr)) {
312.                 return null;
313.             }
314.             // to json
315.             ObjectNode jsonObject = JSONUtils.parseObject(retStr);
317.             // get ResourceManager state
318.             if (!jsonObject.has("clusterInfo")) {
319.                 return null;
320.             }
321.             return jsonObject.get("clusterInfo").path("haState").asText();
322.         }
323.     }
325.     public enum YarnState {
326.         NEW,
327.         NEW_SAVING,
328.         SUBMITTED,
329.         ACCEPTED,
330.         RUNNING,
331.         FINISHED,
332.         FAILED,
333.         KILLED,
334.         SUBMITTING,
335.         UNKNOWN,
336.         ;
338.         // 将字符串转换为枚举
339.         public static YarnState of(String state) {
340.             try {
341.                 return YarnState.valueOf(state);
342.             } catch (IllegalArgumentException | NullPointerException e) {
343.                 // 如果字符串无效，则返回 null
344.                 return null;
345.             }
346.         }
348.         /**
349.          * 任务结束
350.          * @return
351.          */
352.         public boolean isFinalState() {
353.             return this == FINISHED || this == FAILED || this == KILLED;
354.         }
355.     }
356. }

复制代码

可以看到，这里的核心逻辑其实就是去掉之前直接把handle接口重写了，而如今针对YARN使命，只需要实现submitApplication、trackApplicationStatus两个核心接口，cancelApplication这个其实原则上应该代理YarnApplicationManager才好(当前没有整合，不过不影响)。
流式使命前端applicationId显示

dolphinscheduler-ui/src/views/projects/task/instance/use-stream-table.ts

后端封装applicationId为YARN URL

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskInstanceServiceImpl.java 修改

dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/constants/Constants.java 修改

dolphinscheduler-common/src/main/resources/common.properties修改

dolphinscheduler-storage-plugin/dolphinscheduler-storage-hdfs/src/main/java/org/apache/dolphinscheduler/plugin/storage/hdfs/HdfsStorageOperator.java修改

dolphinscheduler-storage-plugin/dolphinscheduler-storage-hdfs/src/main/java/org/apache/dolphinscheduler/plugin/storage/hdfs/HdfsStorageProperties.java修改

页面效果如下 :

注意 : URL粘贴是需要自己写的，上面的代码不包罗

问题追踪

这里其实是有问题，对于state状态来说，是有FINISHED、FAILED、KILLED三种状态，但是FINISHED状态里面还是有FinalStatus，完成不一定是乐成，FINISHED下面其实也有SUCCEEDED、FAILED和KILLED。其实就是FINISHED不能作为DolphinScheduler的终态，需要继承判定而已。
org.apache.dolphinscheduler.plugin.task.api.AbstractYarnTask#handleFinalState

1. private void handleFinalState(YarnState yarnState) {
2.     switch (yarnState) {
3.         case FINISHED:
4.             setExitStatusCode(EXIT_CODE_SUCCESS);
5.             break;
6.         case KILLED:
7.             setExitStatusCode(EXIT_CODE_KILL);
8.             break;
9.         default:
10.             setExitStatusCode(EXIT_CODE_FAILURE);
11.             break;
12.     }
13. }

复制代码

使用HTTP对使命举行kill

1. curl -X PUT -d '{"state":"KILLED"}' \
2. >     -H "Content-Type: application/json" \
3. >     http://xx.xx.xx.xx:8088/ws/v1/cluster/apps/application_1694766249884_1098/state?user.name=hdfs

复制代码

注意 ： 一定要指定user.name，否则不一定能kill掉。

原文链接：https://segmentfault.com/a/1190000045058893

本文由 白鲸开源 提供发布支持！

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。