Skip to content

Commit fd806ec

Browse files
xincai98xincai98
authored
Engine call console finish (#531)
* engine call console after task finish * engine call console after task finish * engine call console after task finish * engine call console after task finish * fix ut
1 parent 255a3b3 commit fd806ec

9 files changed

Lines changed: 337 additions & 32 deletions

File tree

‎geaflow/geaflow-core/geaflow-engine/geaflow-cluster/src/main/java/com/antgroup/geaflow/cluster/client/callback/HttpRequest.java‎

Lines changed: 53 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,53 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with 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,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
package com.antgroup.geaflow.cluster.client.callback;
21+
22+
import java.io.Serializable;
23+
24+
class HttpRequest implements Serializable {
25+
private static final long serialVersionUID = 0L;
26+
private boolean success;
27+
private String message;
28+
private Object data;
29+
30+
public boolean isSuccess() {
31+
return success;
32+
}
33+
34+
public void setSuccess(boolean success) {
35+
this.success = success;
36+
}
37+
38+
public String getMessage() {
39+
return message;
40+
}
41+
42+
public void setMessage(String message) {
43+
this.message = message;
44+
}
45+
46+
public Object getData() {
47+
return data;
48+
}
49+
50+
public void setData(Object data) {
51+
this.data = data;
52+
}
53+
}

‎geaflow/geaflow-core/geaflow-engine/geaflow-cluster/src/main/java/com/antgroup/geaflow/cluster/client/callback/JobOperatorCallback.java‎

Lines changed: 47 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,47 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with 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,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
package com.antgroup.geaflow.cluster.client.callback;
21+
22+
import java.util.Map;
23+
24+
public interface JobOperatorCallback {
25+
26+
/**
27+
* The callback for job finish succeed.
28+
*/
29+
void onFinish();
30+
31+
class JobOperatorMeta {
32+
private final Map<String, String> params;
33+
34+
public JobOperatorMeta(Map<String, String> params) {
35+
this.params = params;
36+
}
37+
38+
public Map<String, String> getParams() {
39+
return params;
40+
}
41+
42+
@Override
43+
public String toString() {
44+
return "JobOperatorMeta{" + "params='" + params + '}';
45+
}
46+
}
47+
}

‎geaflow/geaflow-core/geaflow-engine/geaflow-cluster/src/main/java/com/antgroup/geaflow/cluster/client/callback/JobOperatorCallbackFactory.java‎

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,37 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with 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,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
package com.antgroup.geaflow.cluster.client.callback;
21+
22+
import static com.antgroup.geaflow.common.config.keys.ExecutionConfigKeys.GEAFLOW_GW_ENDPOINT;
23+
24+
import com.antgroup.geaflow.common.config.Configuration;
25+
import org.apache.commons.lang3.StringUtils;
26+
27+
public class JobOperatorCallbackFactory {
28+
29+
public static JobOperatorCallback createJobOperatorCallback(Configuration configuration) {
30+
String callbackUrl = configuration.getString(GEAFLOW_GW_ENDPOINT, "");
31+
if (StringUtils.isNotBlank(callbackUrl)) {
32+
return new RestJobOperatorCallback(configuration, callbackUrl);
33+
}
34+
return new SimpleJobOperatorCallback();
35+
}
36+
37+
}

‎geaflow/geaflow-core/geaflow-engine/geaflow-cluster/src/main/java/com/antgroup/geaflow/cluster/client/callback/RestClusterStartedCallback.java‎

Lines changed: 1 addition & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -23,7 +23,7 @@
2323
import com.antgroup.geaflow.common.config.keys.DSLConfigKeys;
2424
import com.antgroup.geaflow.utils.HttpUtil;
2525
import com.google.gson.Gson;
26-
import java.io.Serializable;
26+
2727
import java.util.HashMap;
2828
import java.util.Map;
2929

@@ -57,34 +57,4 @@ public void onFailure(Throwable e) {
5757
HttpUtil.post(callbackUrl, new Gson().toJson(request), headers);
5858
}
5959

60-
static class HttpRequest implements Serializable {
61-
private boolean success;
62-
private String message;
63-
private Object data;
64-
65-
public boolean isSuccess() {
66-
return success;
67-
}
68-
69-
public void setSuccess(boolean success) {
70-
this.success = success;
71-
}
72-
73-
public String getMessage() {
74-
return message;
75-
}
76-
77-
public void setMessage(String message) {
78-
this.message = message;
79-
}
80-
81-
public Object getData() {
82-
return data;
83-
}
84-
85-
public void setData(Object data) {
86-
this.data = data;
87-
}
88-
}
89-
9060
}

‎geaflow/geaflow-core/geaflow-engine/geaflow-cluster/src/main/java/com/antgroup/geaflow/cluster/client/callback/RestJobOperatorCallback.java‎

Lines changed: 80 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,80 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with 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,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
package com.antgroup.geaflow.cluster.client.callback;
21+
22+
import static com.antgroup.geaflow.common.config.keys.ExecutionConfigKeys.JOB_UNIQUE_ID;
23+
24+
import com.antgroup.geaflow.common.config.Configuration;
25+
import com.antgroup.geaflow.common.config.keys.DSLConfigKeys;
26+
import com.antgroup.geaflow.utils.HttpUtil;
27+
import com.google.gson.Gson;
28+
import java.net.URI;
29+
import java.net.URISyntaxException;
30+
import java.util.HashMap;
31+
import java.util.Map;
32+
import org.slf4j.Logger;
33+
import org.slf4j.LoggerFactory;
34+
35+
public class RestJobOperatorCallback implements JobOperatorCallback {
36+
private static final Logger LOGGER = LoggerFactory.getLogger(RestJobOperatorCallback.class);
37+
private static final String GEAFLOW_TOKEN_KEY = "geaflow-token";
38+
private static final String FINISH_JOB_PATH = "/api/tasks/%s/operations";
39+
private static final String FINISH_ACTION_KEY = "finish";
40+
private static final String JOB_ACTION_KEY = "action";
41+
42+
private final String callbackUrl;
43+
private final Map<String, String> headers;
44+
private final long uniqueId;
45+
46+
public RestJobOperatorCallback(Configuration config, String url) {
47+
this.uniqueId = config.getLong(JOB_UNIQUE_ID);
48+
this.callbackUrl = url;
49+
this.headers = new HashMap<>();
50+
this.headers.put(GEAFLOW_TOKEN_KEY, config.getString(DSLConfigKeys.GEAFLOW_DSL_CATALOG_TOKEN_KEY, ""));
51+
}
52+
53+
@Override
54+
public void onFinish() {
55+
Map<String, String> params = new HashMap<>();
56+
params.put(JOB_ACTION_KEY, FINISH_ACTION_KEY);
57+
JobOperatorCallback.JobOperatorMeta jobOperatorMeta = new JobOperatorCallback.JobOperatorMeta(params);
58+
59+
String fullUrl = getFullUrl(jobOperatorMeta);
60+
if (fullUrl != null) {
61+
HttpRequest request = new HttpRequest();
62+
request.setSuccess(true);
63+
request.setData(jobOperatorMeta);
64+
HttpUtil.post(fullUrl, new Gson().toJson(request), headers);
65+
}
66+
}
67+
68+
private String getFullUrl(JobOperatorMeta jobOperatorMeta) {
69+
String fullUrl = null;
70+
try {
71+
URI uri = new URI(this.callbackUrl);
72+
String path = String.format(FINISH_JOB_PATH, uniqueId);
73+
fullUrl = uri.resolve(path).toString();
74+
} catch (URISyntaxException e) {
75+
LOGGER.error("post {} failed: {}, msg: {}", fullUrl, jobOperatorMeta, e.getMessage());
76+
}
77+
return fullUrl;
78+
}
79+
80+
}

‎geaflow/geaflow-core/geaflow-engine/geaflow-cluster/src/main/java/com/antgroup/geaflow/cluster/client/callback/SimpleJobOperatorCallback.java‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,33 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with 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,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
package com.antgroup.geaflow.cluster.client.callback;
21+
22+
import org.slf4j.Logger;
23+
import org.slf4j.LoggerFactory;
24+
25+
public class SimpleJobOperatorCallback implements JobOperatorCallback {
26+
private static final Logger LOGGER = LoggerFactory.getLogger(SimpleJobOperatorCallback.class);
27+
28+
@Override
29+
public void onFinish() {
30+
LOGGER.info("finish job successfully");
31+
}
32+
33+
}

‎geaflow/geaflow-core/geaflow-engine/geaflow-cluster/src/main/java/com/antgroup/geaflow/cluster/driver/Driver.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,8 @@
1919

2020
package com.antgroup.geaflow.cluster.driver;
2121

22+
import com.antgroup.geaflow.cluster.client.callback.JobOperatorCallback;
23+
import com.antgroup.geaflow.cluster.client.callback.JobOperatorCallbackFactory;
2224
import com.antgroup.geaflow.cluster.common.AbstractContainer;
2325
import com.antgroup.geaflow.cluster.common.ExecutionIdGenerator;
2426
import com.antgroup.geaflow.cluster.constants.ClusterConstants;
@@ -62,6 +64,7 @@ public class Driver extends AbstractContainer implements IDriver<IEvent, Boolean
6264
private DriverContext driverContext;
6365
private ExecutorService executorService;
6466
private Map<PipelineService, IPipelineExecutor> pipelineExecutorMap;
67+
private JobOperatorCallback jobOperatorCallback;
6568

6669
public Driver() {
6770
this(0);
@@ -81,6 +84,7 @@ public void init(DriverContext driverContext) {
8184
1,
8285
ThreadUtil.namedThreadFactory(true, DRIVER_EXECUTOR, ComponentUncaughtExceptionHandler.INSTANCE));
8386
this.pipelineExecutorMap = new HashMap<>();
87+
this.jobOperatorCallback = JobOperatorCallbackFactory.createJobOperatorCallback(configuration);
8488

8589
ExecutionIdGenerator.init(id);
8690
if (driverContext.getPipeline() != null) {
@@ -142,6 +146,8 @@ public Boolean executePipelineInternal(Pipeline pipeline) {
142146
pipelineExecutorMap.put(pipelineService, pipelineExecutor);
143147
pipelineExecutor.startPipelineService(pipelineService);
144148
}
149+
150+
this.jobOperatorCallback.onFinish();
145151
LOGGER.info("finish execute pipeline {}", pipeline);
146152
return true;
147153
} catch (Throwable e) {

‎geaflow/geaflow-core/geaflow-engine/geaflow-cluster/src/test/java/com/antgroup/geaflow/cluster/client/callback/RestClusterStartedCallbackTest.java‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,6 @@
2020
package com.antgroup.geaflow.cluster.client.callback;
2121

2222
import com.antgroup.geaflow.cluster.client.callback.ClusterStartedCallback.ClusterMeta;
23-
import com.antgroup.geaflow.cluster.client.callback.RestClusterStartedCallback.HttpRequest;
2423
import com.antgroup.geaflow.cluster.rpc.ConnectAddress;
2524
import com.antgroup.geaflow.common.config.Configuration;
2625
import com.google.gson.Gson;

0 commit comments

Comments
 (0)