diff --git a/AGENTS.md b/AGENTS.md index 283fa1b..e3603cf 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -27,7 +27,7 @@ Groovy 沙箱默认启用但不是 JVM 隔离：勿把脚本编写权开放给 ## 三、凭据外部化 -真实值一律环境变量注入，禁止写回仓库 yaml：`DATAPOLY_ADMIN_PASSWORD`、`DATAPOLY_REDIS_PASSWORD`、`DATAPOLY_DS_AES_KEY`（轮换前须用旧密钥导出重录）、`DATAPOLY_CORS_ALLOWED_ORIGINS`、compose 的 `MYSQL_ROOT_PASSWORD`/`MYSQL_PASSWORD`（演示默认 123456）。演示凭据 admin/123456、test/test 正式部署必须修改；actuator 已收窄为 health,info。 +真实值一律环境变量注入，禁止写回仓库 yaml：`DATAPOLY_ADMIN_PASSWORD`、`DATAPOLY_REDIS_PASSWORD`、`DATAPOLY_DS_AES_KEY`（轮换前须用旧密钥导出重录）、`DATAPOLY_CORS_ALLOWED_ORIGINS`、compose 的 `MYSQL_ROOT_PASSWORD`/`MYSQL_PASSWORD`（演示默认 123456）。演示凭据 admin/123456、test/test 正式部署必须修改；actuator 已收窄为 health,info。飞书登录凭证 `DATAPOLY_FEISHU_APP_ID`/`DATAPOLY_FEISHU_APP_SECRET` 同样只从环境变量注入：compose 取被忽略的 `install/.env`，发行版 `conf/config.ini` 留空即关闭（`datapolyctl.sh` 只在键有值时导出，空串会让布尔属性宽松绑定失败）。 ## 四、端点防护 @@ -36,7 +36,8 @@ Groovy 沙箱默认启用但不是 JVM 隔离：勿把脚本编写权开放给 ## 五、其他约束 - 新增 JDBC 代码资源必须 try-with-resources；firewall 规则行被删时网关按"全拒绝"处理（fail-closed，属预期）。 -- DataTask 投递 Sink 仓库内置零实现，外部以 Spring Bean / `META-INF/services` 注册（SPI `com.cs.common.datatask.DataTaskSink`）；宿主可自行维护本地扩展：在顶层 `datapoly-extension/`（已被 .gitignore 排除，独立 git 仓库）下用 `backend/` 放 Maven 扩展模块（依赖钉版在模块自身 pom、不进根 reactor，由入库脚本 build-extension.sh 在宿主机 JDK 25 构建后投放 lib-extra/，随发行版装配进各服务 classpath）、`front/` 放扩展 UI；API 扩展点 `ApiAssignmentPostProcessor` 注册方式相同、须同步执行且保持轻量。详见 docs/*/data-task.md。 -- 默认前端扩展目录 `datapoly-extension/front`（同被 .gitignore 排除）经 datapoly-manager-ui 编译期装配：webpack `@extension` 别名自动探测该目录 `src/index.js`（见 build/webpack.base.conf.js）、`src/extension-stub` 为缺省回退、扩展路由与 i18n 词条在 manager-ui 入口深合并——这四处钩子文件（build/webpack.base.conf.js、src/extension-stub、src/router、src/main.js）勿移除或改名；目录不存在时 CI 与普通构建不受影响。front 自带 `package.json` 可直接启动（`npm run dev`，复用宿主 webpack 链，前置为宿主 node_modules 已安装、Node 24——dev server 与生产构建均已实测）。 +- DataTask 投递 Sink 仓库内置零实现，外部以 Spring Bean / `META-INF/services` 注册（SPI `com.cs.common.datatask.DataTaskSink`；服务端导出类投递——如 MaxCompute `UNLOAD` 直写对象存储——另实现可选接口 `DataTaskStatementSink`：`handlesStatement` 判定、`executeStatement` 执行，引擎跳过整条行式管线并在阻塞期间代为续租，取消只在提交前拦得住）；宿主可自行维护本地扩展：在顶层 `datapoly-extension/`（已被 .gitignore 排除，独立 git 仓库）下用 `backend/` 放 Maven 扩展模块（依赖钉版在模块自身 pom、不进根 reactor，由入库脚本 build-extension.sh 在宿主机 JDK 25 构建后投放 lib-extra/，随发行版装配进各服务 classpath）、`front/` 放扩展 UI；API 扩展点 `ApiAssignmentPostProcessor` 注册方式相同、须同步执行且保持轻量。详见 docs/*/data-task.md。 +- 默认前端扩展目录 `datapoly-extension/front`（同被 .gitignore 排除）经 datapoly-manager-ui 编译期装配：webpack `@extension` 别名自动探测该目录 `src/index.js`（见 build/webpack.base.conf.js）、`src/extension-stub` 为缺省回退、扩展路由与 i18n 词条在 manager-ui 入口深合并、登录页扩展区渲染 `@extension` 的 `loginExtras` 组件数组（`src/views/login/index.vue`，stub 为空数组）——这五处钩子（build/webpack.base.conf.js、src/extension-stub、src/router、src/main.js、src/views/login/index.vue 的 loginExtras 挂载点）勿移除或改名；目录不存在时 CI 与普通构建不受影响。front 自带 `package.json` 可直接启动（`npm run dev`，复用宿主 webpack 链，前置为宿主 node_modules 已安装、Node 24——dev server 与生产构建均已实测）。 +- 用户角色见 `DATAPOLY_SYSTEM_USER.user_role`（`ADMIN`/`USER`，v1.4.0 迁移：新行默认 `USER`、存量行回填 `ADMIN`），登录响应 `AccessToken.role` 带回该值；角色只决定界面可见范围，鉴权仍以 token 为准，按角色的端点拦截须同时校验数据库中的角色而非前端传来的值。 - 宿主扩展 jar 经根目录 `lib-extra/` 投放点进入发行版 `lib/common/`（`package.xml` 打包该目录 `*.jar`；目录只占位入库，jar 永不入库）。扩展为独立 git 仓库（内部 GitLab，front+backend 一体）：`build-extension.sh` 按环境变量 `DATAPOLY_EXTENSION_GIT_URL`（真实地址不入库，CI 注入）+ `DATAPOLY_EXTENSION_GIT_REF`（默认 master）浅克隆到 `datapoly-extension/`（仍被 .gitignore 排除），目录已存在则按本地工作区构建（宿主机 JDK 25 优先、低于 25 不可用）、`DATAPOLY_EXTENSION_FORCE_SYNC=1` 强制覆盖本地改动；`build.sh`/`docker-maven-build.sh` 会先调用该脚本，未配置且目录不存在时无操作（纯开源构建零影响）。本地环境变量注入（env.sh）与防误提交钩子集中在被忽略的 `dev-local/`。 - 一次性 token 在校验时即消费（含查库兜底路径；2026-09 修复兜底不消费导致的重放）；并发首用竞态下多 executor 仍可能各放行一次（无分布式锁，已知限制）。 diff --git a/CHANGELOG.md b/CHANGELOG.md index ee724de..15ee6f6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,21 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). ### Added +- DataTask statement sinks (optional capability, backward compatible): the new + interface `com.cs.common.datatask.DataTaskStatementSink` lets a delivery + provider claim a definition and run its rendered statement itself, for + exports that must complete inside the source engine (MaxCompute + `UNLOAD ... INTO LOCATION 'oss://...'`) instead of streaming rows through the + executor. `DataTaskJobEngine` consults the sink once, right after rendering + and before any JDBC work: on a claim it skips the whole row pipeline (no + session, no result set, no row limit or reshaping) and records the returned + `SinkOutcome` like any other artifact. Because such a statement can outlive + `lease-seconds`, a lazily started daemon refreshes the job lease every + `lease-seconds/3` while it blocks; the cancel probe can only stop a statement + before submission, so a late cancel keeps the artifact and is recorded as + `artifactInfo.cancelRequested`. Existing sinks are untouched — a sink that + never claims a definition behaves exactly as before. Documented in + `docs/{zh,en}/data-task.md` §5. - DataTask sink contract (net-neutral extension): `SinkRequest` now carries `columnMetadata` (per-column JDBC type hints, shaped through the same projection as the columns) and the terminal `DataTaskEvent` includes the diff --git a/build-docker/install/docker-compose.yml b/build-docker/install/docker-compose.yml index af7033f..2275be7 100644 --- a/build-docker/install/docker-compose.yml +++ b/build-docker/install/docker-compose.yml @@ -44,6 +44,12 @@ services: DATAPOLY_DS_ENCRYPT: false # 部署后请务必设置：覆盖种子 admin 口令（S5） DATAPOLY_ADMIN_PASSWORD: ${DATAPOLY_ADMIN_PASSWORD:-} + # 飞书登录（datapoly-extension-feishu，relaxed binding 映射为 datapoly.feishu.*）： + # 真实值由 install/.env 注入，勿写回本文件；三项凭证缺一或开关关闭时登录页不显示飞书按钮 + DATAPOLY_FEISHU_ENABLED: ${DATAPOLY_FEISHU_ENABLED:-false} + DATAPOLY_FEISHU_APP_ID: ${DATAPOLY_FEISHU_APP_ID:-} + DATAPOLY_FEISHU_APP_SECRET: ${DATAPOLY_FEISHU_APP_SECRET:-} + DATAPOLY_FEISHU_REDIRECT_URI: ${DATAPOLY_FEISHU_REDIRECT_URI:-} # DATAPOLY_MANAGER_URL: http://www.example.com:8090 # DATAPOLY_GATEWAY_URL: http://www.example.com:8091 depends_on: diff --git a/datapoly-common/src/main/java/com/cs/common/datatask/DataTaskStatementRequest.java b/datapoly-common/src/main/java/com/cs/common/datatask/DataTaskStatementRequest.java new file mode 100644 index 0000000..8312ddc --- /dev/null +++ b/datapoly-common/src/main/java/com/cs/common/datatask/DataTaskStatementRequest.java @@ -0,0 +1,85 @@ +// Use of this source code is governed by a BSD-style license +package com.cs.common.datatask; + +import com.cs.common.enums.ProductTypeEnum; +import lombok.Builder; +import lombok.Data; + +import javax.sql.DataSource; +import java.util.List; +import java.util.Map; +import java.util.function.BooleanSupplier; + +/** + * Everything a {@link DataTaskStatementSink} needs to decide about, and then run, one + * definition's statement outside the engine's row pipeline: the rendered SQL, the + * definition's delivery configuration, the datasource it was authored against and the + * cooperative cancellation probe. + * + *

The engine builds this object right after rendering the template and before any + * JDBC work, so a sink can inspect it in + * {@link DataTaskStatementSink#handlesStatement(DataTaskStatementRequest)} without + * paying for a connection.

+ */ +@Data +@Builder +public class DataTaskStatementRequest { + + private Long jobId; + + private String taskName; + + /** Registration type of the sink this definition selected. */ + private String sinkType; + + /** + * Opaque JSON configuration stored on the task definition; its schema belongs to + * the sink implementation. Secrets placed here are persisted in plain text in the + * meta store — providers should reference server-side credentials instead. + */ + private String sinkConfig; + + /** + * Statement rendered from the definition template. For server-side execution this + * text must be complete: drivers of these engines generally accept no bind + * parameters, so callers author such definitions with {@code ${param}} inlining + * ({@code dollarAllowed}) rather than {@code #{param}}. + */ + private String sql; + + /** + * Bind values the renderer left behind as placeholders; a non-empty list means + * {@link #sql} is not executable as-is and the definition must be rewritten with + * inlined parameters. + */ + private List sqlParameters; + + /** + * Whether {@link #sql} is a query (SELECT/WITH, ignoring leading whitespace, parens and + * comments). Server-side exports wrap the statement in their own command — MaxCompute's + * {@code UNLOAD FROM ()}, for instance — which only accepts a query, so a sink + * claiming such definitions must leave everything else (DML, DDL, engine-specific + * statements) on the ordinary path. + */ + private boolean query; + + /** Boundary values that produced {@link #sql}, for logging and diagnostics. */ + private Map params; + + private Long datasourceId; + + private ProductTypeEnum product; + + /** Pooled datasource of the definition, shared with the synchronous API path. */ + private DataSource dataSource; + + /** Account id of the submitter, as carried by the job record. */ + private String submittedBy; + + /** + * True once the job row left RUNNING or a cancel was requested — the same predicate + * the row pipeline uses. Long-running statements should poll it before submitting; + * an already-submitted server-side job will finish regardless. + */ + private BooleanSupplier cancelled; +} \ No newline at end of file diff --git a/datapoly-common/src/main/java/com/cs/common/datatask/DataTaskStatementSink.java b/datapoly-common/src/main/java/com/cs/common/datatask/DataTaskStatementSink.java new file mode 100644 index 0000000..ae7b3c7 --- /dev/null +++ b/datapoly-common/src/main/java/com/cs/common/datatask/DataTaskStatementSink.java @@ -0,0 +1,40 @@ +// Use of this source code is governed by a BSD-style license +package com.cs.common.datatask; + +/** + * Optional capability of a {@link DataTaskSink}: some deliveries must not be produced by + * this JVM at all. Server-side exports (MaxCompute {@code UNLOAD}, warehouse + * {@code INSERT OVERWRITE DIRECTORY}, ...) complete inside the source engine and write + * straight to the target store, so the engine must hand over the statement instead of + * executing it and pushing rows. + * + *

{@link #handlesStatement(DataTaskStatementRequest)} is consulted once per job, right + * after the template is rendered and before any JDBC execution. Returning {@code true} + * makes the engine skip the row pipeline entirely — no {@link SinkSession}, no row limit, + * no column reshaping — and call {@link #executeStatement(DataTaskStatementRequest)} + * instead; the outcome is recorded on the job like any other sink outcome. Returning + * {@code false} leaves the definition on the ordinary path, so one sink can serve both + * modes (e.g. stream to object storage for most engines, delegate to the engine's own + * export command for one of them).

+ * + *

Implementations own resource cleanup for the delegated path: the engine has no + * session to abort when the statement fails or the job is cancelled mid-flight.

+ */ +public interface DataTaskStatementSink extends DataTaskSink { + + /** + * Non-statement implementations never reach this method — a sink that claims a + * definition via {@link #handlesStatement} is driven through + * {@link #executeStatement} only. Sinks serving both modes override it normally. + */ + @Override + default SinkSession openSession(SinkRequest request) throws Exception { + throw new UnsupportedOperationException( + type() + ": this sink is driven through executeStatement, not openSession"); + } + + /** @return true to have the engine delegate this definition's statement to the sink */ + boolean handlesStatement(DataTaskStatementRequest request); + + SinkOutcome executeStatement(DataTaskStatementRequest request) throws Exception; +} \ No newline at end of file diff --git a/datapoly-common/src/main/java/com/cs/common/dto/AccessToken.java b/datapoly-common/src/main/java/com/cs/common/dto/AccessToken.java index 5e3031d..da3be45 100644 --- a/datapoly-common/src/main/java/com/cs/common/dto/AccessToken.java +++ b/datapoly-common/src/main/java/com/cs/common/dto/AccessToken.java @@ -30,4 +30,7 @@ public class AccessToken implements Serializable { @Schema(description = "有效期(时间段，单位:秒)") private Long expireSeconds; + + @Schema(description = "角色(ADMIN/USER)") + private String role; } diff --git a/datapoly-core/src/main/java/com/cs/core/datatask/DataTaskJobEngine.java b/datapoly-core/src/main/java/com/cs/core/datatask/DataTaskJobEngine.java index 4b0e624..f18fab1 100644 --- a/datapoly-core/src/main/java/com/cs/core/datatask/DataTaskJobEngine.java +++ b/datapoly-core/src/main/java/com/cs/core/datatask/DataTaskJobEngine.java @@ -4,6 +4,8 @@ import com.cs.common.datatask.CellDecorator; import com.cs.common.datatask.ColumnMetadata; import com.cs.common.datatask.DataTaskSink; +import com.cs.common.datatask.DataTaskStatementRequest; +import com.cs.common.datatask.DataTaskStatementSink; import com.cs.common.datatask.SinkOutcome; import com.cs.common.datatask.SinkRequest; import com.cs.common.datatask.SinkSession; @@ -36,10 +38,15 @@ import org.springframework.context.ApplicationEventPublisher; import org.springframework.stereotype.Service; +import jakarta.annotation.PreDestroy; import jakarta.annotation.Resource; import java.io.File; import java.sql.*; import java.util.*; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; import java.util.function.BooleanSupplier; import java.util.function.Consumer; import java.util.function.Function; @@ -52,6 +59,12 @@ * batches through the reshaping plan into the delivery provider session, refreshes * progress/lease while scanning and finalizes the job row together with a Spring * {@link DataTaskEvent} carrying the terminal state. + * + *

A sink that claims the definition as a + * {@link DataTaskStatementSink statement sink} (server-side exports such as MaxCompute + * {@code UNLOAD}) bypasses the row pipeline entirely: the statement is handed over, the + * lease is refreshed from a background keeper while it blocks, and the returned outcome + * is recorded on the job like any other artifact.

*/ @Slf4j @Service @@ -91,6 +104,9 @@ public class DataTaskJobEngine { @Value("${datapoly.data-task.max-rows-default:1000000}") private long maxRowsDefault; + /** Lazily started daemon keeping delegated statements' leases alive; see {@link #runDelegatedStatement}. */ + private volatile ScheduledExecutorService leaseKeeper; + /** Claim exactly one PENDING job for this worker; null once the queue is drained. */ public Long claimNext(String workerAddr) { long now = System.currentTimeMillis(); @@ -116,6 +132,68 @@ protected HikariDataSource loadDataSource(DataSourceEntity dsEntity) { return DataSourceUtils.getHikariDataSource(dsEntity, driverPath.getAbsolutePath()); } + /** + * Runs one statement a {@link DataTaskStatementSink} claimed, refreshing the job lease + * while it blocks: a server-side export routinely outlives the lease window, and the + * reaper would otherwise fail a perfectly healthy job. Progress is not reported here — + * the engine only sees the terminal outcome. + */ + private SinkOutcome runDelegatedStatement(DataTaskStatementSink sink, DataTaskStatementRequest request, + Long jobId) throws Exception { + long intervalMs = Math.max(1000L, leaseSeconds * 1000L / 3); + ScheduledFuture lease = leaseKeeper().scheduleAtFixedRate(() -> refreshLease(jobId), + intervalMs, intervalMs, TimeUnit.MILLISECONDS); + try { + return sink.executeStatement(request); + } finally { + lease.cancel(false); + } + } + + private void refreshLease(Long jobId) { + try { + if (!dataTaskJobDao.heartbeat(jobId, 0L, + new Timestamp(System.currentTimeMillis() + leaseSeconds * 1000L))) { + log.warn("Data task job {} is no longer RUNNING; the delegated statement is left to finish", jobId); + } + } catch (Exception e) { + log.warn("Failed to refresh the lease of data task job {}: {}", jobId, e.getMessage()); + } + } + + private ScheduledExecutorService leaseKeeper() { + ScheduledExecutorService keeper = leaseKeeper; + if (null == keeper) { + synchronized (this) { + keeper = leaseKeeper; + if (null == keeper) { + keeper = Executors.newSingleThreadScheduledExecutor(runnable -> { + Thread thread = new Thread(runnable, "data-task-lease-keeper"); + thread.setDaemon(true); + return thread; + }); + leaseKeeper = keeper; + } + } + } + return keeper; + } + + @PreDestroy + void shutdownLeaseKeeper() { + ScheduledExecutorService keeper = leaseKeeper; + if (null != keeper) { + keeper.shutdownNow(); + } + } + + /** Cooperative-cancel predicate shared by the row pipeline and delegated statements. */ + private boolean isJobCancelled(Long jobId) { + DataTaskJobEntity current = dataTaskJobDao.getById(jobId); + return null == current || !Objects.equals(current.getStatus(), DataTaskStatus.RUNNING) + || Boolean.TRUE.equals(current.getCancelRequested()); + } + public void run(Long jobId) { DataTaskJobEntity job = dataTaskJobDao.getById(jobId); if (null == job) { @@ -157,43 +235,83 @@ public void run(Long jobId) { "datatask.sink.unknown", snapshot.getSinkType()); } - channel = new SessionChannel(dataTaskJobDao, jobId, snapshot, - sinkHolder.get(), job); - StreamSpec streamSpec = StreamSpec.builder() - .dataSource(dataSource) - .product(dsEntity.getType()) - .sqlMeta(sqlMeta) - .naming(null == snapshot.getNamingStrategy() - ? NamingStrategyEnum.CAMEL_CASE : snapshot.getNamingStrategy()) - .cancelSupplier(channel::isJobCancelled) - .rowLimit(effectiveMaxRows(snapshot.getMaxRows())) - .flushIntervalMs(flushIntervalMs) - .fetchSize(fetchSize) - .timeoutSeconds(queryTimeoutSeconds) - .progress(rows -> onProgress(jobId, rows)) - .build(); + DataTaskSink resolved = sinkHolder.get(); + boolean delegated = false; + if (resolved instanceof DataTaskStatementSink) { + // server-side delivery: the sink runs the statement itself (the data never + // travels through this process), so the row pipeline stays out of the way + DataTaskStatementSink statementSink = (DataTaskStatementSink) resolved; + DataTaskStatementRequest statementRequest = DataTaskStatementRequest.builder() + .jobId(jobId) + .taskName(job.getDefName()) + .sinkType(snapshot.getSinkType()) + .sinkConfig(snapshot.getSinkConfig()) + .sql(sqlMeta.getSql()) + .sqlParameters(sqlMeta.getParameter()) + .query(sqlMeta.isQuerySQL()) + .params(params) + .datasourceId(dsEntity.getId()) + .product(dsEntity.getType()) + .dataSource(dataSource) + .submittedBy(job.getSubmittedBy()) + .cancelled(() -> isJobCancelled(jobId)) + .build(); + if (statementSink.handlesStatement(statementRequest)) { + delegated = true; + SinkOutcome outcome = runDelegatedStatement(statementSink, statementRequest, jobId); + artifactInfo.put("statementDelegated", Boolean.TRUE); + if (null != outcome) { + artifactUri = outcome.getArtifactUri(); + if (null != outcome.getInfo()) { + artifactInfo.putAll(outcome.getInfo()); + } + } + if (isRunningAndCancelled(jobId)) { + // an accepted server-side export cannot be stopped: the payload is + // already in the target store, so the artifact is kept and the late + // cancel is recorded rather than dropped + artifactInfo.put("cancelRequested", Boolean.TRUE); + } + } + } - StreamResult result = streamQuery(streamSpec, channel); - delivered = result.getRows(); - session = channel.session(); - artifactInfo.put("sinkStopped", channel.isSinkAskedStop()); - - if (!result.isRowset()) { - artifactInfo.put("updateCount", result.getUpdateCount()); - terminal = DataTaskStatus.SUCCESS; - } else { - SinkOutcome outcome = session.complete(); - session = null; // ownership moved into complete(); nothing left to abort - if (null != outcome) { - artifactUri = outcome.getArtifactUri(); - if (null != outcome.getInfo()) { - artifactInfo.putAll(outcome.getInfo()); + if (!delegated) { + channel = new SessionChannel(jobId, snapshot, resolved, job); + StreamSpec streamSpec = StreamSpec.builder() + .dataSource(dataSource) + .product(dsEntity.getType()) + .sqlMeta(sqlMeta) + .naming(null == snapshot.getNamingStrategy() + ? NamingStrategyEnum.CAMEL_CASE : snapshot.getNamingStrategy()) + .cancelSupplier(channel::isJobCancelled) + .rowLimit(effectiveMaxRows(snapshot.getMaxRows())) + .flushIntervalMs(flushIntervalMs) + .fetchSize(fetchSize) + .timeoutSeconds(queryTimeoutSeconds) + .progress(rows -> onProgress(jobId, rows)) + .build(); + + StreamResult result = streamQuery(streamSpec, channel); + delivered = result.getRows(); + session = channel.session(); + artifactInfo.put("sinkStopped", channel.isSinkAskedStop()); + + if (!result.isRowset()) { + artifactInfo.put("updateCount", result.getUpdateCount()); + } else { + SinkOutcome outcome = session.complete(); + session = null; // ownership moved into complete(); nothing left to abort + if (null != outcome) { + artifactUri = outcome.getArtifactUri(); + if (null != outcome.getInfo()) { + artifactInfo.putAll(outcome.getInfo()); + } } + artifactInfo.put("truncated", result.isTruncated()); + artifactInfo.put("deliveredRows", delivered); } - artifactInfo.put("truncated", result.isTruncated()); - artifactInfo.put("deliveredRows", delivered); - terminal = DataTaskStatus.SUCCESS; } + terminal = DataTaskStatus.SUCCESS; dataTaskJobDao.finishSuccess(jobId, delivered, artifactUri, JsonUtils.toJsonString(artifactInfo), now()); @@ -480,7 +598,6 @@ public static class StreamSpec { } private class SessionChannel implements ResultChannel { - private final DataTaskJobDao jobDao; private final Long jobId; private final DataTaskDefEntity spec; private final DataTaskSink sink; @@ -489,9 +606,7 @@ private class SessionChannel implements ResultChannel { private volatile SinkSession sessionRef; private volatile boolean sinkAskedStop; - SessionChannel(DataTaskJobDao jobDao, Long jobId, DataTaskDefEntity spec, - DataTaskSink sink, DataTaskJobEntity job) { - this.jobDao = jobDao; + SessionChannel(Long jobId, DataTaskDefEntity spec, DataTaskSink sink, DataTaskJobEntity job) { this.jobId = jobId; this.spec = spec; this.sink = sink; @@ -507,9 +622,7 @@ boolean isSinkAskedStop() { } boolean isJobCancelled() { - DataTaskJobEntity current = jobDao.getById(jobId); - return null == current || !Objects.equals(current.getStatus(), DataTaskStatus.RUNNING) - || Boolean.TRUE.equals(current.getCancelRequested()); + return DataTaskJobEngine.this.isJobCancelled(jobId); } @Override diff --git a/datapoly-core/src/main/java/com/cs/core/datatask/DataTaskParamBinder.java b/datapoly-core/src/main/java/com/cs/core/datatask/DataTaskParamBinder.java index 83bc4b4..184084d 100644 --- a/datapoly-core/src/main/java/com/cs/core/datatask/DataTaskParamBinder.java +++ b/datapoly-core/src/main/java/com/cs/core/datatask/DataTaskParamBinder.java @@ -5,6 +5,7 @@ import com.cs.common.dto.ItemParam; import com.cs.common.exception.CommonException; import com.cs.common.exception.ResponseErrorCode; +import com.cs.persistence.util.JsonUtils; import org.apache.commons.lang3.StringUtils; import java.util.*; @@ -46,19 +47,60 @@ private static Object bindScalarRoot(ItemParam decl, Map body, b if (!isArray) { return coerceOrNull(decl, raw, decl.getName(), Boolean.TRUE.equals(decl.getRequired())); } - List values = raw instanceof List ? (List) raw : null; + List values = arrayElements(decl, raw, decl.getName(), true); if (null == values || values.isEmpty()) { return requireOrNothing(decl, raw, decl.getName(), () -> defaultList(decl)); } - List out = new ArrayList<>(values.size()); - for (int i = 0; i < values.size(); i++) { - out.add(coerce(decl.getType(), values.get(i), - String.format("%s[%d]", decl.getName(), i), true)); + return values; + } + + /** + * Element list of an array-declared parameter. Besides a native JSON array, two + * string wire formats that frontends commonly emit are tolerated: a pre-serialized + * JSON array literal (["a","b"]) and a comma-separated list ("a,b"). Every element + * is still coerced to the declared type, so shape loosening does not weaken value + * validation; a String that carries nothing usable yields null for the caller. + */ + private static List arrayElements(BaseParam decl, Object raw, String path, boolean enforce) { + List rawElements; + if (raw instanceof List) { + rawElements = (List) raw; + } else if (raw instanceof String && StringUtils.isNotBlank((String) raw)) { + rawElements = parseArrayString(StringUtils.trim((String) raw)); + } else { + return null; + } + List out = new ArrayList<>(rawElements.size()); + for (int i = 0; i < rawElements.size(); i++) { + out.add(coerce(decl.getType(), rawElements.get(i), + String.format("%s[%d]", path, i), enforce)); } return out; } + private static List parseArrayString(String value) { + if (value.startsWith("[") && value.endsWith("]")) { + try { + return JsonUtils.toBeanList(value, Object.class); + } catch (RuntimeException ignore) { + // not a JSON array after all; fall back to comma splitting + } + } + String[] parts = value.split(","); + List values = new ArrayList<>(parts.length); + for (String part : parts) { + String item = part.trim(); + if (item.length() >= 2 && item.startsWith("\"") && item.endsWith("\"")) { + item = item.substring(1, item.length() - 1); + } + if (!item.isEmpty()) { + values.add(item); + } + } + return values; + } + /** * OBJECT parameters accept either a nested map under the parameter name or flat * {@code name.child} keys on the request body; the nested form wins on conflicts. @@ -92,19 +134,14 @@ private static Object bindObjectChildren(ItemParam decl, Map bod for (BaseParam child : children) { String path = root + "." + child.getName(); if (Boolean.TRUE.equals(child.getIsArray())) { - List raws = container.get(child.getName()) instanceof List - ? (List) container.get(child.getName()) : null; - if (null == raws || raws.isEmpty()) { + List items = arrayElements(child, container.get(child.getName()), path, + Boolean.TRUE.equals(child.getRequired())); + if (null == items || items.isEmpty()) { if (Boolean.TRUE.equals(child.getRequired())) { throw missing(path); } continue; } - List items = new ArrayList<>(raws.size()); - for (int i = 0; i < raws.size(); i++) { - items.add(coerce(child.getType(), raws.get(i), String.format("%s[%d]", path, i), - Boolean.TRUE.equals(child.getRequired()))); - } out.put(child.getName(), items); } else { Object value = coerceOrNull(child, container.get(child.getName()), path, diff --git a/datapoly-core/src/main/java/com/cs/core/service/SystemUserService.java b/datapoly-core/src/main/java/com/cs/core/service/SystemUserService.java index dcbbc7a..5e2e3d3 100644 --- a/datapoly-core/src/main/java/com/cs/core/service/SystemUserService.java +++ b/datapoly-core/src/main/java/com/cs/core/service/SystemUserService.java @@ -46,7 +46,7 @@ public AccessToken login(String username, String password) { String token = TokenUtils.generateValue(); CacheUtils.put(token, user); AccessToken accessTokenWrapper = new AccessToken(user.getRealName(), user.getUsername(), token, - System.currentTimeMillis() / 1000, CacheUtils.CACHE_DURATION_SECONDS); + System.currentTimeMillis() / 1000, CacheUtils.CACHE_DURATION_SECONDS, user.getRole()); return accessTokenWrapper; } diff --git a/datapoly-dist/src/main/assembly/bin/datapolyctl.sh b/datapoly-dist/src/main/assembly/bin/datapolyctl.sh index 938d745..9176db1 100644 --- a/datapoly-dist/src/main/assembly/bin/datapolyctl.sh +++ b/datapoly-dist/src/main/assembly/bin/datapolyctl.sh @@ -55,6 +55,20 @@ export JSON_TIMEZONE=$(get_config_value "JSON_TIMEZONE" "${APP_CONF_PATH}/config export DATAPOLY_MANAGER_URL=$(get_config_value "DATAPOLY_MANAGER_URL" "${APP_CONF_PATH}/config.ini") export DATAPOLY_GATEWAY_URL=$(get_config_value "DATAPOLY_GATEWAY_URL" "${APP_CONF_PATH}/config.ini") +# 飞书登录（datapoly-extension-feishu 扩展模块）：config.ini 里留空即关闭。仅在确实 +# 配置了值时才导出——否则会把 enabled 这类布尔属性导出成空串，宽松绑定会因无法转换而启动失败 +export_feishu_config() { + local key=$1 + local value=$(get_config_value "$key" "${APP_CONF_PATH}/config.ini") + if [ -n "$value" ]; then + export "$key=$value" + fi +} +export_feishu_config "DATAPOLY_FEISHU_ENABLED" +export_feishu_config "DATAPOLY_FEISHU_APP_ID" +export_feishu_config "DATAPOLY_FEISHU_APP_SECRET" +export_feishu_config "DATAPOLY_FEISHU_REDIRECT_URI" + # JVM参数可以在这里设置 # 堆 4G、年轻代/老年代 1:3：长驻对象（Hazelcast token/API 响应缓存、Eureka、Spring 框架 # 对象）占堆内大头，老年代空间优先；年轻代 1G 足以容纳数据任务流式批次与 ≤200 行的 diff --git a/datapoly-dist/src/main/assembly/conf/config.ini b/datapoly-dist/src/main/assembly/conf/config.ini index d2a4885..a09f0be 100644 --- a/datapoly-dist/src/main/assembly/conf/config.ini +++ b/datapoly-dist/src/main/assembly/conf/config.ini @@ -45,4 +45,14 @@ DATAPOLY_DS_ENCRYPT=false # 外部配置化网关/管理地址，默认为空 # DATAPOLY_MANAGER_URL=http://www.example.com:8090 -# DATAPOLY_GATEWAY_URL=http://www.example.com:8091 \ No newline at end of file +# DATAPOLY_GATEWAY_URL=http://www.example.com:8091 + +# 飞书登录（datapoly-extension-feishu 扩展模块；未装配该扩展时全部留空即可） +# App ID / App Secret 取自飞书开发者后台「凭证与基础信息」；真实值请只填在部署现场， +# 不要写回仓库 +# DATAPOLY_FEISHU_ENABLED=true +# DATAPOLY_FEISHU_APP_ID= +# DATAPOLY_FEISHU_APP_SECRET= +# 回调地址必须与开发者后台「安全设置 → 重定向 URL」逐字一致，取对外可达的网关地址； +# 未配置或与登记值不符时飞书会直接跳失败页，登录页则不显示飞书按钮 +# DATAPOLY_FEISHU_REDIRECT_URI=https://www.example.com:8091/user/feishu/callback \ No newline at end of file diff --git a/datapoly-manager-ui/src/extension-stub/index.js b/datapoly-manager-ui/src/extension-stub/index.js index 9672539..702e89e 100644 --- a/datapoly-manager-ui/src/extension-stub/index.js +++ b/datapoly-manager-ui/src/extension-stub/index.js @@ -1,8 +1,11 @@ // Use of this source code is governed by a BSD-style license // Compile-time extension fallback: the webpack '@extension' alias resolves here when -// ../../datapoly-extension/front/src is absent, keeping router and i18n assembly no-ops. +// ../../datapoly-extension/front/src is absent, keeping router, i18n and login-page +// assembly no-ops. export default { routes: [], + // Components rendered below the password login button by src/views/login/index.vue. + loginExtras: [], i18n: { 'zh-CN': {}, 'en-US': {} diff --git a/datapoly-manager-ui/src/views/login/index.vue b/datapoly-manager-ui/src/views/login/index.vue index 4da167d..6938a3f 100644 --- a/datapoly-manager-ui/src/views/login/index.vue +++ b/datapoly-manager-ui/src/views/login/index.vue @@ -43,6 +43,16 @@ + + +

© {{ currentYear }} The DataPoly Authors

@@ -51,6 +61,8 @@