diff --git a/deploy/kubernetes/dolphinscheduler/README.md b/deploy/kubernetes/dolphinscheduler/README.md index 69b5590a0e4f..99de38ec818a 100644 --- a/deploy/kubernetes/dolphinscheduler/README.md +++ b/deploy/kubernetes/dolphinscheduler/README.md @@ -203,6 +203,9 @@ The paths above match the API and Worker server classpaths. Check the configurat | conf.common."resource.storage.type" | string | `"S3"` | resource storage type: HDFS, S3, OSS, GCS, ABS, NONE | | conf.common."resource.storage.upload.base.path" | string | `"/dolphinscheduler"` | resource store on HDFS/S3 path, resource file will store to this base path, self configuration, please make sure the directory exists on hdfs and have read write permissions. "/dolphinscheduler" is recommended | | conf.common."shell.kill.wait.timeout" | int | `10` | If the shell process is still active after this timeout value (in seconds), then will use kill -9 to kill it | +| conf.common."sql.task.alert.max.content.bytes" | int | `60000` | max serialized UTF-8 bytes of the SQL task-result alert content, kept below the 65535-byte capacity of the MySQL TEXT column | +| conf.common."sql.task.alert.max.row.value.length" | int | `100` | max chars of a single string value kept in the SQL task-result alert content, longer values are truncated with an ellipsis | +| conf.common."sql.task.alert.max.rows" | int | `30` | max rows kept in the SQL task-result alert content, rows beyond this are replaced by an omission marker | | conf.common."sudo.enable" | bool | `true` | use sudo or not, if set true, executing user is tenant user and deploy user needs sudo permissions; if set false, executing user is the deploy user and doesn't need sudo permissions | | conf.common."support.hive.oneSession" | bool | `false` | Whether hive SQL is executed in the same session | | conf.common."task.resource.limit.state" | bool | `false` | Task resource limit state | diff --git a/deploy/kubernetes/dolphinscheduler/values.yaml b/deploy/kubernetes/dolphinscheduler/values.yaml index 048da5aa7f99..8f168534e4c1 100644 --- a/deploy/kubernetes/dolphinscheduler/values.yaml +++ b/deploy/kubernetes/dolphinscheduler/values.yaml @@ -393,6 +393,15 @@ conf: # -- way to collect applicationId: log, aop appId.collect: log + # -- max rows kept in the SQL task-result alert content, rows beyond this are replaced by an omission marker + sql.task.alert.max.rows: 30 + + # -- max chars of a single string value kept in the SQL task-result alert content, longer values are truncated with an ellipsis + sql.task.alert.max.row.value.length: 100 + + # -- max serialized UTF-8 bytes of the SQL task-result alert content, kept below the 65535-byte capacity of the MySQL TEXT column + sql.task.alert.max.content.bytes: 60000 + common: ## Configmap configmap: diff --git a/docs/docs/en/architecture/configuration.md b/docs/docs/en/architecture/configuration.md index cb37799a853e..5b5c128ba408 100644 --- a/docs/docs/en/architecture/configuration.md +++ b/docs/docs/en/architecture/configuration.md @@ -223,6 +223,9 @@ The default configuration is as follows: | sudo.enable | true | whether to enable sudo | | zeppelin.rest.url | http://localhost:8080 | the RESTful API url of zeppelin | | appId.collect | log | way to collect applicationId, if use aop, alter the configuration from log to aop, annotation of applicationId auto collection related configuration in `bin/env/dolphinscheduler_env.sh` should be removed. Note: Aop way doesn't support submitting yarn job on remote host by client mode like Beeline, and will failure if override applicationId collection-related environment configuration in dolphinscheduler_env.sh, and . | +| sql.task.alert.max.rows | 30 | max rows kept in the SQL task-result alert content, rows beyond this are replaced by an omission marker. Takes effect after restarting Worker Server | +| sql.task.alert.max.row.value.length | 100 | max chars of a single string value kept in the SQL task-result alert content, longer values are truncated with an ellipsis. Takes effect after restarting Worker Server | +| sql.task.alert.max.content.bytes | 60000 | max serialized UTF-8 bytes of the SQL task-result alert content, kept below the 65535-byte capacity of the MySQL TEXT column that stores alert content. Takes effect after restarting Worker Server | ### Api-server related configuration diff --git a/docs/docs/en/guide/upgrade/incompatible.md b/docs/docs/en/guide/upgrade/incompatible.md index 650b52df380d..2b6649dad2ed 100644 --- a/docs/docs/en/guide/upgrade/incompatible.md +++ b/docs/docs/en/guide/upgrade/incompatible.md @@ -57,3 +57,8 @@ This document records the incompatible updates between each version. You need to * **Removed derived properties**: `cmdTypeIfComplement`, `complementData` (related to complement-data executions; use the detail API to obtain them) * To obtain any of these fields, use the detail API `GET /projects/{projectCode}/workflow-instances/{id}` instead, which continues to return the full `WorkflowInstance` object. ([#18444](https://github.com/apache/dolphinscheduler/pull/18444)) +## 3.5.0 + +* `AlertSender#getAlertData` now explicitly rejects a null `alertType` by throwing `IllegalArgumentException` with an error log, instead of throwing a raw `NullPointerException`. During a rolling upgrade, **the Alert Server must be upgraded before Master/Worker**, otherwise an old Alert Server cannot recognize newly introduced `alert_type` enum values (e.g. `TASK_RESULT`) and alert delivery will fail. See the [upgrade guide](upgrade.md#rolling-upgrade-order) for details.([#18549](https://github.com/apache/dolphinscheduler/pull/18549)) +* SQL task-result alert content is size-limited: by default at most 30 rows are kept, string field values longer than 100 characters are truncated with an ellipsis, and the serialized content is bounded to 60000 UTF-8 bytes (below the 65535-byte capacity of the MySQL TEXT column); rows beyond the budget are represented by an omission marker. These defaults are tunable via `sql.task.alert.max.rows`, `sql.task.alert.max.row.value.length` and `sql.task.alert.max.content.bytes` in `common.properties` (worker restart required). ([#18549](https://github.com/apache/dolphinscheduler/pull/18549)) + diff --git a/docs/docs/en/guide/upgrade/upgrade.md b/docs/docs/en/guide/upgrade/upgrade.md index dfb3016a07d6..67b631e1febc 100644 --- a/docs/docs/en/guide/upgrade/upgrade.md +++ b/docs/docs/en/guide/upgrade/upgrade.md @@ -75,6 +75,28 @@ Execution result: - After version 3.3.X and later, we only support upgrading from 3.0.0. For versions lower than this, please download the historical version and upgrade to 3.0.0. - After version 3.3.X and later, binary packages no longer provide plugins dependencies by default, so when you use them for the first time, you need to download and install them yourself. For more information, please refer to [Pseudo-Cluster](../installation/pseudo-cluster.md). +#### Rolling Upgrade Order + +When performing a rolling upgrade, **the Alert Server must be upgraded before Master/Worker**. + +Starting from 3.5.0, the Master may persist new `alert_type` enum values (e.g. `TASK_RESULT`) into the `t_ds_alert` table. If the Alert Server has not yet been upgraded to a version that includes the new enum value, MyBatis will deserialize the unknown value as `null`, causing `AlertSender` to throw a `NullPointerException` while building the alert data. The alert will remain stuck in `WAIT_EXECUTION` and never be delivered. + +Upgrading in the order "Alert Server → Master → Worker" eliminates this risk. + +#### If the upgrade fails + +If the upgrade order was not followed (e.g. Master/Worker has already been upgraded before the Alert Server) and new `alert_type` values have been persisted to the `t_ds_alert` table, follow these steps to recover: + +1. Upgrade the Alert Server to a version that includes the new enum values and restart it. Alerts stuck in `WAIT_EXECUTION` will be picked up and delivered normally. +2. If the Alert Server cannot be upgraded immediately, stop the old Alert Server first, then back up and clean up the affected alerts so they cannot keep blocking alert delivery: + + ```sql + -- 1. back up the affected alerts + create table t_ds_alert_bak as select * from t_ds_alert where alert_type = 8; + -- 2. remove them from the active table + delete from t_ds_alert where alert_type = 8; + ``` + #### Precautions after the upgrade The alert plugin may have some dirty data. After the upgrade, clear it manually by referring to SQL. diff --git a/docs/docs/zh/architecture/configuration.md b/docs/docs/zh/architecture/configuration.md index a2b94379555b..00012cf0eadd 100644 --- a/docs/docs/zh/architecture/configuration.md +++ b/docs/docs/zh/architecture/configuration.md @@ -223,6 +223,9 @@ common.properties配置文件目前主要是配置hadoop/s3/yarn/applicationId | sudo.enable | true | 是否开启sudo | | zeppelin.rest.url | http://localhost:8080 | zeppelin RESTful API 接口地址 | | appId.collect | log | 收集applicationId方式， 如果用aop方法，将配置log替换为aop，并将`bin/env/dolphinscheduler_env.sh`自动收集applicationId相关环境变量配置的注释取消掉，注意：aop不支持远程主机提交yarn作业的方式比如Beeline客户端提交，且如果用户环境覆盖了dolphinscheduler_env.sh收集applicationId相关环境变量配置，aop方法会失效 | +| sql.task.alert.max.rows | 30 | SQL 任务结果告警内容最多保留的行数，超出的行以省略标记表示。修改后需重启 Worker Server 生效 | +| sql.task.alert.max.row.value.length | 100 | SQL 任务结果告警内容中单个字符串字段值最多保留的字符数，超出的值以省略号截断。修改后需重启 Worker Server 生效 | +| sql.task.alert.max.content.bytes | 60000 | SQL 任务结果告警内容序列化后的最大 UTF-8 字节数，小于存储告警内容的 MySQL TEXT 列 65535 字节容量。修改后需重启 Worker Server 生效 | ## Api-server相关配置 diff --git a/docs/docs/zh/guide/upgrade/incompatible.md b/docs/docs/zh/guide/upgrade/incompatible.md index a0f945501989..7e3de19628b7 100644 --- a/docs/docs/zh/guide/upgrade/incompatible.md +++ b/docs/docs/zh/guide/upgrade/incompatible.md @@ -56,3 +56,8 @@ * **移除的派生属性**：`cmdTypeIfComplement`、`complementData`（补数执行相关，如需获取请使用详情接口） * 如需获取这些字段，请使用详情接口 `GET /projects/{projectCode}/workflow-instances/{id}`，该接口仍返回完整的 `WorkflowInstance` 对象 ([#18444](https://github.com/apache/dolphinscheduler/pull/18444)) +## 3.5.0 + +* `AlertSender#getAlertData` 对 null `alertType` 增加了显式拒绝（抛 `IllegalArgumentException` 并记录错误日志），而非直接抛 `NullPointerException`。滚动升级时**必须先升级 Alert Server，再升级 Master/Worker**，否则旧 Alert Server 无法识别新增的 `alert_type` 枚举值（如 `TASK_RESULT`），会导致告警投递失败。详见[升级文档](upgrade.md#滚动升降级顺序)。([#18549](https://github.com/apache/dolphinscheduler/pull/18549)) +* SQL 任务结果告警内容有大小限制：默认最多保留 30 行，超过 100 字符的字符串字段值会被截断并以省略号标记；序列化后内容总大小限制在 60000 UTF-8 字节以内（兼容 MySQL TEXT 列 65535 字节容量），超出预算的行以省略标记对象表示。以上默认值可通过 `common.properties` 中的 `sql.task.alert.max.rows`、`sql.task.alert.max.row.value.length`、`sql.task.alert.max.content.bytes` 配置调整，修改后需重启 Worker 生效。 ([#18549](https://github.com/apache/dolphinscheduler/pull/18549)) + diff --git a/docs/docs/zh/guide/upgrade/upgrade.md b/docs/docs/zh/guide/upgrade/upgrade.md index 19ccad73cab2..0e5221442ec6 100644 --- a/docs/docs/zh/guide/upgrade/upgrade.md +++ b/docs/docs/zh/guide/upgrade/upgrade.md @@ -76,6 +76,28 @@ jar 包 并添加到 `./tools/libs` 目录下，设置以下环境变量 - 在 3.3.X 以及之后的版本，我们仅支持从 3.0.0 开始进行升级，低于此版本的请下载历史版本升级至 3.0.0。 - 在 3.3.X 以及之后的版本，二进制包不再默认提供插件依赖，因此第一次使用时，需要自行下载安装。具体请参考请参照[伪集群部署(Pseudo-Cluster)](../installation/pseudo-cluster.md) +#### 滚动升降级顺序 + +在滚动升降级场景下，**必须先升级 Alert Server，再升级 Master/Worker**。 + +从 3.5.0 起，Master 可能向 `t_ds_alert` 表写入新的 `alert_type` 枚举值（如 `TASK_RESULT`）。如果 Alert Server 尚未升级到包含该枚举值的版本，MyBatis 会将未知的枚举值反序列化为 `null`，导致 `AlertSender` 在构建告警数据时抛出 `NullPointerException`，告警将卡在 `WAIT_EXECUTION` 状态无法投递。 + +按照"Alert Server → Master → Worker"的顺序升级可消除此风险。 + +#### 如果升级失败 + +如果未按上述顺序升级（例如在 Alert Server 之前先升级了 Master/Worker），且新的 `alert_type` 枚举值已写入 `t_ds_alert` 表，请按以下步骤恢复： + +1. 将 Alert Server 升级到包含新枚举值的版本并重启，卡在 `WAIT_EXECUTION` 状态的告警会被正常拉取并投递。 +2. 如果暂时无法升级 Alert Server，请先停止旧版 Alert Server，再备份并清理受影响告警，避免其持续阻塞告警投递： + + ```sql + -- 1. 备份受影响告警 + create table t_ds_alert_bak as select * from t_ds_alert where alert_type = 8; + -- 2. 从主表清理 + delete from t_ds_alert where alert_type = 8; + ``` + #### 升级后的注意事项 在历史版本中可能告警插件会有一些脏数据，升级后请参考一下 SQL 手动清理。 diff --git a/dolphinscheduler-alert/dolphinscheduler-alert-server/src/main/java/org/apache/dolphinscheduler/alert/service/AlertSender.java b/dolphinscheduler-alert/dolphinscheduler-alert-server/src/main/java/org/apache/dolphinscheduler/alert/service/AlertSender.java index 9c9cd034bdb6..2f829e5bcd92 100644 --- a/dolphinscheduler-alert/dolphinscheduler-alert-server/src/main/java/org/apache/dolphinscheduler/alert/service/AlertSender.java +++ b/dolphinscheduler-alert/dolphinscheduler-alert-server/src/main/java/org/apache/dolphinscheduler/alert/service/AlertSender.java @@ -100,6 +100,13 @@ public List getAlertPluginInstanceList(Alert event) { @Override public AlertData getAlertData(Alert event) { + if (event.getAlertType() == null) { + log.error("Alert {} has null alertType, refusing to send. This indicates the Alert Server version " + + "is older than the Master that created the alert. Please upgrade the Alert Server first " + + "(see upgrade guide: Alert Server -> Master -> Worker).", event.getId()); + throw new IllegalArgumentException( + "Cannot build AlertData with null alertType for alert " + event.getId()); + } return AlertData.builder() .id(event.getId()) .content(event.getContent()) diff --git a/dolphinscheduler-alert/dolphinscheduler-alert-server/src/test/java/org/apache/dolphinscheduler/alert/runner/AlertSenderTest.java b/dolphinscheduler-alert/dolphinscheduler-alert-server/src/test/java/org/apache/dolphinscheduler/alert/runner/AlertSenderTest.java index 18246f485ab2..2e8f5b2672d0 100644 --- a/dolphinscheduler-alert/dolphinscheduler-alert-server/src/test/java/org/apache/dolphinscheduler/alert/runner/AlertSenderTest.java +++ b/dolphinscheduler-alert/dolphinscheduler-alert-server/src/test/java/org/apache/dolphinscheduler/alert/runner/AlertSenderTest.java @@ -25,6 +25,7 @@ import static org.mockito.Mockito.when; import org.apache.dolphinscheduler.alert.api.AlertChannel; +import org.apache.dolphinscheduler.alert.api.AlertData; import org.apache.dolphinscheduler.alert.api.AlertResult; import org.apache.dolphinscheduler.alert.config.AlertConfig; import org.apache.dolphinscheduler.alert.plugin.AlertPluginManager; @@ -210,4 +211,33 @@ void testSendAlert() { .thenReturn(paramsMap); alertSender.syncTestSend(PLUGIN_DEFINE_ID, PLUGIN_INSTANCE_PARAMS); } + + @Test + void testGetAlertDataWithNullAlertType() { + Alert alert = new Alert(); + alert.setId(1); + alert.setTitle(TITLE); + alert.setContent(CONTENT); + alert.setAlertGroupId(ALERT_GROUP_ID); + alert.setWarningType(WarningType.FAILURE); + alert.setAlertType(null); + + Assertions.assertThrows(IllegalArgumentException.class, + () -> alertSender.getAlertData(alert)); + } + + @Test + void testGetAlertDataWithValidAlertType() { + Alert alert = new Alert(); + alert.setId(2); + alert.setTitle(TITLE); + alert.setContent(CONTENT); + alert.setAlertGroupId(ALERT_GROUP_ID); + alert.setWarningType(WarningType.FAILURE); + alert.setAlertType(AlertType.TASK_RESULT); + + AlertData alertData = alertSender.getAlertData(alert); + Assertions.assertNotNull(alertData); + Assertions.assertEquals(AlertType.TASK_RESULT.getCode(), alertData.getAlertType()); + } } diff --git a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/constants/Constants.java b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/constants/Constants.java index 81783ed28528..e5815701df9e 100644 --- a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/constants/Constants.java +++ b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/constants/Constants.java @@ -59,6 +59,23 @@ public final class Constants { */ public static final String SHELL_KILL_WAIT_TIMEOUT = "shell.kill.wait.timeout"; + /** + * sql.task.alert.max.rows: this property defines the max rows kept in the task-result alert content. + */ + public static final String SQL_TASK_ALERT_MAX_ROWS = "sql.task.alert.max.rows"; + + /** + * sql.task.alert.max.row.value.length: this property defines the max chars of a single string value kept in the + * task-result alert content. + */ + public static final String SQL_TASK_ALERT_MAX_ROW_VALUE_LENGTH = "sql.task.alert.max.row.value.length"; + + /** + * sql.task.alert.max.content.bytes: this property defines the max serialized UTF-8 bytes of the task-result alert + * content. + */ + public static final String SQL_TASK_ALERT_MAX_CONTENT_BYTES = "sql.task.alert.max.content.bytes"; + public static final String SUDO_ENABLE = "sudo.enable"; public static final String RESOURCE_STORAGE_TYPE = "resource.storage.type"; diff --git a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/AlertType.java b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/AlertType.java index 058afcb3fc71..6cc850d6f141 100644 --- a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/AlertType.java +++ b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/AlertType.java @@ -29,7 +29,7 @@ public enum AlertType { /** * 0 workflow instance failure, 1 workflow instance success, 2 workflow instance blocked, 3 workflow instance timeout, 4 fault tolerance warning, - * 5 task failure, 6 task success, 7 task timeout + * 5 task failure, 6 task success, 7 task timeout, 8 task result */ WORKFLOW_INSTANCE_FAILURE(0, "workflow instance failure"), WORKFLOW_INSTANCE_SUCCESS(1, "workflow instance success"), @@ -39,6 +39,7 @@ public enum AlertType { TASK_FAILURE(5, "task failure"), TASK_SUCCESS(6, "task success"), TASK_TIMEOUT(7, "task timeout"), + TASK_RESULT(8, "task result"), ; AlertType(int code, String descp) { diff --git a/dolphinscheduler-common/src/main/resources/common.properties b/dolphinscheduler-common/src/main/resources/common.properties index 9e5adea62ea1..fa397d7ce904 100644 --- a/dolphinscheduler-common/src/main/resources/common.properties +++ b/dolphinscheduler-common/src/main/resources/common.properties @@ -108,3 +108,11 @@ shell.interceptor.type=bash # Whether to enable remote logging remote.logging.enable=false + +# SQL task-result alert content limits, take effect on worker server restart: +# sql.task.alert.max.rows: max rows kept in the alert content, default: 30 +# sql.task.alert.max.row.value.length: max chars of a single string value kept in the alert content, default: 100 +# sql.task.alert.max.content.bytes: max serialized UTF-8 bytes of the alert content, default: 60000 +sql.task.alert.max.rows=30 +sql.task.alert.max.row.value.length=100 +sql.task.alert.max.content.bytes=60000 diff --git a/dolphinscheduler-common/src/test/resources/common.properties b/dolphinscheduler-common/src/test/resources/common.properties index f849b1c3e04d..99ec046b940d 100644 --- a/dolphinscheduler-common/src/test/resources/common.properties +++ b/dolphinscheduler-common/src/test/resources/common.properties @@ -201,4 +201,10 @@ remote.logging.google.cloud.storage.credential=/path/to/credential # gcs bucket name, required if you set remote.logging.target=GCS remote.logging.google.cloud.storage.bucket.name= - +# SQL task-result alert content limits, take effect on worker server restart: +# sql.task.alert.max.rows: max rows kept in the alert content, default: 30 +# sql.task.alert.max.row.value.length: max chars of a single string value kept in the alert content, default: 100 +# sql.task.alert.max.content.bytes: max serialized UTF-8 bytes of the alert content, default: 60000 +sql.task.alert.max.rows=30 +sql.task.alert.max.row.value.length=100 +sql.task.alert.max.content.bytes=60000 diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/AlertDao.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/AlertDao.java index 19ed2d70f2df..f5a4ab3c48d1 100644 --- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/AlertDao.java +++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/AlertDao.java @@ -108,6 +108,24 @@ public int addAlert(Alert alert) { return count; } + /** + * Insert a task-result alert. + * + * @return insert count + */ + public int addTaskResultAlert(Alert alert) { + if (null == alert.getAlertGroupId() || NumberUtils.INTEGER_ZERO.equals(alert.getAlertGroupId())) { + log.warn("the value of alertGroupId is null or 0 "); + return 0; + } + + String sign = generateSign(alert); + alert.setSign(sign); + int count = alertMapper.insert(alert); + log.info("add task result alert to db , alert: {}", alert); + return count; + } + /** * update alert sending(execution) status * @@ -132,6 +150,13 @@ public int updateAlert(AlertStatus alertStatus, String log, int id) { * @return sign's str */ private String generateSign(Alert alert) { + // Task-result alerts include the task instance ID in the sign so that different + // tasks returning identical results get different signs, and repeated success + // events of the same task yield the same sign. For other alert types + // taskInstanceId is null and the sign falls back to content-only. + if (alert.getTaskInstanceId() != null) { + return DigestUtils.sha1Hex(alert.getTaskInstanceId() + "|" + alert.getContent()).toLowerCase(); + } return Optional.of(alert) .map(Alert::getContent) .map(DigestUtils::sha1Hex) diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Alert.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Alert.java index 96ad2890ba54..5916bedc3734 100644 --- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Alert.java +++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Alert.java @@ -84,6 +84,12 @@ public class Alert { @TableField("alert_type") private AlertType alertType; + /** + * Transient — used only for sign generation in idempotent task-result alert inserts. + */ + @TableField(exist = false) + private Integer taskInstanceId; + @TableField(exist = false) private Map info = new HashMap<>(); diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/lifecycle/event/TaskSuccessLifecycleEvent.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/lifecycle/event/TaskSuccessLifecycleEvent.java index 9c7f41e1b527..4568768bcaa0 100644 --- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/lifecycle/event/TaskSuccessLifecycleEvent.java +++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/lifecycle/event/TaskSuccessLifecycleEvent.java @@ -18,6 +18,7 @@ package org.apache.dolphinscheduler.server.master.engine.task.lifecycle.event; import org.apache.dolphinscheduler.plugin.task.api.model.Property; +import org.apache.dolphinscheduler.plugin.task.api.model.TaskAlertInfo; import org.apache.dolphinscheduler.server.master.engine.ILifecycleEventType; import org.apache.dolphinscheduler.server.master.engine.task.execution.ITaskExecution; import org.apache.dolphinscheduler.server.master.engine.task.lifecycle.AbstractTaskLifecycleEvent; @@ -41,6 +42,10 @@ public class TaskSuccessLifecycleEvent extends AbstractTaskLifecycleEvent { private final List varPool; + private final boolean needAlert; + + private final TaskAlertInfo taskAlertInfo; + @Override public ILifecycleEventType getEventType() { return TaskLifecycleEventType.SUCCEEDED; @@ -52,6 +57,7 @@ public String toString() { "task=" + taskExecution.getName() + ", endTime=" + endTime + ", varPool='" + varPool + '\'' + + ", needAlert=" + needAlert + '}'; } } diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/lifecycle/handler/TaskSuccessLifecycleEventHandler.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/lifecycle/handler/TaskSuccessLifecycleEventHandler.java index 86a6ec179927..d9e9f88f5f78 100644 --- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/lifecycle/handler/TaskSuccessLifecycleEventHandler.java +++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/lifecycle/handler/TaskSuccessLifecycleEventHandler.java @@ -17,6 +17,7 @@ package org.apache.dolphinscheduler.server.master.engine.task.lifecycle.handler; +import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus; import org.apache.dolphinscheduler.server.master.engine.ILifecycleEventType; import org.apache.dolphinscheduler.server.master.engine.task.client.TaskExecutorClient; import org.apache.dolphinscheduler.server.master.engine.task.execution.ITaskExecution; @@ -24,18 +25,26 @@ import org.apache.dolphinscheduler.server.master.engine.task.lifecycle.event.TaskSuccessLifecycleEvent; import org.apache.dolphinscheduler.server.master.engine.task.statemachine.ITaskStateAction; import org.apache.dolphinscheduler.server.master.engine.workflow.execution.IWorkflowExecution; +import org.apache.dolphinscheduler.service.alert.WorkflowAlertManager; import org.apache.dolphinscheduler.task.executor.eventbus.ITaskExecutorLifecycleEventReporter; import org.apache.dolphinscheduler.task.executor.events.TaskExecutorLifecycleEventType; +import lombok.extern.slf4j.Slf4j; + import org.springframework.stereotype.Component; +@Slf4j @Component public class TaskSuccessLifecycleEventHandler extends AbstractTaskLifecycleEventHandler { private final TaskExecutorClient taskExecutorClient; - public TaskSuccessLifecycleEventHandler(final TaskExecutorClient taskExecutorClient) { + private final WorkflowAlertManager workflowAlertManager; + + public TaskSuccessLifecycleEventHandler(final TaskExecutorClient taskExecutorClient, + final WorkflowAlertManager workflowAlertManager) { this.taskExecutorClient = taskExecutorClient; + this.workflowAlertManager = workflowAlertManager; } @Override @@ -44,6 +53,19 @@ public void handle(final ITaskStateAction taskStateAction, final ITaskExecution taskExecution, final TaskSuccessLifecycleEvent taskSuccessEvent) { taskStateAction.onSucceedEvent(workflowExecution, taskExecution, taskSuccessEvent); + + // Only persist the task result alert when the success event has been accepted and the + // task has actually transitioned to SUCCESS. State actions that reject a late success + // event (e.g. a paused or killed task) only log a warning and keep the task in its + // current state, so the alert must be skipped as well. + if (taskExecution.getTaskInstance().getState() == TaskExecutionStatus.SUCCESS + && taskSuccessEvent.isNeedAlert()) { + workflowAlertManager.sendTaskResultAlert( + taskExecution.getWorkflowInstance(), + taskExecution.getTaskInstance(), + taskSuccessEvent.getTaskAlertInfo()); + } + taskExecutorClient.ackTaskExecutorLifecycleEvent( taskExecution, new ITaskExecutorLifecycleEventReporter.TaskExecutorLifecycleEventAck( diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/rpc/TaskExecutorEventListenerImpl.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/rpc/TaskExecutorEventListenerImpl.java index 278f5160dc80..0566c839858b 100644 --- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/rpc/TaskExecutorEventListenerImpl.java +++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/rpc/TaskExecutorEventListenerImpl.java @@ -116,6 +116,8 @@ public void onTaskExecutorSuccess(final TaskExecutorSuccessLifecycleEvent taskEx .taskExecution(taskExecution) .endTime(new Date(taskExecutorSuccessLifecycleEvent.getEndTime())) .varPool(taskExecutorSuccessLifecycleEvent.getVarPool()) + .needAlert(taskExecutorSuccessLifecycleEvent.isNeedAlert()) + .taskAlertInfo(taskExecutorSuccessLifecycleEvent.getTaskAlertInfo()) .build(); taskExecution.getWorkflowEventBus().publish(taskSuccessEvent); } finally { diff --git a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/task/lifecycle/handler/TaskSuccessLifecycleEventHandlerTest.java b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/task/lifecycle/handler/TaskSuccessLifecycleEventHandlerTest.java new file mode 100644 index 000000000000..8545becfd958 --- /dev/null +++ b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/task/lifecycle/handler/TaskSuccessLifecycleEventHandlerTest.java @@ -0,0 +1,173 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.dolphinscheduler.server.master.engine.task.lifecycle.handler; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.when; + +import org.apache.dolphinscheduler.dao.entity.TaskInstance; +import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus; +import org.apache.dolphinscheduler.plugin.task.api.model.TaskAlertInfo; +import org.apache.dolphinscheduler.server.master.engine.task.client.TaskExecutorClient; +import org.apache.dolphinscheduler.server.master.engine.task.execution.ITaskExecution; +import org.apache.dolphinscheduler.server.master.engine.task.lifecycle.event.TaskSuccessLifecycleEvent; +import org.apache.dolphinscheduler.server.master.engine.task.statemachine.ITaskStateAction; +import org.apache.dolphinscheduler.server.master.engine.workflow.execution.IWorkflowExecution; +import org.apache.dolphinscheduler.service.alert.WorkflowAlertManager; +import org.apache.dolphinscheduler.task.executor.eventbus.ITaskExecutorLifecycleEventReporter; + +import java.util.Collections; +import java.util.Date; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +@ExtendWith(MockitoExtension.class) +class TaskSuccessLifecycleEventHandlerTest { + + @Mock + private TaskExecutorClient taskExecutorClient; + + @Mock + private WorkflowAlertManager workflowAlertManager; + + @Mock + private ITaskStateAction taskStateAction; + + @Mock + private ITaskExecution taskExecution; + + @Mock + private IWorkflowExecution workflowExecution; + + private TaskInstance taskInstance; + + private TaskSuccessLifecycleEventHandler taskSuccessLifecycleEventHandler; + + @BeforeEach + void setUp() { + taskSuccessLifecycleEventHandler = + new TaskSuccessLifecycleEventHandler(taskExecutorClient, workflowAlertManager); + + taskInstance = new TaskInstance(); + taskInstance.setState(TaskExecutionStatus.RUNNING_EXECUTION); + taskInstance.setHost("localhost"); + + when(taskExecution.getTaskInstance()).thenReturn(taskInstance); + when(taskExecution.getId()).thenReturn(1); + } + + @Test + void shouldSendResultAlertWhenSucceedEventTransitionsTaskToSuccess() { + doAnswer(invocation -> { + taskInstance.setState(TaskExecutionStatus.SUCCESS); + return null; + }).when(taskStateAction) + .onSucceedEvent(any(), any(), any()); + + final TaskSuccessLifecycleEvent taskSuccessEvent = newSuccessEvent(true); + + taskSuccessLifecycleEventHandler.handle( + taskStateAction, workflowExecution, taskExecution, taskSuccessEvent); + + verify(workflowAlertManager).sendTaskResultAlert( + any(), eq(taskInstance), any()); + verify(taskExecutorClient).ackTaskExecutorLifecycleEvent( + eq(taskExecution), any(ITaskExecutorLifecycleEventReporter.TaskExecutorLifecycleEventAck.class)); + } + + @Test + void shouldNotSendResultAlertWhenSucceedEventArrivesOnPausedTask() { + taskInstance.setState(TaskExecutionStatus.PAUSE); + + final TaskSuccessLifecycleEvent taskSuccessEvent = newSuccessEvent(true); + + taskSuccessLifecycleEventHandler.handle( + taskStateAction, workflowExecution, taskExecution, taskSuccessEvent); + + verifyNoInteractions(workflowAlertManager); + verify(taskExecutorClient).ackTaskExecutorLifecycleEvent( + eq(taskExecution), any(ITaskExecutorLifecycleEventReporter.TaskExecutorLifecycleEventAck.class)); + } + + @Test + void shouldNotSendResultAlertWhenSucceedEventArrivesOnKilledTask() { + taskInstance.setState(TaskExecutionStatus.KILL); + + final TaskSuccessLifecycleEvent taskSuccessEvent = newSuccessEvent(true); + + taskSuccessLifecycleEventHandler.handle( + taskStateAction, workflowExecution, taskExecution, taskSuccessEvent); + + verifyNoInteractions(workflowAlertManager); + verify(taskExecutorClient).ackTaskExecutorLifecycleEvent( + eq(taskExecution), any(ITaskExecutorLifecycleEventReporter.TaskExecutorLifecycleEventAck.class)); + } + + @Test + void shouldStillSendResultAlertForRepeatedSucceedEvent() { + taskInstance.setState(TaskExecutionStatus.SUCCESS); + + final TaskSuccessLifecycleEvent taskSuccessEvent = newSuccessEvent(true); + + taskSuccessLifecycleEventHandler.handle( + taskStateAction, workflowExecution, taskExecution, taskSuccessEvent); + + // Repeated success events are accepted idempotently: the alert is sent again, + // which may insert a duplicate row; occasional duplicates are accepted. + verify(workflowAlertManager).sendTaskResultAlert( + any(), eq(taskInstance), any()); + verify(taskExecutorClient).ackTaskExecutorLifecycleEvent( + eq(taskExecution), any(ITaskExecutorLifecycleEventReporter.TaskExecutorLifecycleEventAck.class)); + } + + @Test + void shouldNotSendResultAlertWhenNeedAlertIsFalse() { + doAnswer(invocation -> { + taskInstance.setState(TaskExecutionStatus.SUCCESS); + return null; + }).when(taskStateAction) + .onSucceedEvent(any(), any(), any()); + + final TaskSuccessLifecycleEvent taskSuccessEvent = newSuccessEvent(false); + + taskSuccessLifecycleEventHandler.handle( + taskStateAction, workflowExecution, taskExecution, taskSuccessEvent); + + verifyNoInteractions(workflowAlertManager); + verify(taskExecutorClient).ackTaskExecutorLifecycleEvent( + eq(taskExecution), any(ITaskExecutorLifecycleEventReporter.TaskExecutorLifecycleEventAck.class)); + } + + private TaskSuccessLifecycleEvent newSuccessEvent(final boolean needAlert) { + return TaskSuccessLifecycleEvent.builder() + .taskExecution(taskExecution) + .endTime(new Date()) + .varPool(Collections.emptyList()) + .needAlert(needAlert) + .taskAlertInfo(new TaskAlertInfo()) + .build(); + } +} diff --git a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/alert/WorkflowAlertManager.java b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/alert/WorkflowAlertManager.java index 9335e9551eb7..cba5179b6e30 100644 --- a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/alert/WorkflowAlertManager.java +++ b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/alert/WorkflowAlertManager.java @@ -17,6 +17,7 @@ package org.apache.dolphinscheduler.service.alert; +import org.apache.dolphinscheduler.common.enums.AlertStatus; import org.apache.dolphinscheduler.common.enums.AlertType; import org.apache.dolphinscheduler.common.enums.CommandType; import org.apache.dolphinscheduler.common.enums.Flag; @@ -34,6 +35,7 @@ import org.apache.dolphinscheduler.dao.repository.ProjectDao; import org.apache.dolphinscheduler.dao.repository.UserDao; import org.apache.dolphinscheduler.dao.repository.WorkflowDefinitionLogDao; +import org.apache.dolphinscheduler.plugin.task.api.model.TaskAlertInfo; import java.util.ArrayList; import java.util.Date; @@ -208,4 +210,30 @@ public void sendWorkflowTimeoutAlert(WorkflowInstance workflowInstance) { alertDao.sendWorkflowTimeoutAlert(workflowInstance, projectUser); } + public void sendTaskResultAlert(WorkflowInstance workflowInstance, + TaskInstance taskInstance, + TaskAlertInfo taskAlertInfo) { + if (taskAlertInfo == null || taskAlertInfo.getAlertGroupId() == null + || taskAlertInfo.getAlertGroupId() <= 0) { + return; + } + + Alert alert = new Alert(); + alert.setTitle(taskAlertInfo.getTitle()); + alert.setContent(taskAlertInfo.getContent()); + alert.setWarningType(WarningType.SUCCESS); + alert.setAlertStatus(AlertStatus.WAIT_EXECUTION); + alert.setCreateTime(new Date()); + alert.setAlertGroupId(taskAlertInfo.getAlertGroupId()); + alert.setProjectCode(workflowInstance.getProjectCode()); + alert.setWorkflowDefinitionCode(workflowInstance.getWorkflowDefinitionCode()); + alert.setWorkflowInstanceId(workflowInstance.getId()); + alert.setAlertType(taskAlertInfo.getAlertType() != null + ? taskAlertInfo.getAlertType() + : AlertType.TASK_RESULT); + alert.setTaskInstanceId(taskInstance.getId()); + alertDao.addTaskResultAlert(alert); + log.info("Send task result alert for task: {} in workflow: {}", + taskInstance.getName(), workflowInstance.getName()); + } } diff --git a/dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/alert/WorkflowAlertManagerTest.java b/dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/alert/WorkflowAlertManagerTest.java new file mode 100644 index 000000000000..0fcdfe0cbcb7 --- /dev/null +++ b/dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/alert/WorkflowAlertManagerTest.java @@ -0,0 +1,123 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.dolphinscheduler.service.alert; + +import org.apache.dolphinscheduler.common.enums.AlertStatus; +import org.apache.dolphinscheduler.common.enums.AlertType; +import org.apache.dolphinscheduler.common.enums.WarningType; +import org.apache.dolphinscheduler.dao.AlertDao; +import org.apache.dolphinscheduler.dao.entity.Alert; +import org.apache.dolphinscheduler.dao.entity.TaskInstance; +import org.apache.dolphinscheduler.dao.entity.WorkflowInstance; +import org.apache.dolphinscheduler.dao.repository.ProjectDao; +import org.apache.dolphinscheduler.dao.repository.UserDao; +import org.apache.dolphinscheduler.dao.repository.WorkflowDefinitionLogDao; +import org.apache.dolphinscheduler.plugin.task.api.model.TaskAlertInfo; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.ArgumentCaptor; +import org.mockito.InjectMocks; +import org.mockito.Mock; +import org.mockito.Mockito; +import org.mockito.junit.jupiter.MockitoExtension; + +@ExtendWith(MockitoExtension.class) +class WorkflowAlertManagerTest { + + @Mock + private AlertDao alertDao; + + @Mock + private WorkflowDefinitionLogDao workflowDefinitionLogDao; + + @Mock + private UserDao userDao; + + @Mock + private ProjectDao projectDao; + + @InjectMocks + private WorkflowAlertManager workflowAlertManager; + + @Test + void testSendTaskResultAlertSetsWaitExecutionStatus() { + WorkflowInstance workflowInstance = new WorkflowInstance(); + workflowInstance.setId(1001); + workflowInstance.setName("test-workflow"); + workflowInstance.setProjectCode(2001L); + workflowInstance.setWorkflowDefinitionCode(3001L); + + TaskInstance taskInstance = new TaskInstance(); + taskInstance.setId(5001); + taskInstance.setName("sql-query-task"); + + TaskAlertInfo taskAlertInfo = new TaskAlertInfo(); + taskAlertInfo.setTitle("SQL Task Result"); + taskAlertInfo.setContent("[{\"taskName\":\"sql-query-task\",\"result\":\"ok\"}]"); + taskAlertInfo.setAlertGroupId(1); + taskAlertInfo.setAlertType(AlertType.TASK_RESULT); + + Mockito.when(alertDao.addTaskResultAlert(Mockito.any(Alert.class))).thenReturn(1); + + workflowAlertManager.sendTaskResultAlert(workflowInstance, taskInstance, taskAlertInfo); + + ArgumentCaptor alertCaptor = ArgumentCaptor.forClass(Alert.class); + Mockito.verify(alertDao).addTaskResultAlert(alertCaptor.capture()); + + Alert captured = alertCaptor.getValue(); + Assertions.assertEquals(AlertStatus.WAIT_EXECUTION, captured.getAlertStatus()); + Assertions.assertEquals("SQL Task Result", captured.getTitle()); + Assertions.assertEquals(WarningType.SUCCESS, captured.getWarningType()); + Assertions.assertEquals(AlertType.TASK_RESULT, captured.getAlertType()); + Assertions.assertEquals(1001, captured.getWorkflowInstanceId()); + Assertions.assertEquals(2001L, captured.getProjectCode()); + Assertions.assertEquals(3001L, captured.getWorkflowDefinitionCode()); + Assertions.assertEquals(1, captured.getAlertGroupId()); + Assertions.assertNotNull(captured.getCreateTime()); + Assertions.assertEquals(5001, captured.getTaskInstanceId()); + } + + @Test + void testSendTaskResultAlertNullTaskAlertInfoDoesNothing() { + WorkflowInstance workflowInstance = new WorkflowInstance(); + workflowInstance.setId(1001); + TaskInstance taskInstance = new TaskInstance(); + + workflowAlertManager.sendTaskResultAlert(workflowInstance, taskInstance, null); + + Mockito.verifyNoInteractions(alertDao); + } + + @Test + void testSendTaskResultAlertNullAlertGroupIdDoesNothing() { + WorkflowInstance workflowInstance = new WorkflowInstance(); + workflowInstance.setId(1001); + TaskInstance taskInstance = new TaskInstance(); + + TaskAlertInfo taskAlertInfo = new TaskAlertInfo(); + taskAlertInfo.setTitle("SQL Task Result"); + taskAlertInfo.setContent("[]"); + taskAlertInfo.setAlertGroupId(null); + + workflowAlertManager.sendTaskResultAlert(workflowInstance, taskInstance, taskAlertInfo); + + Mockito.verifyNoInteractions(alertDao); + } +} diff --git a/dolphinscheduler-task-executor/src/main/java/org/apache/dolphinscheduler/task/executor/events/TaskExecutorSuccessLifecycleEvent.java b/dolphinscheduler-task-executor/src/main/java/org/apache/dolphinscheduler/task/executor/events/TaskExecutorSuccessLifecycleEvent.java index 947d1edf3633..2c55c80c365c 100644 --- a/dolphinscheduler-task-executor/src/main/java/org/apache/dolphinscheduler/task/executor/events/TaskExecutorSuccessLifecycleEvent.java +++ b/dolphinscheduler-task-executor/src/main/java/org/apache/dolphinscheduler/task/executor/events/TaskExecutorSuccessLifecycleEvent.java @@ -19,6 +19,7 @@ import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext; import org.apache.dolphinscheduler.plugin.task.api.model.Property; +import org.apache.dolphinscheduler.plugin.task.api.model.TaskAlertInfo; import org.apache.dolphinscheduler.task.executor.ITaskExecutor; import java.util.List; @@ -48,6 +49,10 @@ public class TaskExecutorSuccessLifecycleEvent extends AbstractTaskExecutorLifec private Long latestReportTime; + private boolean needAlert; + + private TaskAlertInfo taskAlertInfo; + public static TaskExecutorSuccessLifecycleEvent of(final ITaskExecutor taskExecutor) { final TaskExecutionContext taskExecutionContext = taskExecutor.getTaskExecutionContext(); return TaskExecutorSuccessLifecycleEvent.builder() @@ -57,6 +62,8 @@ public static TaskExecutorSuccessLifecycleEvent of(final ITaskExecutor taskExecu .varPool(taskExecutionContext.getVarPool()) .type(TaskExecutorLifecycleEventType.SUCCESS) .endTime(taskExecutionContext.getEndTime()) + .needAlert(taskExecutionContext.isNeedAlert()) + .taskAlertInfo(taskExecutionContext.getTaskAlertInfo()) .build(); } diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/AbstractTask.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/AbstractTask.java index 0c8f208fcbae..afa990e7bbdb 100644 --- a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/AbstractTask.java +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/AbstractTask.java @@ -56,8 +56,22 @@ public abstract class AbstractTask { */ protected volatile int exitStatusCode = -1; + /** + * @deprecated Use {@link TaskExecutionContext#isNeedAlert()} on the + * {@code taskRequest} field instead. Retained for binary + * compatibility with third-party plugins compiled against + * the previous AbstractTask API. + */ + @Deprecated protected boolean needAlert = false; + /** + * @deprecated Use {@link TaskExecutionContext#getTaskAlertInfo()} on the + * {@code taskRequest} field instead. Retained for binary + * compatibility with third-party plugins compiled against + * the previous AbstractTask API. + */ + @Deprecated protected TaskAlertInfo taskAlertInfo; /** @@ -109,20 +123,53 @@ public void setAppIds(String appIds) { this.appIds = appIds; } + /** + * @deprecated Use {@link TaskExecutionContext#isNeedAlert()} on the + * {@code taskRequest} field instead. This method is retained as a + * compatibility bridge so that third-party plugins compiled against + * the previous AbstractTask API continue to work. The value is merged + * with the legacy field so that direct field writes from such plugins + * are still honored. + */ + @Deprecated public boolean getNeedAlert() { - return needAlert; + return needAlert || (taskRequest != null && taskRequest.isNeedAlert()); } + /** + * @deprecated Use {@link TaskExecutionContext#setNeedAlert(boolean)} on the + * {@code taskRequest} field instead. + */ + @Deprecated public void setNeedAlert(boolean needAlert) { this.needAlert = needAlert; + if (taskRequest != null) { + taskRequest.setNeedAlert(needAlert); + } } + /** + * @deprecated Use {@link TaskExecutionContext#getTaskAlertInfo()} on the + * {@code taskRequest} field instead. + */ + @Deprecated public TaskAlertInfo getTaskAlertInfo() { - return taskAlertInfo; + if (taskAlertInfo != null) { + return taskAlertInfo; + } + return taskRequest != null ? taskRequest.getTaskAlertInfo() : null; } + /** + * @deprecated Use {@link TaskExecutionContext#setTaskAlertInfo(TaskAlertInfo)} + * on the {@code taskRequest} field instead. + */ + @Deprecated public void setTaskAlertInfo(TaskAlertInfo taskAlertInfo) { this.taskAlertInfo = taskAlertInfo; + if (taskRequest != null) { + taskRequest.setTaskAlertInfo(taskAlertInfo); + } } /** diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/TaskExecutionContext.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/TaskExecutionContext.java index 85d6f682196b..a072258011df 100644 --- a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/TaskExecutionContext.java +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/TaskExecutionContext.java @@ -19,6 +19,7 @@ import org.apache.dolphinscheduler.plugin.task.api.enums.TaskTimeoutStrategy; import org.apache.dolphinscheduler.plugin.task.api.model.Property; +import org.apache.dolphinscheduler.plugin.task.api.model.TaskAlertInfo; import org.apache.dolphinscheduler.plugin.task.api.parameters.resource.ResourceParametersHelper; import org.apache.dolphinscheduler.plugin.task.api.resource.ResourceContext; @@ -130,6 +131,10 @@ public class TaskExecutionContext implements Serializable { private final long firstDispatchTime = System.currentTimeMillis(); + private boolean needAlert; + + private TaskAlertInfo taskAlertInfo; + public int increaseDispatchFailTimes() { return ++dispatchFailTimes; } diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/model/TaskAlertInfo.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/model/TaskAlertInfo.java index e72242e93473..d2c22e565a79 100644 --- a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/model/TaskAlertInfo.java +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/model/TaskAlertInfo.java @@ -17,6 +17,8 @@ package org.apache.dolphinscheduler.plugin.task.api.model; +import org.apache.dolphinscheduler.common.enums.AlertType; + import lombok.Data; @Data @@ -27,4 +29,7 @@ public class TaskAlertInfo { private String content; private Integer alertGroupId; + + private AlertType alertType; + } diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/test/java/org/apache/dolphinscheduler/plugin/task/api/AbstractTaskTest.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/test/java/org/apache/dolphinscheduler/plugin/task/api/AbstractTaskTest.java index 5421c0557554..f3a8a8b53467 100644 --- a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/test/java/org/apache/dolphinscheduler/plugin/task/api/AbstractTaskTest.java +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/test/java/org/apache/dolphinscheduler/plugin/task/api/AbstractTaskTest.java @@ -18,6 +18,7 @@ package org.apache.dolphinscheduler.plugin.task.api; import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus; +import org.apache.dolphinscheduler.plugin.task.api.model.TaskAlertInfo; import org.apache.dolphinscheduler.plugin.task.api.parameters.AbstractParameters; import org.apache.dolphinscheduler.plugin.task.api.parameters.SubWorkflowParameters; @@ -62,12 +63,65 @@ public void testGetExitStatus() { Assertions.assertEquals(TaskExecutionStatus.FAILURE, task.getExitStatus()); } + @Test + public void testDirectFieldAssignmentIsExposedByGetters() { + AbstractTask task = new TestTask(); + // legacy plugins may assign the protected fields directly + task.needAlert = true; + TaskAlertInfo alertInfo = new TaskAlertInfo(); + alertInfo.setTitle("legacy alert"); + task.taskAlertInfo = alertInfo; + + Assertions.assertTrue(task.getNeedAlert()); + Assertions.assertSame(alertInfo, task.getTaskAlertInfo()); + } + + @Test + public void testSettersWriteBothFieldAndContext() { + TaskExecutionContext ctx = new TaskExecutionContext(); + AbstractTask task = new TestTask(ctx); + task.setNeedAlert(true); + TaskAlertInfo alertInfo = new TaskAlertInfo(); + alertInfo.setTitle("setter alert"); + task.setTaskAlertInfo(alertInfo); + + Assertions.assertTrue(ctx.isNeedAlert()); + Assertions.assertSame(alertInfo, ctx.getTaskAlertInfo()); + Assertions.assertTrue(task.getNeedAlert()); + Assertions.assertSame(alertInfo, task.getTaskAlertInfo()); + } + + @Test + public void testGettersFallBackToContextWhenFieldNotSet() { + TaskExecutionContext ctx = new TaskExecutionContext(); + AbstractTask task = new TestTask(ctx); + ctx.setNeedAlert(true); + TaskAlertInfo alertInfo = new TaskAlertInfo(); + alertInfo.setTitle("context alert"); + ctx.setTaskAlertInfo(alertInfo); + + Assertions.assertTrue(task.getNeedAlert()); + Assertions.assertSame(alertInfo, task.getTaskAlertInfo()); + } + + @Test + public void testGettersReturnDefaultsWhenNothingSet() { + AbstractTask task = new TestTask(); + + Assertions.assertFalse(task.getNeedAlert()); + Assertions.assertNull(task.getTaskAlertInfo()); + } + private static final class TestTask extends AbstractTask { private TestTask() { super(new TaskExecutionContext()); } + private TestTask(TaskExecutionContext taskExecutionContext) { + super(taskExecutionContext); + } + @Override public void handle(TaskCallBack taskCallBack) { } diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-sql/src/main/java/org/apache/dolphinscheduler/plugin/task/sql/SqlTask.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-sql/src/main/java/org/apache/dolphinscheduler/plugin/task/sql/SqlTask.java index 113dfbcd9eb8..12c0cce3231a 100644 --- a/dolphinscheduler-task-plugin/dolphinscheduler-task-sql/src/main/java/org/apache/dolphinscheduler/plugin/task/sql/SqlTask.java +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-sql/src/main/java/org/apache/dolphinscheduler/plugin/task/sql/SqlTask.java @@ -17,8 +17,11 @@ package org.apache.dolphinscheduler.plugin.task.sql; +import org.apache.dolphinscheduler.common.constants.Constants; +import org.apache.dolphinscheduler.common.enums.AlertType; import org.apache.dolphinscheduler.common.utils.DateUtils; import org.apache.dolphinscheduler.common.utils.JSONUtils; +import org.apache.dolphinscheduler.common.utils.PropertyUtils; import org.apache.dolphinscheduler.plugin.datasource.api.plugin.DataSourceClientProvider; import org.apache.dolphinscheduler.plugin.datasource.api.plugin.DataSourcePluginManager; import org.apache.dolphinscheduler.plugin.datasource.api.utils.DataSourceUtils; @@ -56,6 +59,7 @@ import java.util.Collections; import java.util.HashMap; import java.util.HashSet; +import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.Optional; @@ -66,6 +70,7 @@ import lombok.extern.slf4j.Slf4j; +import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.node.ArrayNode; import com.fasterxml.jackson.databind.node.ObjectNode; @@ -90,6 +95,34 @@ public class SqlTask extends AbstractTask { */ private static final int QUERY_LIMIT = 10000; + /** + * default max rows kept in the task-result alert content; rows beyond this are replaced by an omission marker, + * overridable via the sql.task.alert.max.rows property in common.properties + */ + private static final int DEFAULT_ALERT_MAX_ROWS = 30; + + /** + * default max length of a single string field value kept in the alert content; longer values are truncated + * with an ellipsis, overridable via the sql.task.alert.max.row.value.length property in common.properties + */ + private static final int DEFAULT_ALERT_MAX_ROW_VALUE_LENGTH = 100; + + /** + * default max serialized UTF-8 bytes of the alert content, kept below the 65535-byte capacity + * of the MySQL TEXT column that stores t_ds_alert.content, overridable via the + * sql.task.alert.max.content.bytes property in common.properties + */ + private static final int DEFAULT_ALERT_MAX_CONTENT_BYTES = 60000; + + private final int alertMaxRows = + PropertyUtils.getInt(Constants.SQL_TASK_ALERT_MAX_ROWS, DEFAULT_ALERT_MAX_ROWS); + + private final int alertMaxRowValueLength = + PropertyUtils.getInt(Constants.SQL_TASK_ALERT_MAX_ROW_VALUE_LENGTH, DEFAULT_ALERT_MAX_ROW_VALUE_LENGTH); + + private final int alertMaxContentBytes = + PropertyUtils.getInt(Constants.SQL_TASK_ALERT_MAX_CONTENT_BYTES, DEFAULT_ALERT_MAX_CONTENT_BYTES); + private final SQLTaskExecutionContext sqlTaskExecutionContext; private final DbType dbType; @@ -287,13 +320,18 @@ private String resultProcess(ResultSet resultSet) throws Exception { } } - String result = resultJSONArray.isEmpty() ? JSONUtils.toJsonString(generateEmptyRow(resultSet)) - : JSONUtils.toJsonString(resultJSONArray); + String result; + ArrayNode alertArray = resultJSONArray; + if (resultJSONArray.isEmpty()) { + ArrayNode emptyRow = generateEmptyRow(resultSet); + result = JSONUtils.toJsonString(emptyRow); + alertArray = emptyRow; + } else { + result = JSONUtils.toJsonString(resultJSONArray); + } if (Boolean.TRUE.equals(sqlParameters.getSendEmail())) { - sendAttachment(sqlParameters.getGroupId(), StringUtils.isNotEmpty(sqlParameters.getTitle()) - ? sqlParameters.getTitle() - : taskExecutionContext.getTaskName() + " query result sets", result); + prepareTaskResultAlert(alertArray); } log.debug("execute sql result : {}", result); return result; @@ -320,18 +358,99 @@ private ArrayNode generateEmptyRow(ResultSet resultSet) throws SQLException { } /** - * send alert as an attachment - * - * @param title title - * @param content content + * Prepare task result alert info with the complete query result as content. */ - private void sendAttachment(int groupId, String title, String content) { - setNeedAlert(Boolean.TRUE); + private void prepareTaskResultAlert(ArrayNode resultJSONArray) { TaskAlertInfo taskAlertInfo = new TaskAlertInfo(); - taskAlertInfo.setAlertGroupId(groupId); - taskAlertInfo.setContent(content); - taskAlertInfo.setTitle(title); - setTaskAlertInfo(taskAlertInfo); + taskAlertInfo.setAlertGroupId(sqlParameters.getGroupId()); + taskAlertInfo.setTitle(StringUtils.isNotEmpty(sqlParameters.getTitle()) + ? sqlParameters.getTitle() + : taskExecutionContext.getTaskName() + " query result sets"); + taskAlertInfo.setContent(limitAlertContent(resultJSONArray)); + taskAlertInfo.setAlertType(AlertType.TASK_RESULT); + + taskExecutionContext.setNeedAlert(true); + taskExecutionContext.setTaskAlertInfo(taskAlertInfo); + log.debug("Prepare task result alert: title={}, alertGroupId={}, alertType={}, totalRows={}", + taskAlertInfo.getTitle(), taskAlertInfo.getAlertGroupId(), taskAlertInfo.getAlertType(), + resultJSONArray.size()); + } + + /** + * Build a size-limited alert content that stays a valid JSON array. + * At most the configured max rows are kept; string field values longer than the configured + * max value length are truncated with an ellipsis. The serialized content is bounded by + * the configured max UTF-8 bytes: rows are dropped from the tail until it fits; a row that + * alone exceeds the budget is replaced by the omission marker. + */ + private String limitAlertContent(ArrayNode resultJSONArray) { + int totalRows = resultJSONArray.size(); + ArrayNode limited = JSONUtils.createArrayNode(); + int rowsToKeep = Math.min(totalRows, alertMaxRows); + for (int i = 0; i < rowsToKeep; i++) { + limited.add(limitRowValues(resultJSONArray.get(i))); + } + + int omittedRows = totalRows - rowsToKeep; + while (true) { + if (omittedRows > 0) { + limited.add(createOmissionMarker(omittedRows)); + } + String content = JSONUtils.toJsonString(limited); + if (content.getBytes(StandardCharsets.UTF_8).length <= alertMaxContentBytes) { + return content; + } + if (omittedRows > 0) { + limited.remove(limited.size() - 1); + } + if (limited.isEmpty()) { + // Even a lone marker exceeds the budget (pathological budget); return the minimal notice. + return JSONUtils.toJsonString(createOmissionMarkerArray(totalRows)); + } + limited.remove(limited.size() - 1); + omittedRows++; + } + } + + private ObjectNode createOmissionMarker(int omittedRows) { + ObjectNode omitted = JSONUtils.createObjectNode(); + omitted.put("...", omittedRows + " more rows omitted"); + return omitted; + } + + private ArrayNode createOmissionMarkerArray(int totalRows) { + ArrayNode markerArray = JSONUtils.createArrayNode(); + markerArray.add(createOmissionMarker(totalRows)); + return markerArray; + } + + private JsonNode limitRowValues(JsonNode row) { + if (!row.isObject()) { + return row; + } + ObjectNode limitedRow = JSONUtils.createObjectNode(); + Iterator> fields = row.fields(); + while (fields.hasNext()) { + Map.Entry entry = fields.next(); + JsonNode value = entry.getValue(); + if (value.isTextual() && value.textValue().length() > alertMaxRowValueLength) { + limitedRow.put(entry.getKey(), + truncateAtCodePointBoundary(value.textValue(), alertMaxRowValueLength) + "..."); + } else { + limitedRow.set(entry.getKey(), value); + } + } + return limitedRow; + } + + /** + * Cut a string to at most maxChars characters without splitting a surrogate pair. + */ + private String truncateAtCodePointBoundary(String text, int maxChars) { + if (text.codePointCount(0, text.length()) <= maxChars) { + return text; + } + return text.substring(0, text.offsetByCodePoints(0, maxChars)); } private String executeQuery(Connection connection, SqlBinds sqlBinds, String handlerType) throws Exception { diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-sql/src/test/java/org/apache/dolphinscheduler/plugin/task/sql/SqlTaskTest.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-sql/src/test/java/org/apache/dolphinscheduler/plugin/task/sql/SqlTaskTest.java index 92f0fa0702d2..7dc36bc9e22c 100644 --- a/dolphinscheduler-task-plugin/dolphinscheduler-task-sql/src/test/java/org/apache/dolphinscheduler/plugin/task/sql/SqlTaskTest.java +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-sql/src/test/java/org/apache/dolphinscheduler/plugin/task/sql/SqlTaskTest.java @@ -20,8 +20,10 @@ import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; +import org.apache.dolphinscheduler.common.enums.AlertType; import org.apache.dolphinscheduler.common.utils.DateUtils; import org.apache.dolphinscheduler.common.utils.JSONUtils; +import org.apache.dolphinscheduler.common.utils.PropertyUtils; import org.apache.dolphinscheduler.plugin.task.api.TaskConstants; import org.apache.dolphinscheduler.plugin.task.api.TaskException; import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext; @@ -29,6 +31,7 @@ import org.apache.dolphinscheduler.plugin.task.api.enums.Direct; import org.apache.dolphinscheduler.plugin.task.api.enums.ResourceType; import org.apache.dolphinscheduler.plugin.task.api.model.Property; +import org.apache.dolphinscheduler.plugin.task.api.model.TaskAlertInfo; import org.apache.dolphinscheduler.plugin.task.api.parameters.SqlParameters; import org.apache.dolphinscheduler.plugin.task.api.parameters.resource.DataSourceParameters; import org.apache.dolphinscheduler.plugin.task.api.parameters.resource.ResourceParametersHelper; @@ -36,6 +39,7 @@ import org.apache.dolphinscheduler.plugin.task.api.utils.ParameterUtils; import org.apache.dolphinscheduler.spi.enums.DbType; +import java.lang.reflect.Field; import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Method; import java.nio.charset.StandardCharsets; @@ -50,7 +54,10 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; +import org.mockito.MockedStatic; +import org.mockito.Mockito; +import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.node.ArrayNode; import com.fasterxml.jackson.databind.node.ObjectNode; import com.google.common.collect.Lists; @@ -489,4 +496,390 @@ private ResourceParametersHelper getResourceParametersHelperWithDatasourceType(D return resourceParametersHelper; } + private TaskExecutionContext createSqlTaskWithAlert(ArrayNode resultArray, int displayRows, String title, + int groupId) { + SqlParameters sqlParameters = new SqlParameters(); + sqlParameters.setType("HIVE"); + sqlParameters.setDatasource(1); + sqlParameters.setSql("select 1"); + sqlParameters.setSendEmail(true); + sqlParameters.setDisplayRows(displayRows); + sqlParameters.setGroupId(groupId); + if (title != null) { + sqlParameters.setTitle(title); + } + + TaskExecutionContext ctx = new TaskExecutionContext(); + ctx.setTaskParams(JSONUtils.toJsonString(sqlParameters)); + ctx.setTaskName("test_sql_task"); + ctx.setResourceParametersHelper(getResourceParametersHelperWithDatasourceType(DbType.HIVE)); + + SqlTask task = new SqlTask(ctx); + try { + Method method = SqlTask.class.getDeclaredMethod("prepareTaskResultAlert", ArrayNode.class); + method.setAccessible(true); + method.invoke(task, resultArray); + } catch (Exception e) { + throw new RuntimeException(e); + } + return ctx; + } + + @Test + void testPrepareTaskResultAlertSetsNeedAlertAndTaskAlertInfo() { + ArrayNode resultArray = JSONUtils.createArrayNode(); + resultArray.add(JSONUtils.parseObject("{\"id\":\"1\",\"name\":\"alice\"}")); + resultArray.add(JSONUtils.parseObject("{\"id\":\"2\",\"name\":\"bob\"}")); + + TaskExecutionContext ctx = createSqlTaskWithAlert(resultArray, 10, "My Alert Title", 5); + + Assertions.assertTrue(ctx.isNeedAlert()); + + TaskAlertInfo alertInfo = ctx.getTaskAlertInfo(); + Assertions.assertNotNull(alertInfo); + Assertions.assertEquals(5, alertInfo.getAlertGroupId()); + Assertions.assertEquals("My Alert Title", alertInfo.getTitle()); + Assertions.assertEquals(AlertType.TASK_RESULT, alertInfo.getAlertType()); + Assertions.assertNotNull(alertInfo.getContent()); + + ArrayNode contentArray = JSONUtils.parseArray(alertInfo.getContent()); + Assertions.assertEquals(2, contentArray.size()); + } + + @Test + void testPrepareTaskResultAlertKeepsFullContentRegardlessOfDisplayRows() throws Exception { + int totalRows = readStaticIntConstant("DEFAULT_ALERT_MAX_ROWS") - 1; + ArrayNode resultArray = JSONUtils.createArrayNode(); + for (int i = 0; i < totalRows; i++) { + resultArray.add(JSONUtils.parseObject("{\"id\":\"" + i + "\"}")); + } + + // displayRows is only a log-display setting, the alert content must keep all rows + TaskExecutionContext ctx = createSqlTaskWithAlert(resultArray, 3, null, 1); + + TaskAlertInfo alertInfo = ctx.getTaskAlertInfo(); + Assertions.assertNotNull(alertInfo); + + ArrayNode contentArray = JSONUtils.parseArray(alertInfo.getContent()); + Assertions.assertEquals(totalRows, contentArray.size()); + Assertions.assertEquals("0", contentArray.get(0).get("id").asText()); + Assertions.assertEquals(String.valueOf(totalRows - 1), contentArray.get(totalRows - 1).get("id").asText()); + } + + @Test + void testPrepareTaskResultAlertKeepsFullContentWhenDisplayRowsUnset() throws Exception { + int totalRows = readStaticIntConstant("DEFAULT_ALERT_MAX_ROWS") - 1; + ArrayNode resultArray = JSONUtils.createArrayNode(); + for (int i = 0; i < totalRows; i++) { + resultArray.add(JSONUtils.parseObject("{\"id\":\"" + i + "\"}")); + } + + // displayRows = 0 means unset, the alert content still keeps all rows + TaskExecutionContext ctx = createSqlTaskWithAlert(resultArray, 0, "default rows test", 2); + + TaskAlertInfo alertInfo = ctx.getTaskAlertInfo(); + Assertions.assertNotNull(alertInfo); + + ArrayNode contentArray = JSONUtils.parseArray(alertInfo.getContent()); + Assertions.assertEquals(totalRows, contentArray.size()); + } + + @Test + void testPrepareTaskResultAlertUsesCustomTitle() { + ArrayNode resultArray = JSONUtils.createArrayNode(); + resultArray.add(JSONUtils.parseObject("{\"id\":\"1\"}")); + + TaskExecutionContext ctx = createSqlTaskWithAlert(resultArray, 10, "Custom Title", 1); + + TaskAlertInfo alertInfo = ctx.getTaskAlertInfo(); + Assertions.assertNotNull(alertInfo); + Assertions.assertEquals("Custom Title", alertInfo.getTitle()); + } + + @Test + void testPrepareTaskResultAlertUsesDefaultTitleWhenEmpty() { + ArrayNode resultArray = JSONUtils.createArrayNode(); + resultArray.add(JSONUtils.parseObject("{\"id\":\"1\"}")); + + TaskExecutionContext ctx = createSqlTaskWithAlert(resultArray, 10, null, 1); + + TaskAlertInfo alertInfo = ctx.getTaskAlertInfo(); + Assertions.assertNotNull(alertInfo); + Assertions.assertEquals("test_sql_task query result sets", alertInfo.getTitle()); + } + + @Test + void testPrepareTaskResultAlertTruncatesRowsBeyondMax() throws Exception { + int maxRows = readStaticIntConstant("DEFAULT_ALERT_MAX_ROWS"); + int totalRows = maxRows + 10; + ArrayNode resultArray = JSONUtils.createArrayNode(); + for (int i = 0; i < totalRows; i++) { + resultArray.add(JSONUtils.parseObject("{\"id\":\"" + i + "\"}")); + } + + TaskExecutionContext ctx = createSqlTaskWithAlert(resultArray, 10, "truncate rows test", 1); + + TaskAlertInfo alertInfo = ctx.getTaskAlertInfo(); + ArrayNode contentArray = JSONUtils.parseArray(alertInfo.getContent()); + Assertions.assertEquals(maxRows + 1, contentArray.size()); + Assertions.assertEquals("0", contentArray.get(0).get("id").asText()); + JsonNode last = contentArray.get(contentArray.size() - 1); + Assertions.assertTrue(last.has("...")); + Assertions.assertEquals((totalRows - maxRows) + " more rows omitted", last.get("...").asText()); + } + + @Test + void testPrepareTaskResultAlertTruncatesLongFieldValue() throws Exception { + int maxLen = readStaticIntConstant("DEFAULT_ALERT_MAX_ROW_VALUE_LENGTH"); + StringBuilder longValue = new StringBuilder(); + for (int i = 0; i < maxLen + 50; i++) { + longValue.append("x"); + } + ArrayNode resultArray = JSONUtils.createArrayNode(); + resultArray.add(JSONUtils.parseObject("{\"id\":\"1\",\"payload\":\"" + longValue + "\"}")); + + TaskExecutionContext ctx = createSqlTaskWithAlert(resultArray, 10, "truncate field test", 1); + + TaskAlertInfo alertInfo = ctx.getTaskAlertInfo(); + ArrayNode contentArray = JSONUtils.parseArray(alertInfo.getContent()); + Assertions.assertEquals(1, contentArray.size()); + String payload = contentArray.get(0).get("payload").asText(); + Assertions.assertEquals(maxLen + 3, payload.length()); + Assertions.assertTrue(payload.endsWith("...")); + } + + @Test + void testPrepareTaskResultAlertBoundsTotalBytesWithWideRows() throws Exception { + int maxRows = readStaticIntConstant("DEFAULT_ALERT_MAX_ROWS"); + int maxLen = readStaticIntConstant("DEFAULT_ALERT_MAX_ROW_VALUE_LENGTH"); + int budget = readStaticIntConstant("DEFAULT_ALERT_MAX_CONTENT_BYTES"); + + // Wide rows: enough columns that keeping all rows overflows the byte budget + // even though every single value stays under the per-value character limit. + int columns = budget / (maxRows * (maxLen - 1)) + 5; + int totalRows = maxRows; + ArrayNode resultArray = JSONUtils.createArrayNode(); + for (int i = 0; i < totalRows; i++) { + ObjectNode row = JSONUtils.createObjectNode(); + for (int c = 0; c < columns; c++) { + row.put("col_" + c, repeatChar('v', maxLen - 1)); + } + resultArray.add(row); + } + + TaskExecutionContext ctx = createSqlTaskWithAlert(resultArray, 10, "wide rows test", 1); + + TaskAlertInfo alertInfo = ctx.getTaskAlertInfo(); + int contentBytes = alertInfo.getContent().getBytes(StandardCharsets.UTF_8).length; + Assertions.assertTrue(contentBytes <= budget, + "serialized alert content must stay within the byte budget: " + contentBytes); + + ArrayNode contentArray = JSONUtils.parseArray(alertInfo.getContent()); + int keptRows = contentArray.size() - 1; + Assertions.assertTrue(keptRows < totalRows, "rows must be dropped from the tail to fit the budget"); + Assertions.assertTrue(contentArray.get(0).has("col_0")); + JsonNode last = contentArray.get(contentArray.size() - 1); + Assertions.assertTrue(last.has("...")); + Assertions.assertEquals((totalRows - keptRows) + " more rows omitted", last.get("...").asText()); + } + + @Test + void testPrepareTaskResultAlertBoundsTotalBytesWithMultibyteText() throws Exception { + int maxRows = readStaticIntConstant("DEFAULT_ALERT_MAX_ROWS"); + int maxLen = readStaticIntConstant("DEFAULT_ALERT_MAX_ROW_VALUE_LENGTH"); + int budget = readStaticIntConstant("DEFAULT_ALERT_MAX_CONTENT_BYTES"); + + // Each character is 3 UTF-8 bytes, so rows within the character limits still + // overflow the byte budget and rows must be dropped from the tail. + char cjk = '\u4e2d'; + int columns = budget / (maxRows * 3 * maxLen) + 5; + int totalRows = maxRows; + ArrayNode resultArray = JSONUtils.createArrayNode(); + for (int i = 0; i < totalRows; i++) { + ObjectNode row = JSONUtils.createObjectNode(); + for (int c = 0; c < columns; c++) { + row.put("col_" + c, repeatChar(cjk, maxLen)); + } + resultArray.add(row); + } + + TaskExecutionContext ctx = createSqlTaskWithAlert(resultArray, 10, "multibyte test", 1); + + TaskAlertInfo alertInfo = ctx.getTaskAlertInfo(); + int contentBytes = alertInfo.getContent().getBytes(StandardCharsets.UTF_8).length; + Assertions.assertTrue(contentBytes <= budget, + "multibyte alert content must stay within the byte budget: " + contentBytes); + + ArrayNode contentArray = JSONUtils.parseArray(alertInfo.getContent()); + int keptRows = contentArray.size() - 1; + Assertions.assertTrue(keptRows < totalRows, "rows must be dropped from the tail to fit the budget"); + // Kept values are exactly at the character limit, so they must stay complete. + String firstValue = contentArray.get(0).get("col_0").asText(); + Assertions.assertEquals(repeatChar(cjk, maxLen), firstValue); + JsonNode last = contentArray.get(contentArray.size() - 1); + Assertions.assertTrue(last.has("...")); + Assertions.assertEquals((totalRows - keptRows) + " more rows omitted", last.get("...").asText()); + } + + @Test + void testPrepareTaskResultAlertReplacesRowExceedingByteBudgetAlone() throws Exception { + int budget = readStaticIntConstant("DEFAULT_ALERT_MAX_CONTENT_BYTES"); + + // Nested values are not truncated per-field, so a single row can exceed the + // budget alone; it must be replaced by the omission marker. + ObjectNode row = JSONUtils.createObjectNode(); + row.put("id", "1"); + ObjectNode payload = JSONUtils.createObjectNode(); + payload.put("blob", repeatChar('n', budget * 2)); + row.set("payload", payload); + + ArrayNode resultArray = JSONUtils.createArrayNode(); + resultArray.add(row); + + TaskExecutionContext ctx = createSqlTaskWithAlert(resultArray, 10, "huge nested value test", 1); + + TaskAlertInfo alertInfo = ctx.getTaskAlertInfo(); + int contentBytes = alertInfo.getContent().getBytes(StandardCharsets.UTF_8).length; + Assertions.assertTrue(contentBytes <= budget); + + ArrayNode contentArray = JSONUtils.parseArray(alertInfo.getContent()); + Assertions.assertEquals(1, contentArray.size()); + JsonNode only = contentArray.get(0); + Assertions.assertTrue(only.has("...")); + Assertions.assertEquals("1 more rows omitted", only.get("...").asText()); + } + + private String repeatChar(char c, int count) { + StringBuilder builder = new StringBuilder(count); + for (int i = 0; i < count; i++) { + builder.append(c); + } + return builder.toString(); + } + + @Test + void testPrepareTaskResultAlertRespectsConfiguredLimits() throws Exception { + int configuredMaxRows = 3; + int configuredMaxValueLength = 10; + int configuredMaxBytes = 60000; + int totalRows = 10; + + ArrayNode resultArray = JSONUtils.createArrayNode(); + for (int i = 0; i < totalRows; i++) { + resultArray.add(JSONUtils.parseObject("{\"id\":\"" + i + "\"}")); + } + + try (MockedStatic propertyUtils = Mockito.mockStatic(PropertyUtils.class)) { + propertyUtils.when(() -> PropertyUtils.getInt("sql.task.alert.max.rows", 30)).thenReturn(configuredMaxRows); + propertyUtils.when(() -> PropertyUtils.getInt("sql.task.alert.max.row.value.length", 100)) + .thenReturn(configuredMaxValueLength); + propertyUtils.when(() -> PropertyUtils.getInt("sql.task.alert.max.content.bytes", 60000)) + .thenReturn(configuredMaxBytes); + + TaskExecutionContext ctx = createSqlTaskWithAlert(resultArray, 10, "configured limits test", 1); + + TaskAlertInfo alertInfo = ctx.getTaskAlertInfo(); + ArrayNode contentArray = JSONUtils.parseArray(alertInfo.getContent()); + Assertions.assertEquals(configuredMaxRows + 1, contentArray.size()); + JsonNode last = contentArray.get(contentArray.size() - 1); + Assertions.assertTrue(last.has("...")); + Assertions.assertEquals((totalRows - configuredMaxRows) + " more rows omitted", + last.get("...").asText()); + } + } + + private int readStaticIntConstant(String name) throws Exception { + Field field = SqlTask.class.getDeclaredField(name); + field.setAccessible(true); + return field.getInt(null); + } + + @Test + void testResultProcessSendEmailDisabledDoesNotSetNeedAlert() throws Exception { + // Build a SqlTask with sendEmail = false + String taskParams = "{\"type\":\"HIVE\",\"datasource\":1,\"sql\":\"select 1\",\"sendEmail\":false}"; + TaskExecutionContext ctx = new TaskExecutionContext(); + ctx.setTaskParams(taskParams); + ctx.setTaskName("no_alert_task"); + ctx.setResourceParametersHelper(getResourceParametersHelperWithDatasourceType(DbType.HIVE)); + + SqlTask task = new SqlTask(ctx); + + ResultSet mockResultSet = mock(ResultSet.class); + ResultSetMetaData mockMetaData = mock(ResultSetMetaData.class); + when(mockResultSet.getMetaData()).thenReturn(mockMetaData); + when(mockMetaData.getColumnCount()).thenReturn(1); + when(mockMetaData.getColumnLabel(1)).thenReturn("id"); + when(mockResultSet.next()).thenReturn(true, false); + when(mockResultSet.getObject(1)).thenReturn("1"); + + Method resultProcessMethod = SqlTask.class.getDeclaredMethod("resultProcess", ResultSet.class); + resultProcessMethod.setAccessible(true); + resultProcessMethod.invoke(task, mockResultSet); + + Assertions.assertFalse(ctx.isNeedAlert()); + Assertions.assertNull(ctx.getTaskAlertInfo()); + } + + @Test + void testResultProcessEmptyResultSetPreparesAlertWithEmptyRow() throws Exception { + String taskParams = "{\"type\":\"HIVE\",\"datasource\":1,\"sql\":\"select 1\"" + + ",\"sendEmail\":true,\"displayRows\":10,\"groupId\":3,\"title\":\"empty result\"}"; + + TaskExecutionContext ctx = new TaskExecutionContext(); + ctx.setTaskParams(taskParams); + ctx.setTaskName("empty_result_task"); + ctx.setResourceParametersHelper(getResourceParametersHelperWithDatasourceType(DbType.HIVE)); + + SqlTask task = new SqlTask(ctx); + + ResultSet mockResultSet = mock(ResultSet.class); + ResultSetMetaData mockMetaData = mock(ResultSetMetaData.class); + when(mockResultSet.getMetaData()).thenReturn(mockMetaData); + when(mockMetaData.getColumnCount()).thenReturn(2); + when(mockMetaData.getColumnLabel(1)).thenReturn("id"); + when(mockMetaData.getColumnLabel(2)).thenReturn("name"); + when(mockResultSet.next()).thenReturn(false); // empty result set + + Method resultProcessMethod = SqlTask.class.getDeclaredMethod("resultProcess", ResultSet.class); + resultProcessMethod.setAccessible(true); + resultProcessMethod.invoke(task, mockResultSet); + + Assertions.assertTrue(ctx.isNeedAlert()); + + TaskAlertInfo alertInfo = ctx.getTaskAlertInfo(); + Assertions.assertNotNull(alertInfo); + Assertions.assertEquals(3, alertInfo.getAlertGroupId()); + Assertions.assertEquals("empty result", alertInfo.getTitle()); + Assertions.assertEquals(AlertType.TASK_RESULT, alertInfo.getAlertType()); + + ArrayNode contentArray = JSONUtils.parseArray(alertInfo.getContent()); + Assertions.assertEquals(1, contentArray.size()); + Assertions.assertEquals("", contentArray.get(0).get("id").asText()); + Assertions.assertEquals("", contentArray.get(0).get("name").asText()); + } + + @Test + void testTaskExecutionContextCarriesAlertInfoForEventPropagation() { + TaskExecutionContext ctx = new TaskExecutionContext(); + ctx.setNeedAlert(true); + + TaskAlertInfo alertInfo = new TaskAlertInfo(); + alertInfo.setTitle("test title"); + alertInfo.setContent("[{\"id\":\"1\"}]"); + alertInfo.setAlertGroupId(7); + alertInfo.setAlertType(AlertType.TASK_RESULT); + ctx.setTaskAlertInfo(alertInfo); + + Assertions.assertTrue(ctx.isNeedAlert()); + Assertions.assertNotNull(ctx.getTaskAlertInfo()); + Assertions.assertEquals("test title", ctx.getTaskAlertInfo().getTitle()); + Assertions.assertEquals(7, ctx.getTaskAlertInfo().getAlertGroupId()); + Assertions.assertEquals(AlertType.TASK_RESULT, ctx.getTaskAlertInfo().getAlertType()); + + TaskExecutionContext ctx2 = new TaskExecutionContext(); + Assertions.assertFalse(ctx2.isNeedAlert()); + Assertions.assertNull(ctx2.getTaskAlertInfo()); + } + } diff --git a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/executor/PhysicalTaskExecutor.java b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/executor/PhysicalTaskExecutor.java index 7118a099d5f9..7d589599b282 100644 --- a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/executor/PhysicalTaskExecutor.java +++ b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/executor/PhysicalTaskExecutor.java @@ -23,6 +23,7 @@ import org.apache.dolphinscheduler.plugin.storage.api.StorageOperator; import org.apache.dolphinscheduler.plugin.task.api.AbstractTask; import org.apache.dolphinscheduler.plugin.task.api.TaskCallBack; +import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus; import org.apache.dolphinscheduler.plugin.task.api.log.TaskLogMarkers; import org.apache.dolphinscheduler.plugin.task.api.model.ApplicationInfo; import org.apache.dolphinscheduler.plugin.task.api.resource.ResourceContext; @@ -91,6 +92,13 @@ public void updateTaskInstanceInfo(final int taskInstanceId) { @Override protected TaskExecutorState doTrackTaskPluginStatus() { + if (physicalTask.getExitStatus() == TaskExecutionStatus.SUCCESS) { + // Synchronize the legacy needAlert/taskAlertInfo fields written directly + // by AbstractTask subclasses into the context, so that the success + // lifecycle event carries the alert info of third-party plugins. + taskExecutionContext.setNeedAlert(physicalTask.getNeedAlert()); + taskExecutionContext.setTaskAlertInfo(physicalTask.getTaskAlertInfo()); + } return TaskExecutorStateMappings.mapState(physicalTask.getExitStatus()); }