Skip to content

Commit 1d9a8b3

Browse files
fcb-xiaobofcb-xiaobo
authored
[Feature] [rest-api] Support Rest Api to upload file and submit task (#8442)
1 parent 7964c9d commit 1d9a8b3

File tree

9 files changed

+258
-1
lines changed

9 files changed

+258
-1
lines changed

‎docs/en/seatunnel-engine/rest-api-v2.md‎

Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -483,6 +483,40 @@ sink {
483483

484484
------------------------------------------------------------------------------------------
485485

486+
### Submit A Job By Upload Config File
487+
488+
<details>
489+
<summary><code>POST</code> <code><b>/submit-job/upload</b></code> <code>(Returns jobId and jobName if job submitted successfully.)</code></summary>
490+
491+
#### Parameters
492+
493+
> | name | type | data type | description |
494+
> |----------------------|----------|-----------|-----------------------------------|
495+
> | jobId | optional | string | job id |
496+
> | jobName | optional | string | job name |
497+
> | isStartWithSavePoint | optional | string | if job is started with save point |
498+
499+
#### Request Body
500+
The name of the uploaded file key is config_file, and the file suffix json is parsed in json format. The conf or config file suffix is parsed in hocon format
501+
502+
curl Example :
503+
```
504+
curl --location 'http://127.0.0.1:8080/submit-job/upload' --form 'config_file=@"/temp/fake_to_console.conf"'
505+
506+
```
507+
#### Responses
508+
509+
```json
510+
{
511+
"jobId": 733584788375666689,
512+
"jobName": "SeaTunnel_Job"
513+
}
514+
```
515+
516+
</details>
517+
518+
------------------------------------------------------------------------------------------
519+
486520
### Batch Submit Jobs
487521

488522
<details>

‎docs/zh/seatunnel-engine/rest-api-v2.md‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -463,7 +463,40 @@ sink {
463463
</details>
464464

465465
------------------------------------------------------------------------------------------
466+
### 提交作业来源上传配置文件
466467

468+
<details>
469+
<summary><code>POST</code> <code><b>/submit-job</b></code> <code>(如果作业提交成功，返回jobId和jobName。)</code></summary>
470+
471+
#### 参数
472+
473+
> | 参数名称 | 是否必传 | 参数类型 | 参数描述 |
474+
> |----------------------|----------|-----------------------------------|-----------------------------------|
475+
> | jobId | optional | string | job id |
476+
> | jobName | optional | string | job name |
477+
> | isStartWithSavePoint | optional | string | if job is started with save point |
478+
479+
#### 请求体
480+
上传文件key的名称是config_file,文件后缀json的按照json格式来解析,conf或config文件后缀按照hocon格式解析
481+
482+
curl Example
483+
484+
```
485+
curl --location 'http://127.0.0.1:8080/submit-job/upload' --form 'config_file=@"/temp/fake_to_console.conf"'
486+
487+
```
488+
#### 响应
489+
490+
```json
491+
{
492+
"jobId": 733584788375666689,
493+
"jobName": "SeaTunnel_Job"
494+
}
495+
```
496+
497+
</details>
498+
499+
------------------------------------------------------------------------------------------
467500

468501
### 批量提交作业
469502

‎seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/java/org/apache/seatunnel/engine/e2e/ClusterSeaTunnelEngineContainer.java‎

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -37,7 +37,9 @@
3737
import io.restassured.response.Response;
3838
import scala.Tuple3;
3939

40+
import java.io.File;
4041
import java.io.IOException;
42+
import java.net.URL;
4143
import java.nio.file.Path;
4244
import java.nio.file.Paths;
4345
import java.util.ArrayList;
@@ -253,6 +255,34 @@ public void testStartWithSavePointWithoutJobIdV2() {
253255
});
254256
}
255257

258+
@Test
259+
public void testRestApiSubmitJobByUploadFileV2() {
260+
Arrays.asList(server, secondServer)
261+
.forEach(
262+
container -> {
263+
Tuple3<Integer, String, Long> task = tasks.get(1);
264+
URL resource =
265+
this.getClass().getClassLoader().getResource("upload-file");
266+
File fileDirect = new File(resource.getFile());
267+
File[] files = fileDirect.listFiles();
268+
for (File file : files) {
269+
Response response =
270+
given().multiPart("config_file", file)
271+
.baseUri(
272+
http
273+
+ container.getHost()
274+
+ colon
275+
+ task._1())
276+
.basePath(
277+
RestConstant
278+
.REST_URL_SUBMIT_JOB_BY_UPLOAD_FILE)
279+
.when()
280+
.post();
281+
Assertions.assertEquals(200, response.getStatusCode());
282+
}
283+
});
284+
}
285+
256286
@Test
257287
public void testStopJob() {
258288
AtomicInteger i = new AtomicInteger();

‎seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/resources/upload-file/fake_to_console.conf‎

Lines changed: 48 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,48 @@
1+
2+
#
3+
# Licensed to the Apache Software Foundation (ASF) under one or more
4+
# contributor license agreements. See the NOTICE file distributed with
5+
# this work for additional information regarding copyright ownership.
6+
# The ASF licenses this file to You under the Apache License, Version 2.0
7+
# (the "License"); you may not use this file except in compliance with
8+
# the License. You may obtain a copy of the License at
9+
#
10+
# http://www.apache.org/licenses/LICENSE-2.0
11+
#
12+
# Unless required by applicable law or agreed to in writing, software
13+
# distributed under the License is distributed on an "AS IS" BASIS,
14+
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
# See the License for the specific language governing permissions and
16+
# limitations under the License.
17+
#
18+
######
19+
###### This config file is a demonstration of streaming processing in seatunnel config
20+
######
21+
22+
env {
23+
parallelism = 1
24+
job.mode = "BATCH"
25+
}
26+
27+
source {
28+
# This is a example source plugin **only for test and demonstrate the feature source plugin**
29+
FakeSource {
30+
plugin_output = "fake"
31+
parallelism = 1
32+
schema = {
33+
fields {
34+
name = "string"
35+
age = "int"
36+
}
37+
}
38+
}
39+
}
40+
41+
transform {
42+
}
43+
44+
sink {
45+
console {
46+
plugin_input="fake"
47+
}
48+
}

‎seatunnel-e2e/seatunnel-engine-e2e/connector-seatunnel-e2e-base/src/test/resources/upload-file/fake_to_console.json‎

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,27 @@
1+
{
2+
"env": {
3+
"job.mode": "batch"
4+
},
5+
"source": [
6+
{
7+
"plugin_name": "FakeSource",
8+
"plugin_output": "fake",
9+
"row.num": 100,
10+
"schema": {
11+
"fields": {
12+
"name": "string",
13+
"age": "int",
14+
"card": "int"
15+
}
16+
}
17+
}
18+
],
19+
"transform": [
20+
],
21+
"sink": [
22+
{
23+
"plugin_name": "Console",
24+
"plugin_input": ["fake"]
25+
}
26+
]
27+
}

‎seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/JettyService.java‎

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,7 @@
3737
import org.apache.seatunnel.engine.server.rest.servlet.RunningThreadsServlet;
3838
import org.apache.seatunnel.engine.server.rest.servlet.StopJobServlet;
3939
import org.apache.seatunnel.engine.server.rest.servlet.StopJobsServlet;
40+
import org.apache.seatunnel.engine.server.rest.servlet.SubmitJobByUploadFileServlet;
4041
import org.apache.seatunnel.engine.server.rest.servlet.SubmitJobServlet;
4142
import org.apache.seatunnel.engine.server.rest.servlet.SubmitJobsServlet;
4243
import org.apache.seatunnel.engine.server.rest.servlet.SystemMonitoringServlet;
@@ -47,6 +48,7 @@
4748
import lombok.extern.slf4j.Slf4j;
4849

4950
import javax.servlet.DispatcherType;
51+
import javax.servlet.MultipartConfigElement;
5052

5153
import java.io.IOException;
5254
import java.net.DatagramSocket;
@@ -70,6 +72,7 @@
7072
import static org.apache.seatunnel.engine.server.rest.RestConstant.REST_URL_STOP_JOBS;
7173
import static org.apache.seatunnel.engine.server.rest.RestConstant.REST_URL_SUBMIT_JOB;
7274
import static org.apache.seatunnel.engine.server.rest.RestConstant.REST_URL_SUBMIT_JOBS;
75+
import static org.apache.seatunnel.engine.server.rest.RestConstant.REST_URL_SUBMIT_JOB_BY_UPLOAD_FILE;
7376
import static org.apache.seatunnel.engine.server.rest.RestConstant.REST_URL_SYSTEM_MONITORING_INFORMATION;
7477
import static org.apache.seatunnel.engine.server.rest.RestConstant.REST_URL_THREAD_DUMP;
7578
import static org.apache.seatunnel.engine.server.rest.RestConstant.REST_URL_UPDATE_TAGS;
@@ -122,6 +125,9 @@ public void createJettyServer() {
122125
ServletHolder threadDumpHolder = new ServletHolder(new ThreadDumpServlet(nodeEngine));
123126

124127
ServletHolder submitJobHolder = new ServletHolder(new SubmitJobServlet(nodeEngine));
128+
ServletHolder submitJobByUploadFileHolder =
129+
new ServletHolder(new SubmitJobByUploadFileServlet(nodeEngine));
130+
125131
ServletHolder submitJobsHolder = new ServletHolder(new SubmitJobsServlet(nodeEngine));
126132
ServletHolder stopJobHolder = new ServletHolder(new StopJobServlet(nodeEngine));
127133
ServletHolder stopJobsHolder = new ServletHolder(new StopJobsServlet(nodeEngine));
@@ -147,7 +153,10 @@ public void createJettyServer() {
147153
context.addServlet(jobInfoHolder, convertUrlToPath(REST_URL_JOB_INFO));
148154
context.addServlet(jobInfoHolder, convertUrlToPath(REST_URL_RUNNING_JOB));
149155
context.addServlet(threadDumpHolder, convertUrlToPath(REST_URL_THREAD_DUMP));
150-
156+
MultipartConfigElement multipartConfigElement = new MultipartConfigElement("");
157+
submitJobByUploadFileHolder.getRegistration().setMultipartConfig(multipartConfigElement);
158+
context.addServlet(
159+
submitJobByUploadFileHolder, convertUrlToPath(REST_URL_SUBMIT_JOB_BY_UPLOAD_FILE));
151160
context.addServlet(submitJobHolder, convertUrlToPath(REST_URL_SUBMIT_JOB));
152161
context.addServlet(submitJobsHolder, convertUrlToPath(REST_URL_SUBMIT_JOBS));
153162
context.addServlet(stopJobHolder, convertUrlToPath(REST_URL_STOP_JOB));

‎seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/RestConstant.java‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -75,6 +75,9 @@ public class RestConstant {
7575
public static final String REST_URL_SYSTEM_MONITORING_INFORMATION =
7676
"/system-monitoring-information";
7777
public static final String REST_URL_SUBMIT_JOB = "/submit-job";
78+
79+
public static final String REST_URL_SUBMIT_JOB_BY_UPLOAD_FILE = "/submit-job/upload";
80+
7881
public static final String REST_URL_SUBMIT_JOBS = "/submit-jobs";
7982
public static final String REST_URL_STOP_JOB = "/stop-job";
8083
public static final String REST_URL_STOP_JOBS = "/stop-jobs";

‎seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/service/JobInfoService.java‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -173,6 +173,15 @@ public JsonObject submitJob(Map<String, String> requestParams, byte[] requestBod
173173
return submitJobInternal(config, requestParams, seaTunnelServer, nodeEngine.getNode());
174174
}
175175

176+
public JsonObject submitJob(Map<String, String> requestParams, Config config) {
177+
if (Boolean.parseBoolean(requestParams.get(RestConstant.IS_START_WITH_SAVE_POINT))
178+
&& requestParams.get(RestConstant.JOB_ID) == null) {
179+
throw new IllegalArgumentException("Please provide jobId when start with save point.");
180+
}
181+
SeaTunnelServer seaTunnelServer = getSeaTunnelServer(false);
182+
return submitJobInternal(config, requestParams, seaTunnelServer, nodeEngine.getNode());
183+
}
184+
176185
public JsonArray submitJobs(byte[] requestBody) {
177186
List<Tuple2<Map<String, String>, Config>> configTuples =
178187
RestUtil.buildConfigList(requestHandle(requestBody), false);

‎seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/servlet/SubmitJobByUploadFileServlet.java‎

Lines changed: 64 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,64 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.seatunnel.engine.server.rest.servlet;
19+
20+
import org.apache.seatunnel.shade.com.typesafe.config.Config;
21+
import org.apache.seatunnel.shade.com.typesafe.config.ConfigFactory;
22+
import org.apache.seatunnel.shade.com.typesafe.config.ConfigParseOptions;
23+
import org.apache.seatunnel.shade.com.typesafe.config.ConfigSyntax;
24+
25+
import org.apache.seatunnel.engine.server.rest.service.JobInfoService;
26+
27+
import org.apache.commons.io.IOUtils;
28+
29+
import com.hazelcast.spi.impl.NodeEngineImpl;
30+
31+
import javax.servlet.ServletException;
32+
import javax.servlet.http.HttpServletRequest;
33+
import javax.servlet.http.HttpServletResponse;
34+
import javax.servlet.http.Part;
35+
36+
import java.io.IOException;
37+
import java.nio.charset.StandardCharsets;
38+
39+
public class SubmitJobByUploadFileServlet extends BaseServlet {
40+
private final JobInfoService jobInfoService;
41+
42+
public SubmitJobByUploadFileServlet(NodeEngineImpl nodeEngine) {
43+
super(nodeEngine);
44+
this.jobInfoService = new JobInfoService(nodeEngine);
45+
}
46+
47+
@Override
48+
public void doPost(HttpServletRequest req, HttpServletResponse resp)
49+
throws IOException, ServletException {
50+
51+
Part filePart = req.getPart("config_file");
52+
String submittedFileName = filePart.getSubmittedFileName();
53+
String content = IOUtils.toString(filePart.getInputStream(), StandardCharsets.UTF_8);
54+
Config config;
55+
if (submittedFileName.endsWith(".json")) {
56+
config =
57+
ConfigFactory.parseString(
58+
content, ConfigParseOptions.defaults().setSyntax(ConfigSyntax.JSON));
59+
} else {
60+
config = ConfigFactory.parseString(content);
61+
}
62+
writeJson(resp, jobInfoService.submitJob(getParameterMap(req), config));
63+
}
64+
}

0 commit comments

Comments
 (0)