You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@wayang.apache.org by rp...@apache.org on 2021/05/17 20:14:22 UTC
[incubator-wayang] branch debugger-sidecar updated (9b93f92 ->
89ce68e)
This is an automated email from the ASF dual-hosted git repository.
rpardomeza pushed a change to branch debugger-sidecar
in repository https://gitbox.apache.org/repos/asf/incubator-wayang.git.
from 9b93f92 [WAYANG-30] Web Service local cluster demo execution
new 7a9a09c [WAYANG-30] Added the required yaml to run Flink Cluster on Kubernetes
new 7934417 [WAYANG-30] Executor Manager to follow created threads
new a0e7988 [WAYANG-30] Function to start Flink Cluster
new 89ce68e [WAYANG-30] Function to check Thread status
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../hackit/sidecar/webservice/ExecutorManager.java | 103 +++++++++++++++++++++
.../hackit/sidecar/webservice/WebService.java | 51 +++++++++-
.../src/main/resources/jobmanager-service.yaml | 18 ++++
.../src/main/resources/jobmanager.yaml | 33 +++++++
.../src/main/resources/taskmanager.yaml | 31 +++++++
5 files changed, 232 insertions(+), 4 deletions(-)
create mode 100644 wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/java/org/apache/wayang/hackit/sidecar/webservice/ExecutorManager.java
create mode 100644 wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/resources/jobmanager-service.yaml
create mode 100644 wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/resources/jobmanager.yaml
create mode 100644 wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/resources/taskmanager.yaml
[incubator-wayang] 03/04: [WAYANG-30] Function to start Flink
Cluster
Posted by rp...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rpardomeza pushed a commit to branch debugger-sidecar
in repository https://gitbox.apache.org/repos/asf/incubator-wayang.git
commit a0e7988555113acaaf3d6d1a1ccc2d0c148169f4
Author: rodrigopardomeza <ro...@gmail.com>
AuthorDate: Mon May 17 16:12:45 2021 -0400
[WAYANG-30] Function to start Flink Cluster
---
.../hackit/sidecar/webservice/WebService.java | 32 +++++++++++++++++++---
1 file changed, 28 insertions(+), 4 deletions(-)
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/java/org/apache/wayang/hackit/sidecar/webservice/WebService.java b/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/java/org/apache/wayang/hackit/sidecar/webservice/WebService.java
index 850edee..b463b59 100644
--- a/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/java/org/apache/wayang/hackit/sidecar/webservice/WebService.java
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/java/org/apache/wayang/hackit/sidecar/webservice/WebService.java
@@ -4,10 +4,17 @@ import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.operators.ProjectOperator;
+import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.api.java.tuple.Tuple4;
import org.springframework.http.HttpStatus;
-import org.springframework.web.bind.annotation.GetMapping;
-import org.springframework.web.bind.annotation.RequestMapping;
-import org.springframework.web.bind.annotation.RestController;
+import org.springframework.web.bind.annotation.*;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
@RestController
@RequestMapping("/sidecar")
@@ -32,7 +39,24 @@ public class WebService {
}
@GetMapping("/debug/start")
- public HttpStatus serviceStart(){
+ public UUID serviceStart(){
+
+ ProcessBuilder builder_proc1 = new ProcessBuilder("kubectl", "create", "-f", "jobmanager.yaml");
+ ProcessBuilder builder_proc2 = new ProcessBuilder("kubectl", "create", "-f", "jobmanager-service.yaml");
+ ProcessBuilder builder_proc3 = new ProcessBuilder("kubectl", "create", "-f", "taskmanger.yaml");
+
+ List<ProcessBuilder> processes = new ArrayList<>();
+ processes.add(0, builder_proc1);
+ processes.add(1, builder_proc2);
+ processes.add(2, builder_proc3);
+
+ UUID ProcessID = ExecutorManager.addThread(processes);
+
+ return ProcessID;
+ }
+
+ @GetMapping("/debug/demo")
+ public HttpStatus serviceDemo(){
try {
// LocalEnvironment
[incubator-wayang] 01/04: [WAYANG-30] Added the required yaml to
run Flink Cluster on Kubernetes
Posted by rp...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rpardomeza pushed a commit to branch debugger-sidecar
in repository https://gitbox.apache.org/repos/asf/incubator-wayang.git
commit 7a9a09c778e179163017f8d9b832964586580323
Author: rodrigopardomeza <ro...@gmail.com>
AuthorDate: Mon May 17 16:07:51 2021 -0400
[WAYANG-30] Added the required yaml to run Flink Cluster on Kubernetes
---
.../src/main/resources/jobmanager-service.yaml | 18 ++++++++++++
.../src/main/resources/jobmanager.yaml | 33 ++++++++++++++++++++++
.../src/main/resources/taskmanager.yaml | 31 ++++++++++++++++++++
3 files changed, 82 insertions(+)
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/resources/jobmanager-service.yaml b/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/resources/jobmanager-service.yaml
new file mode 100644
index 0000000..a3c235a
--- /dev/null
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/resources/jobmanager-service.yaml
@@ -0,0 +1,18 @@
+apiVersion: v1
+kind: Service
+metadata:
+ name: jobmanager
+spec:
+ type: NodePort
+ ports:
+ - name: rpc
+ port: 6123
+ - name: blob
+ port: 6124
+ - name: query
+ port: 6125
+ - name: ui
+ port: 8081
+ selector:
+ app: flink
+ component: master
\ No newline at end of file
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/resources/jobmanager.yaml b/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/resources/jobmanager.yaml
new file mode 100644
index 0000000..6919861
--- /dev/null
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/resources/jobmanager.yaml
@@ -0,0 +1,33 @@
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+ name: jobmanager
+spec:
+ selector:
+ matchLabels:
+ app: flink
+ replicas: 1
+ template:
+ metadata:
+ labels:
+ app: flink
+ component: master
+ spec:
+ containers:
+ - name: master
+ image: flink:1.13.0-scala_2.11-java8
+ workingDir: /opt/flink
+ args:
+ - jobmanager
+ ports:
+ - containerPort: 6123
+ name: rpc
+ - containerPort: 6124
+ name: blob
+ - containerPort: 6125
+ name: query
+ - containerPort: 6126
+ name: ui
+ env:
+ - name: JOB_MANAGER_RPC_ADDRESS
+ value: jobmanager
\ No newline at end of file
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/resources/taskmanager.yaml b/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/resources/taskmanager.yaml
new file mode 100644
index 0000000..03e1ae8
--- /dev/null
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/resources/taskmanager.yaml
@@ -0,0 +1,31 @@
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+ name: taskmanager
+spec:
+ selector:
+ matchLabels:
+ app: flink
+ replicas: 2
+ template:
+ metadata:
+ labels:
+ app: flink
+ component: task
+ spec:
+ containers:
+ - name: task
+ image: flink:1.13.0-scala_2.11-java8
+ workingDir: /opt/flink
+ args:
+ - taskmanager
+ ports:
+ - containerPort: 6121
+ name: data
+ - containerPort: 6122
+ name: rpc
+ - containerPort: 6125
+ name: query
+ env:
+ - name: JOB_MANAGER_RPC_ADDRESS
+ value: jobmanager
\ No newline at end of file
[incubator-wayang] 02/04: [WAYANG-30] Executor Manager to follow
created threads
Posted by rp...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rpardomeza pushed a commit to branch debugger-sidecar
in repository https://gitbox.apache.org/repos/asf/incubator-wayang.git
commit 793441744cab037ef01cbe0bcf287007b4319882
Author: rodrigopardomeza <ro...@gmail.com>
AuthorDate: Mon May 17 16:11:55 2021 -0400
[WAYANG-30] Executor Manager to follow created threads
---
.../hackit/sidecar/webservice/ExecutorManager.java | 103 +++++++++++++++++++++
1 file changed, 103 insertions(+)
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/java/org/apache/wayang/hackit/sidecar/webservice/ExecutorManager.java b/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/java/org/apache/wayang/hackit/sidecar/webservice/ExecutorManager.java
new file mode 100644
index 0000000..bef9676
--- /dev/null
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/java/org/apache/wayang/hackit/sidecar/webservice/ExecutorManager.java
@@ -0,0 +1,103 @@
+package org.apache.wayang.hackit.sidecar.webservice;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.util.*;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+public class ExecutorManager {
+
+ private static ExecutorService ex;
+ public static Map<UUID, Thread> threadsMap;
+ public static Map<UUID, List<String>> processesLogs;
+
+ public static void init(){
+ ex = Executors.newFixedThreadPool(1);
+ threadsMap = new HashMap<>();
+ processesLogs = new HashMap<>();
+ }
+
+ public ExecutorManager(){
+
+ }
+
+ public static UUID addThread(ProcessBuilder processNotExecuted) {
+
+ UUID key = UUID.randomUUID();
+ List<String> logs = new ArrayList<>();
+
+ processesLogs.put(key, logs);
+
+ ex.submit(() -> {
+ try {
+ Process p = processNotExecuted.start();
+ BufferedReader stdInput = new BufferedReader(new
+ InputStreamReader(p.getInputStream()));
+ stdInput.lines()
+ .forEach(
+ line -> {
+ logs.add(line);
+ }
+ );
+
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+ });
+
+ return key;
+ }
+
+ public static UUID addThread(List<ProcessBuilder> processesNotExecuted) {
+
+ UUID key = UUID.randomUUID();
+ List<String> logs = new ArrayList<>();
+
+ processesLogs.put(key, logs);
+
+ ex.submit(() -> {
+ try {
+
+ BufferedReader stdInput;
+ for (ProcessBuilder pb : processesNotExecuted) {
+ Process p = pb.start();
+ stdInput = new BufferedReader(new
+ InputStreamReader(p.getInputStream()));
+
+ stdInput.lines()
+ .forEach(
+ line -> {
+ logs.add(line);
+ }
+ );
+ }
+
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+ });
+
+ return key;
+ }
+
+ public static List<String> getProcessLogs(UUID pID){
+
+ return ExecutorManager.processesLogs.get(pID);
+ }
+
+ public static List<String> getProcessLogs(UUID pID, Integer startFrom){
+
+ List<String> logs = ExecutorManager.processesLogs.get(pID);
+ List<String> ls = new ArrayList<>();
+
+ Object[] values = logs.toArray();
+ for (int i = startFrom - 1; i < values.length; i++) {
+ ls.add(values[i].toString());
+ }
+
+ return ls;
+ }
+
+}
[incubator-wayang] 04/04: [WAYANG-30] Function to check Thread
status
Posted by rp...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rpardomeza pushed a commit to branch debugger-sidecar
in repository https://gitbox.apache.org/repos/asf/incubator-wayang.git
commit 89ce68e3889eeb7b74499f8be7b0f2cf9d383621
Author: rodrigopardomeza <ro...@gmail.com>
AuthorDate: Mon May 17 16:13:59 2021 -0400
[WAYANG-30] Function to check Thread status
---
.../wayang/hackit/sidecar/webservice/WebService.java | 19 +++++++++++++++++++
1 file changed, 19 insertions(+)
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/java/org/apache/wayang/hackit/sidecar/webservice/WebService.java b/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/java/org/apache/wayang/hackit/sidecar/webservice/WebService.java
index b463b59..38a5003 100644
--- a/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/java/org/apache/wayang/hackit/sidecar/webservice/WebService.java
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/java/org/apache/wayang/hackit/sidecar/webservice/WebService.java
@@ -85,4 +85,23 @@ public class WebService {
return HttpStatus.ACCEPTED;
}
+
+ @PostMapping(path = "/debug/check")
+ public String checkStatus(
+ @RequestParam("key") String key,
+ @RequestParam(value = "startfrom", required = false) Integer startFrom
+ ){
+ List<String> ls;
+
+ if(startFrom == null)
+ ls = ExecutorManager.getProcessLogs(UUID.fromString(key));
+ else
+ ls = ExecutorManager.getProcessLogs(UUID.fromString(key), startFrom);
+
+ String res = "";
+ for (String s: ls)
+ res += s + "\n";
+
+ return res;
+ }
}