You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@shardingsphere.apache.org by wu...@apache.org on 2022/11/21 10:07:45 UTC
[shardingsphere-elasticjob] branch master updated: Enhance Dump Job (#2048)
This is an automated email from the ASF dual-hosted git repository.
wuweijie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shardingsphere-elasticjob.git
The following commit(s) were added to refs/heads/master by this push:
new ca2995d83 Enhance Dump Job (#2048)
ca2995d83 is described below
commit ca2995d834c1b208e9ebcb7d27189e9711a68ee0
Author: verysu <25...@qq.com>
AuthorDate: Mon Nov 21 18:07:36 2022 +0800
Enhance Dump Job (#2048)
* add JobSystemException Constructor
fixed Script Job throw Exception stack was ignored
* Add unit case for JobSystemException
* Add customized Tracing DataSource
* Add custom Tracing DataSource
* Add custom Tracing DataSource: Refactor
* Add custom Tracing DataSource : add param javadoc
* Enhance Dump Job
* Enhance Dump Job: avoid use the '.*'
* maybe ouccur build fail.
* remove it, maybe ouccur build fail.
* Fixed UnitTest
* Fixed TestUnit
* Fixed UnitTest
* Fixed UnitTest
* update variable order
* revert error log
* Fixed UnitTest
* Fixed UnitTest
* Fixed UnitTest
* remove UT
* remove unuse import
---
.../lite/internal/snapshot/SnapshotService.java | 46 ++++++++++++++++++++--
.../snapshot/SnapshotServiceEnableTest.java | 14 ++++++-
.../lite/internal/snapshot/SocketUtils.java | 13 ++++--
.../lite/lifecycle/api/JobOperateAPI.java | 13 ++++++
.../internal/operate/JobOperateAPIImpl.java | 7 ++++
.../internal/operate/JobOperateAPIImplTest.java | 5 ++-
6 files changed, 87 insertions(+), 11 deletions(-)
diff --git a/elasticjob-lite/elasticjob-lite-core/src/main/java/org/apache/shardingsphere/elasticjob/lite/internal/snapshot/SnapshotService.java b/elasticjob-lite/elasticjob-lite-core/src/main/java/org/apache/shardingsphere/elasticjob/lite/internal/snapshot/SnapshotService.java
index 4aa2c8445..fe109f5b0 100644
--- a/elasticjob-lite/elasticjob-lite-core/src/main/java/org/apache/shardingsphere/elasticjob/lite/internal/snapshot/SnapshotService.java
+++ b/elasticjob-lite/elasticjob-lite-core/src/main/java/org/apache/shardingsphere/elasticjob/lite/internal/snapshot/SnapshotService.java
@@ -70,6 +70,7 @@ public final class SnapshotService {
}
private int openSocket(final int port) throws IOException {
+ closed = false;
serverSocket = new ServerSocket(port);
int localPort = serverSocket.getLocalPort();
String threadName = String.format("elasticjob-snapshot-service-%d", localPort);
@@ -99,10 +100,9 @@ public final class SnapshotService {
Socket ignored = socket) {
String cmdLine = reader.readLine();
if (null != cmdLine && cmdLine.startsWith(DUMP_COMMAND) && cmdLine.split("@").length == 2) {
- List<String> result = new ArrayList<>();
String jobName = cmdLine.split("@")[1];
- dumpDirectly("/" + jobName, jobName, result);
- outputMessage(writer, String.join("\n", SensitiveInfoUtils.filterSensitiveIps(result)) + "\n");
+ String result = dumpJobDirectly(jobName);
+ outputMessage(writer, result);
}
}
}
@@ -130,7 +130,45 @@ public final class SnapshotService {
dumpDirectly(zkPath, jobName, result);
}
}
-
+
+ /**
+ * Dump job.
+ * @param jobName job's name
+ * @return dump job's info
+ */
+ public String dumpJobDirectly(final String jobName) {
+ String path = "/" + jobName;
+ final List<String> result = new ArrayList<>();
+ dumpDirectly(path, jobName, result);
+ return String.join("\n", SensitiveInfoUtils.filterSensitiveIps(result)) + "\n";
+ }
+
+ /**
+ * Dump job.
+ * @param instanceIp job instance ip addr
+ * @param dumpPort dump port
+ * @param jobName job's name
+ * @return dump job's info
+ * @throws IOException i/o exception
+ */
+ public static String dumpJob(final String instanceIp, final int dumpPort, final String jobName) throws IOException {
+ try (
+ Socket socket = new Socket(instanceIp, dumpPort);
+ BufferedReader reader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
+ BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(socket.getOutputStream()))
+ ) {
+ writer.write(DUMP_COMMAND + jobName);
+ writer.newLine();
+ writer.flush();
+ StringBuilder sb = new StringBuilder();
+ String line;
+ while (null != (line = reader.readLine())) {
+ sb.append(line).append("\n");
+ }
+ return sb.toString();
+ }
+ }
+
private void outputMessage(final BufferedWriter outputWriter, final String msg) throws IOException {
outputWriter.append(msg);
outputWriter.flush();
diff --git a/elasticjob-lite/elasticjob-lite-core/src/test/java/org/apache/shardingsphere/elasticjob/lite/internal/snapshot/SnapshotServiceEnableTest.java b/elasticjob-lite/elasticjob-lite-core/src/test/java/org/apache/shardingsphere/elasticjob/lite/internal/snapshot/SnapshotServiceEnableTest.java
index e1c66f6f5..74879562f 100644
--- a/elasticjob-lite/elasticjob-lite-core/src/test/java/org/apache/shardingsphere/elasticjob/lite/internal/snapshot/SnapshotServiceEnableTest.java
+++ b/elasticjob-lite/elasticjob-lite-core/src/test/java/org/apache/shardingsphere/elasticjob/lite/internal/snapshot/SnapshotServiceEnableTest.java
@@ -25,7 +25,7 @@ import org.junit.Test;
import java.io.IOException;
import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertEquals;
public final class SnapshotServiceEnableTest extends BaseSnapshotServiceTest {
@@ -46,6 +46,16 @@ public final class SnapshotServiceEnableTest extends BaseSnapshotServiceTest {
@Test
public void assertMonitorWithCommand() throws IOException {
assertNotNull(SocketUtils.sendCommand(SnapshotService.DUMP_COMMAND + getJobName(), DUMP_PORT));
- assertNull(SocketUtils.sendCommand("unknown_command", DUMP_PORT));
+ assertEquals(SocketUtils.sendCommand("unknown_command", DUMP_PORT), "");
+ }
+
+ @Test
+ public void assertDumpJobDirectly() throws IOException {
+ assertNotNull(getSnapshotService().dumpJobDirectly(getJobName()));
+ }
+
+ @Test
+ public void assertDumpJob() throws IOException {
+ assertNotNull(SnapshotService.dumpJob("127.0.0.1", DUMP_PORT, getJobName()));
}
}
diff --git a/elasticjob-lite/elasticjob-lite-core/src/test/java/org/apache/shardingsphere/elasticjob/lite/internal/snapshot/SocketUtils.java b/elasticjob-lite/elasticjob-lite-core/src/test/java/org/apache/shardingsphere/elasticjob/lite/internal/snapshot/SocketUtils.java
index d71465671..5435a838c 100644
--- a/elasticjob-lite/elasticjob-lite-core/src/test/java/org/apache/shardingsphere/elasticjob/lite/internal/snapshot/SocketUtils.java
+++ b/elasticjob-lite/elasticjob-lite-core/src/test/java/org/apache/shardingsphere/elasticjob/lite/internal/snapshot/SocketUtils.java
@@ -17,6 +17,9 @@
package org.apache.shardingsphere.elasticjob.lite.internal.snapshot;
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.IOException;
@@ -24,9 +27,6 @@ import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.net.Socket;
-import lombok.AccessLevel;
-import lombok.NoArgsConstructor;
-
@NoArgsConstructor(access = AccessLevel.PRIVATE)
public final class SocketUtils {
@@ -47,7 +47,12 @@ public final class SocketUtils {
writer.write(command);
writer.newLine();
writer.flush();
- return reader.readLine();
+ StringBuilder sb = new StringBuilder();
+ String line;
+ while (null != (line = reader.readLine())) {
+ sb.append(line).append("\n");
+ }
+ return sb.toString();
}
}
}
diff --git a/elasticjob-lite/elasticjob-lite-lifecycle/src/main/java/org/apache/shardingsphere/elasticjob/lite/lifecycle/api/JobOperateAPI.java b/elasticjob-lite/elasticjob-lite-lifecycle/src/main/java/org/apache/shardingsphere/elasticjob/lite/lifecycle/api/JobOperateAPI.java
index 3d1e56762..80b3f2e73 100644
--- a/elasticjob-lite/elasticjob-lite-lifecycle/src/main/java/org/apache/shardingsphere/elasticjob/lite/lifecycle/api/JobOperateAPI.java
+++ b/elasticjob-lite/elasticjob-lite-lifecycle/src/main/java/org/apache/shardingsphere/elasticjob/lite/lifecycle/api/JobOperateAPI.java
@@ -17,6 +17,8 @@
package org.apache.shardingsphere.elasticjob.lite.lifecycle.api;
+import java.io.IOException;
+
/**
* Job operate API.
*/
@@ -64,4 +66,15 @@ public interface JobOperateAPI {
* @param serverIp server IP address
*/
void remove(String jobName, String serverIp);
+
+ /**
+ * Dump job.
+ *
+ * @param jobName job name
+ * @param instanceIp instance IP address
+ * @param dumpPort dump port
+ * @return dump job result
+ * @throws IOException i/o exception
+ */
+ String dump(String jobName, String instanceIp, int dumpPort) throws IOException;
}
diff --git a/elasticjob-lite/elasticjob-lite-lifecycle/src/main/java/org/apache/shardingsphere/elasticjob/lite/lifecycle/internal/operate/JobOperateAPIImpl.java b/elasticjob-lite/elasticjob-lite-lifecycle/src/main/java/org/apache/shardingsphere/elasticjob/lite/lifecycle/internal/operate/JobOperateAPIImpl.java
index 50016107b..1e072096f 100644
--- a/elasticjob-lite/elasticjob-lite-lifecycle/src/main/java/org/apache/shardingsphere/elasticjob/lite/lifecycle/internal/operate/JobOperateAPIImpl.java
+++ b/elasticjob-lite/elasticjob-lite-lifecycle/src/main/java/org/apache/shardingsphere/elasticjob/lite/lifecycle/internal/operate/JobOperateAPIImpl.java
@@ -22,10 +22,12 @@ import org.apache.shardingsphere.elasticjob.infra.handler.sharding.JobInstance;
import org.apache.shardingsphere.elasticjob.infra.yaml.YamlEngine;
import org.apache.shardingsphere.elasticjob.lite.internal.instance.InstanceService;
import org.apache.shardingsphere.elasticjob.lite.internal.server.ServerStatus;
+import org.apache.shardingsphere.elasticjob.lite.internal.snapshot.SnapshotService;
import org.apache.shardingsphere.elasticjob.lite.internal.storage.JobNodePath;
import org.apache.shardingsphere.elasticjob.lite.lifecycle.api.JobOperateAPI;
import org.apache.shardingsphere.elasticjob.reg.base.CoordinatorRegistryCenter;
+import java.io.IOException;
import java.util.List;
/**
@@ -137,4 +139,9 @@ public final class JobOperateAPIImpl implements JobOperateAPI {
}
}
}
+
+ @Override
+ public String dump(final String jobName, final String instanceIp, final int dumpPort) throws IOException {
+ return SnapshotService.dumpJob(instanceIp, dumpPort, jobName);
+ }
}
diff --git a/elasticjob-lite/elasticjob-lite-lifecycle/src/test/java/org/apache/shardingsphere/elasticjob/lite/lifecycle/internal/operate/JobOperateAPIImplTest.java b/elasticjob-lite/elasticjob-lite-lifecycle/src/test/java/org/apache/shardingsphere/elasticjob/lite/lifecycle/internal/operate/JobOperateAPIImplTest.java
index 7e85aafaf..f188757d4 100644
--- a/elasticjob-lite/elasticjob-lite-lifecycle/src/test/java/org/apache/shardingsphere/elasticjob/lite/lifecycle/internal/operate/JobOperateAPIImplTest.java
+++ b/elasticjob-lite/elasticjob-lite-lifecycle/src/test/java/org/apache/shardingsphere/elasticjob/lite/lifecycle/internal/operate/JobOperateAPIImplTest.java
@@ -35,7 +35,9 @@ import static org.mockito.Mockito.when;
@RunWith(MockitoJUnitRunner.class)
public final class JobOperateAPIImplTest {
-
+
+ static final int DUMP_PORT = 9000;
+
private JobOperateAPI jobOperateAPI;
@Mock
@@ -166,4 +168,5 @@ public final class JobOperateAPIImplTest {
assertFalse(regCenter.isExisted("/test_job1/servers/ip1"));
assertFalse(regCenter.isExisted("/test_job2/servers/ip1"));
}
+
}