dailidong
5 years ago
181 changed files with 5438 additions and 2021 deletions
@ -1,137 +0,0 @@
|
||||
/* |
||||
* Licensed to the Apache Software Foundation (ASF) under one or more |
||||
* contributor license agreements. See the NOTICE file distributed with |
||||
* this work for additional information regarding copyright ownership. |
||||
* The ASF licenses this file to You under the Apache License, Version 2.0 |
||||
* (the "License"); you may not use this file except in compliance with |
||||
* the License. You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
*/ |
||||
package org.apache.dolphinscheduler.api.log; |
||||
|
||||
import io.grpc.ManagedChannel; |
||||
import io.grpc.ManagedChannelBuilder; |
||||
import io.grpc.StatusRuntimeException; |
||||
import org.apache.dolphinscheduler.rpc.*; |
||||
import org.slf4j.Logger; |
||||
import org.slf4j.LoggerFactory; |
||||
|
||||
import java.util.concurrent.TimeUnit; |
||||
|
||||
/** |
||||
* log client |
||||
*/ |
||||
public class LogClient { |
||||
|
||||
private static final Logger logger = LoggerFactory.getLogger(LogClient.class); |
||||
|
||||
private final ManagedChannel channel; |
||||
private final LogViewServiceGrpc.LogViewServiceBlockingStub blockingStub; |
||||
|
||||
/** |
||||
* construct client connecting to HelloWorld server at {@code host:port} |
||||
* |
||||
* @param host host |
||||
* @param port port |
||||
*/ |
||||
public LogClient(String host, int port) { |
||||
this(ManagedChannelBuilder.forAddress(host, port) |
||||
// Channels are secure by default (via SSL/TLS). For the example we disable TLS to avoid
|
||||
// needing certificates.
|
||||
.usePlaintext(true)); |
||||
} |
||||
|
||||
/** |
||||
* construct client for accessing RouteGuide server using the existing channel |
||||
* |
||||
*/ |
||||
LogClient(ManagedChannelBuilder<?> channelBuilder) { |
||||
/** |
||||
* set max read size |
||||
*/ |
||||
channelBuilder.maxInboundMessageSize(Integer.MAX_VALUE); |
||||
channel = channelBuilder.build(); |
||||
blockingStub = LogViewServiceGrpc.newBlockingStub(channel); |
||||
} |
||||
|
||||
/** |
||||
* shutdown |
||||
* |
||||
* @throws InterruptedException InterruptedException |
||||
*/ |
||||
public void shutdown() throws InterruptedException { |
||||
channel.shutdown().awaitTermination(5, TimeUnit.SECONDS); |
||||
} |
||||
|
||||
/** |
||||
* roll view log |
||||
* |
||||
* @param path path |
||||
* @param skipLineNum skip line number |
||||
* @param limit limit |
||||
* @return log content |
||||
*/ |
||||
public String rollViewLog(String path,int skipLineNum,int limit) { |
||||
logger.info("roll view log : path {},skipLineNum {} ,limit {}", path, skipLineNum, limit); |
||||
LogParameter pathParameter = LogParameter |
||||
.newBuilder() |
||||
.setPath(path) |
||||
.setSkipLineNum(skipLineNum) |
||||
.setLimit(limit) |
||||
.build(); |
||||
RetStrInfo retStrInfo; |
||||
try { |
||||
retStrInfo = blockingStub.rollViewLog(pathParameter); |
||||
return retStrInfo.getMsg(); |
||||
} catch (StatusRuntimeException e) { |
||||
logger.error("roll view log error", e); |
||||
return null; |
||||
} |
||||
} |
||||
|
||||
/** |
||||
* view log |
||||
* |
||||
* @param path path |
||||
* @return log content |
||||
*/ |
||||
public String viewLog(String path) { |
||||
logger.info("view log path {}",path); |
||||
PathParameter pathParameter = PathParameter.newBuilder().setPath(path).build(); |
||||
RetStrInfo retStrInfo; |
||||
try { |
||||
retStrInfo = blockingStub.viewLog(pathParameter); |
||||
return retStrInfo.getMsg(); |
||||
} catch (StatusRuntimeException e) { |
||||
logger.error("view log error", e); |
||||
return null; |
||||
} |
||||
} |
||||
|
||||
/** |
||||
* get log size |
||||
* |
||||
* @param path log path |
||||
* @return log content bytes |
||||
*/ |
||||
public byte[] getLogBytes(String path) { |
||||
logger.info("log path {}",path); |
||||
PathParameter pathParameter = PathParameter.newBuilder().setPath(path).build(); |
||||
RetByteInfo retByteInfo; |
||||
try { |
||||
retByteInfo = blockingStub.getLogBytes(pathParameter); |
||||
return retByteInfo.getData().toByteArray(); |
||||
} catch (StatusRuntimeException e) { |
||||
logger.error("log size error", e); |
||||
return null; |
||||
} |
||||
} |
||||
|
||||
} |
||||
@ -0,0 +1,218 @@
|
||||
/* |
||||
* Licensed to the Apache Software Foundation (ASF) under one or more |
||||
* contributor license agreements. See the NOTICE file distributed with |
||||
* this work for additional information regarding copyright ownership. |
||||
* The ASF licenses this file to You under the Apache License, Version 2.0 |
||||
* (the "License"); you may not use this file except in compliance with |
||||
* the License. You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
*/ |
||||
|
||||
package org.apache.dolphinscheduler.api.utils; |
||||
|
||||
import org.junit.Assert; |
||||
import org.junit.Before; |
||||
import org.junit.Ignore; |
||||
import org.junit.Test; |
||||
import org.junit.runner.RunWith; |
||||
import org.mockito.InjectMocks; |
||||
import org.mockito.Mock; |
||||
import org.mockito.Mockito; |
||||
import org.powermock.api.mockito.PowerMockito; |
||||
import org.powermock.core.classloader.annotations.PrepareForTest; |
||||
import org.powermock.modules.junit4.PowerMockRunner; |
||||
import org.slf4j.Logger; |
||||
import org.slf4j.LoggerFactory; |
||||
|
||||
import java.io.ByteArrayInputStream; |
||||
import java.io.ByteArrayOutputStream; |
||||
import java.io.InputStream; |
||||
import java.net.InetSocketAddress; |
||||
import java.net.Socket; |
||||
import java.net.SocketTimeoutException; |
||||
|
||||
import static org.junit.Assert.assertEquals; |
||||
import static org.mockito.ArgumentMatchers.any; |
||||
import static org.mockito.Mockito.doThrow; |
||||
import static org.mockito.Mockito.when; |
||||
|
||||
@RunWith(PowerMockRunner.class) |
||||
@PrepareForTest({FourLetterWordMain.class, Socket.class}) |
||||
public class FourLetterWordMainTest { |
||||
|
||||
private static final Logger logger = |
||||
LoggerFactory.getLogger(FourLetterWordMainTest.class); |
||||
private static final String NEW_LINE = "\n"; |
||||
|
||||
@InjectMocks |
||||
private FourLetterWordMain fourLetterWord; |
||||
@Mock |
||||
private Socket socket; |
||||
@Mock |
||||
private InetSocketAddress socketAddress; |
||||
|
||||
private final String localHost = "127.0.0.1"; |
||||
private final int zkPort = 2181; |
||||
private ByteArrayOutputStream byteArrayOutputStream; |
||||
private InputStream inputStream; |
||||
|
||||
private String cmd; |
||||
private String testResult; |
||||
private String expectedStr; |
||||
|
||||
@Before |
||||
public void setUp() { |
||||
// mock socket class
|
||||
PowerMockito.mockStatic(Socket.class); |
||||
try { |
||||
PowerMockito.whenNew(Socket.class).withNoArguments() |
||||
.thenReturn(socket); |
||||
} catch (Exception e) { |
||||
e.printStackTrace(); |
||||
} |
||||
} |
||||
|
||||
/** |
||||
* None mock test method, just to check zookeeper status. |
||||
* Comment @Before notation to run this test. |
||||
* Zookeeper status will be as: |
||||
* Zookeeper version: 3.4.11 ... |
||||
* Received: 6739707 |
||||
* Sent: 6739773 |
||||
* Connections: 20 |
||||
* Outstanding: 0 |
||||
* Zxid: 0x9ba |
||||
* Mode: standalone |
||||
* Node count: 263 |
||||
*/ |
||||
public void testCmd() { |
||||
// "192.168.64.11"
|
||||
// final String zkHost = localHost;
|
||||
final String zkHost = "192.168.64.11"; |
||||
cmd = "srvr"; |
||||
try { |
||||
// Change localhost to right zk host ip.
|
||||
final String result = FourLetterWordMain |
||||
.send4LetterWord(zkHost, zkPort, cmd); |
||||
logger.info(cmd + ": " + result + "<<<"); |
||||
} catch (Exception e) { |
||||
e.printStackTrace(); |
||||
} |
||||
} |
||||
|
||||
@Test |
||||
public void testEmptyCmd() { |
||||
cmd = ""; |
||||
expectedStr = ""; |
||||
testSend4LetterWord(cmd, expectedStr); |
||||
} |
||||
|
||||
@Test |
||||
public void testNullCmd() { |
||||
cmd = null; |
||||
|
||||
try { |
||||
testResult = FourLetterWordMain |
||||
.send4LetterWord(localHost, zkPort, cmd); |
||||
} catch (Exception e) { |
||||
testResult = e.getMessage(); |
||||
} |
||||
|
||||
logger.info("testNullCmd result: " + testResult); |
||||
assertEquals("cmd must not be null", testResult); |
||||
} |
||||
|
||||
@Test |
||||
public void testNullSocketOutput() { |
||||
cmd = "test null socket output"; |
||||
expectedStr = null; |
||||
testSend4LetterWord(cmd, expectedStr); |
||||
} |
||||
|
||||
@Test |
||||
public void testOneLineOutput() { |
||||
cmd = "line 1"; |
||||
|
||||
// line end without \n
|
||||
expectedStr = "line 1" + NEW_LINE; |
||||
testSend4LetterWord(cmd, expectedStr); |
||||
|
||||
// line end with \n
|
||||
expectedStr = "line 1\n" + NEW_LINE; |
||||
testSend4LetterWord(cmd, expectedStr); |
||||
} |
||||
|
||||
@Test |
||||
public void testMultiline() { |
||||
cmd = "line 1 " + NEW_LINE + |
||||
"line 2 " + NEW_LINE + |
||||
"line 3 " + NEW_LINE; |
||||
|
||||
expectedStr = cmd + NEW_LINE; |
||||
testSend4LetterWord(cmd, expectedStr); |
||||
|
||||
expectedStr = NEW_LINE + NEW_LINE + NEW_LINE; |
||||
testSend4LetterWord(cmd, expectedStr); |
||||
} |
||||
|
||||
@Test |
||||
public void testSocketTimeOut() { |
||||
cmd = "test socket time out"; |
||||
|
||||
try { |
||||
doThrow(new SocketTimeoutException()) |
||||
.when(socket) |
||||
.connect(any(InetSocketAddress.class), Mockito.anyInt()); |
||||
testResult = FourLetterWordMain |
||||
.send4LetterWord(localHost, zkPort, cmd); |
||||
} catch (Exception e) { |
||||
testResult = e.getMessage(); |
||||
} |
||||
|
||||
logger.info("testSocketTimeOut result: " + testResult); |
||||
assertEquals( |
||||
"Exception while executing four letter word: " + cmd, |
||||
testResult |
||||
); |
||||
} |
||||
|
||||
/** |
||||
* Test FourLetterWordMain.send4LetterWord() with input cmd and output |
||||
* string. |
||||
* @param cmd |
||||
* @param expectedStr |
||||
*/ |
||||
public void testSend4LetterWord(String cmd, String expectedStr) { |
||||
try { |
||||
final byte[] strBytes = cmd.getBytes(); |
||||
byteArrayOutputStream = new ByteArrayOutputStream(strBytes.length); |
||||
byteArrayOutputStream.write(strBytes, 0, strBytes.length); |
||||
|
||||
inputStream = new ByteArrayInputStream(expectedStr.getBytes()); |
||||
|
||||
when(socket.getOutputStream()) |
||||
.thenReturn(byteArrayOutputStream); |
||||
when(socket.getInputStream()).thenReturn(inputStream); |
||||
|
||||
final String result = FourLetterWordMain |
||||
.send4LetterWord(localHost, zkPort, cmd); |
||||
logger.info( |
||||
"testSend4LetterWord: " + |
||||
"cmd: " + cmd + |
||||
", expectedStr: " + expectedStr + |
||||
", result: " + result + "." |
||||
); |
||||
Assert.assertEquals(expectedStr, result); |
||||
} catch (Exception e) { |
||||
e.printStackTrace(); |
||||
} |
||||
} |
||||
|
||||
} |
||||
@ -0,0 +1,43 @@
|
||||
/* |
||||
* Licensed to the Apache Software Foundation (ASF) under one or more |
||||
* contributor license agreements. See the NOTICE file distributed with |
||||
* this work for additional information regarding copyright ownership. |
||||
* The ASF licenses this file to You under the Apache License, Version 2.0 |
||||
* (the "License"); you may not use this file except in compliance with |
||||
* the License. You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
*/ |
||||
package org.apache.dolphinscheduler.common.utils; |
||||
|
||||
import org.junit.Assert; |
||||
import org.junit.Test; |
||||
import java.util.Date; |
||||
import org.apache.dolphinscheduler.common.model.Server; |
||||
|
||||
public class ResInfoTest { |
||||
@Test |
||||
public void testGetHeartBeatInfo() { |
||||
String info = ResInfo.getHeartBeatInfo(new Date()); |
||||
Assert.assertEquals(7, info.split(",").length); |
||||
} |
||||
|
||||
@Test |
||||
public void testParseHeartbeatForZKInfo() { |
||||
//normal info
|
||||
String info = ResInfo.getHeartBeatInfo(new Date()); |
||||
Server s = ResInfo.parseHeartbeatForZKInfo(info); |
||||
Assert.assertNotNull(s); |
||||
Assert.assertNotNull(s.getResInfo()); |
||||
|
||||
//null param
|
||||
s = ResInfo.parseHeartbeatForZKInfo(null); |
||||
Assert.assertNull(s); |
||||
} |
||||
} |
||||
@ -0,0 +1,70 @@
|
||||
/* |
||||
* Licensed to the Apache Software Foundation (ASF) under one or more |
||||
* contributor license agreements. See the NOTICE file distributed with |
||||
* this work for additional information regarding copyright ownership. |
||||
* The ASF licenses this file to You under the Apache License, Version 2.0 |
||||
* (the "License"); you may not use this file except in compliance with |
||||
* the License. You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
*/ |
||||
package org.apache.dolphinscheduler.common.utils; |
||||
import org.junit.Assert; |
||||
import org.junit.Test; |
||||
import org.mockito.Mockito; |
||||
import java.io.StringReader; |
||||
import java.sql.*; |
||||
|
||||
public class ScriptRunnerTest { |
||||
@Test |
||||
public void testRunScript() { |
||||
//connection is null
|
||||
Exception exception = null; |
||||
ScriptRunner s = new ScriptRunner(null, true, true); |
||||
try { |
||||
s.runScript(new StringReader("select 1")); |
||||
} catch (Exception e) { |
||||
exception = e; |
||||
} |
||||
Assert.assertNotNull(exception); |
||||
|
||||
//connect is not null
|
||||
runScript(""); |
||||
} |
||||
|
||||
private void runScript(String dbName) { |
||||
try { |
||||
Connection conn = Mockito.mock(Connection.class); |
||||
Mockito.when(conn.getAutoCommit()).thenReturn(true); |
||||
PreparedStatement st = Mockito.mock(PreparedStatement.class); |
||||
Mockito.when(conn.createStatement()).thenReturn(st); |
||||
ResultSet rs = Mockito.mock(ResultSet.class); |
||||
Mockito.when(st.getResultSet()).thenReturn(rs); |
||||
ResultSetMetaData md = Mockito.mock(ResultSetMetaData.class); |
||||
Mockito.when(rs.getMetaData()).thenReturn(md); |
||||
Mockito.when(md.getColumnCount()).thenReturn(2); |
||||
Mockito.when(rs.next()).thenReturn(true, false); |
||||
ScriptRunner s = new ScriptRunner(conn, true, true); |
||||
if (dbName.isEmpty()) { |
||||
s.runScript(new StringReader("select 1;")); |
||||
} else { |
||||
s.runScript(new StringReader("select 1;"), dbName); |
||||
} |
||||
Mockito.verify(md).getColumnLabel(1); |
||||
} catch(Exception e) { |
||||
Assert.assertNotNull(e); |
||||
} |
||||
} |
||||
|
||||
@Test |
||||
public void testRunScriptWithDbName() { |
||||
//connect is not null
|
||||
runScript("db_test"); |
||||
} |
||||
} |
||||
@ -0,0 +1,66 @@
|
||||
/* |
||||
* Licensed to the Apache Software Foundation (ASF) under one or more |
||||
* contributor license agreements. See the NOTICE file distributed with |
||||
* this work for additional information regarding copyright ownership. |
||||
* The ASF licenses this file to You under the Apache License, Version 2.0 |
||||
* (the "License"); you may not use this file except in compliance with |
||||
* the License. You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
*/ |
||||
package org.apache.dolphinscheduler.common.utils; |
||||
|
||||
import org.junit.Assert; |
||||
import org.junit.Test; |
||||
|
||||
import java.util.ArrayList; |
||||
|
||||
public class StringUtilsTest { |
||||
@Test |
||||
public void testIsNotEmpty() { |
||||
//null string
|
||||
boolean b = StringUtils.isNotEmpty(null); |
||||
Assert.assertFalse(b); |
||||
|
||||
//"" string
|
||||
b = StringUtils.isNotEmpty(""); |
||||
Assert.assertFalse(b); |
||||
|
||||
//" " string
|
||||
b = StringUtils.isNotEmpty(" "); |
||||
Assert.assertTrue(b); |
||||
|
||||
//"test" string
|
||||
b = StringUtils.isNotEmpty("test"); |
||||
Assert.assertTrue(b); |
||||
} |
||||
|
||||
@Test |
||||
public void testIsNotBlank() { |
||||
//null string
|
||||
boolean b = StringUtils.isNotBlank(null); |
||||
Assert.assertFalse(b); |
||||
|
||||
//"" string
|
||||
b = StringUtils.isNotBlank(""); |
||||
Assert.assertFalse(b); |
||||
|
||||
//" " string
|
||||
b = StringUtils.isNotBlank(" "); |
||||
Assert.assertFalse(b); |
||||
|
||||
//" test " string
|
||||
b = StringUtils.isNotBlank(" test "); |
||||
Assert.assertTrue(b); |
||||
|
||||
//"test" string
|
||||
b = StringUtils.isNotBlank("test"); |
||||
Assert.assertTrue(b); |
||||
} |
||||
} |
||||
@ -0,0 +1,44 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?> |
||||
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" |
||||
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> |
||||
<parent> |
||||
<artifactId>dolphinscheduler</artifactId> |
||||
<groupId>org.apache.dolphinscheduler</groupId> |
||||
<version>1.2.1-SNAPSHOT</version> |
||||
</parent> |
||||
<modelVersion>4.0.0</modelVersion> |
||||
|
||||
<artifactId>dolphinscheduler-remote</artifactId> |
||||
|
||||
<name>dolphinscheduler-remote</name> |
||||
<!-- FIXME change it to the project's website --> |
||||
<url>http://www.example.com</url> |
||||
|
||||
<properties> |
||||
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> |
||||
<maven.compiler.source>1.7</maven.compiler.source> |
||||
<maven.compiler.target>1.7</maven.compiler.target> |
||||
</properties> |
||||
|
||||
<dependencies> |
||||
<dependency> |
||||
<groupId>io.netty</groupId> |
||||
<artifactId>netty-all</artifactId> |
||||
</dependency> |
||||
<dependency> |
||||
<groupId>org.slf4j</groupId> |
||||
<artifactId>slf4j-api</artifactId> |
||||
</dependency> |
||||
<dependency> |
||||
<groupId>com.alibaba</groupId> |
||||
<artifactId>fastjson</artifactId> |
||||
</dependency> |
||||
<dependency> |
||||
<groupId>junit</groupId> |
||||
<artifactId>junit</artifactId> |
||||
<scope>test</scope> |
||||
</dependency> |
||||
|
||||
</dependencies> |
||||
</project> |
||||
@ -0,0 +1,272 @@
|
||||
/* |
||||
* Licensed to the Apache Software Foundation (ASF) under one or more |
||||
* contributor license agreements. See the NOTICE file distributed with |
||||
* this work for additional information regarding copyright ownership. |
||||
* The ASF licenses this file to You under the Apache License, Version 2.0 |
||||
* (the "License"); you may not use this file except in compliance with |
||||
* the License. You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
*/ |
||||
|
||||
package org.apache.dolphinscheduler.remote; |
||||
|
||||
import io.netty.bootstrap.Bootstrap; |
||||
import io.netty.channel.*; |
||||
import io.netty.channel.nio.NioEventLoopGroup; |
||||
import io.netty.channel.socket.SocketChannel; |
||||
import io.netty.channel.socket.nio.NioSocketChannel; |
||||
import org.apache.dolphinscheduler.remote.codec.NettyDecoder; |
||||
import org.apache.dolphinscheduler.remote.codec.NettyEncoder; |
||||
import org.apache.dolphinscheduler.remote.command.Command; |
||||
import org.apache.dolphinscheduler.remote.command.CommandType; |
||||
import org.apache.dolphinscheduler.remote.config.NettyClientConfig; |
||||
import org.apache.dolphinscheduler.remote.exceptions.RemotingException; |
||||
import org.apache.dolphinscheduler.remote.handler.NettyClientHandler; |
||||
import org.apache.dolphinscheduler.remote.processor.NettyRequestProcessor; |
||||
import org.apache.dolphinscheduler.remote.utils.Address; |
||||
import org.apache.dolphinscheduler.remote.utils.Constants; |
||||
import org.slf4j.Logger; |
||||
import org.slf4j.LoggerFactory; |
||||
|
||||
import java.net.InetSocketAddress; |
||||
import java.util.concurrent.ConcurrentHashMap; |
||||
import java.util.concurrent.ExecutorService; |
||||
import java.util.concurrent.Executors; |
||||
import java.util.concurrent.ThreadFactory; |
||||
import java.util.concurrent.atomic.AtomicBoolean; |
||||
import java.util.concurrent.atomic.AtomicInteger; |
||||
|
||||
/** |
||||
* remoting netty client |
||||
*/ |
||||
public class NettyRemotingClient { |
||||
|
||||
private final Logger logger = LoggerFactory.getLogger(NettyRemotingClient.class); |
||||
|
||||
/** |
||||
* bootstrap |
||||
*/ |
||||
private final Bootstrap bootstrap = new Bootstrap(); |
||||
|
||||
/** |
||||
* encoder |
||||
*/ |
||||
private final NettyEncoder encoder = new NettyEncoder(); |
||||
|
||||
/** |
||||
* channels |
||||
*/ |
||||
private final ConcurrentHashMap<Address, Channel> channels = new ConcurrentHashMap(); |
||||
|
||||
/** |
||||
* default executor |
||||
*/ |
||||
private final ExecutorService defaultExecutor = Executors.newFixedThreadPool(Constants.CPUS); |
||||
|
||||
/** |
||||
* started flag |
||||
*/ |
||||
private final AtomicBoolean isStarted = new AtomicBoolean(false); |
||||
|
||||
/** |
||||
* worker group |
||||
*/ |
||||
private final NioEventLoopGroup workerGroup; |
||||
|
||||
/** |
||||
* client handler |
||||
*/ |
||||
private final NettyClientHandler clientHandler = new NettyClientHandler(this); |
||||
|
||||
/** |
||||
* netty client config |
||||
*/ |
||||
private final NettyClientConfig clientConfig; |
||||
|
||||
/** |
||||
* netty client init |
||||
* |
||||
* @param clientConfig client config |
||||
*/ |
||||
public NettyRemotingClient(final NettyClientConfig clientConfig){ |
||||
this.clientConfig = clientConfig; |
||||
this.workerGroup = new NioEventLoopGroup(clientConfig.getWorkerThreads(), new ThreadFactory() { |
||||
private AtomicInteger threadIndex = new AtomicInteger(0); |
||||
|
||||
@Override |
||||
public Thread newThread(Runnable r) { |
||||
return new Thread(r, String.format("NettyClient_%d", this.threadIndex.incrementAndGet())); |
||||
} |
||||
}); |
||||
this.start(); |
||||
} |
||||
|
||||
/** |
||||
* netty server start |
||||
*/ |
||||
private void start(){ |
||||
|
||||
this.bootstrap |
||||
.group(this.workerGroup) |
||||
.channel(NioSocketChannel.class) |
||||
.option(ChannelOption.SO_KEEPALIVE, clientConfig.isSoKeepalive()) |
||||
.option(ChannelOption.TCP_NODELAY, clientConfig.isTcpNoDelay()) |
||||
.option(ChannelOption.SO_SNDBUF, clientConfig.getSendBufferSize()) |
||||
.option(ChannelOption.SO_RCVBUF, clientConfig.getReceiveBufferSize()) |
||||
.handler(new ChannelInitializer<SocketChannel>() { |
||||
@Override |
||||
public void initChannel(SocketChannel ch) throws Exception { |
||||
ch.pipeline().addLast( |
||||
new NettyDecoder(), |
||||
clientHandler, |
||||
encoder); |
||||
} |
||||
}); |
||||
isStarted.compareAndSet(false, true); |
||||
} |
||||
|
||||
/** |
||||
* register processor |
||||
* |
||||
* @param commandType command type |
||||
* @param processor processor |
||||
*/ |
||||
public void registerProcessor(final CommandType commandType, final NettyRequestProcessor processor) { |
||||
registerProcessor(commandType, processor, null); |
||||
} |
||||
|
||||
/** |
||||
* register processor |
||||
* |
||||
* @param commandType command type |
||||
* @param processor processor |
||||
* @param executor thread executor |
||||
*/ |
||||
public void registerProcessor(final CommandType commandType, final NettyRequestProcessor processor, final ExecutorService executor) { |
||||
this.clientHandler.registerProcessor(commandType, processor, executor); |
||||
} |
||||
|
||||
/** |
||||
* send connect |
||||
* @param address address |
||||
* @param command command |
||||
* @throws RemotingException |
||||
*/ |
||||
public void send(final Address address, final Command command) throws RemotingException { |
||||
final Channel channel = getChannel(address); |
||||
if (channel == null) { |
||||
throw new RemotingException("network error"); |
||||
} |
||||
try { |
||||
channel.writeAndFlush(command).addListener(new ChannelFutureListener(){ |
||||
|
||||
@Override |
||||
public void operationComplete(ChannelFuture future) throws Exception { |
||||
if(future.isSuccess()){ |
||||
logger.info("sent command {} to {}", command, address); |
||||
} else{ |
||||
logger.error("send command {} to {} failed, error {}", command, address, future.cause()); |
||||
} |
||||
} |
||||
}); |
||||
} catch (Exception ex) { |
||||
String msg = String.format("send command %s to address %s encounter error", command, address); |
||||
throw new RemotingException(msg, ex); |
||||
} |
||||
} |
||||
|
||||
/** |
||||
* get channel |
||||
* @param address address |
||||
* @return channel |
||||
*/ |
||||
public Channel getChannel(Address address) { |
||||
Channel channel = channels.get(address); |
||||
if(channel != null && channel.isActive()){ |
||||
return channel; |
||||
} |
||||
return createChannel(address, true); |
||||
} |
||||
|
||||
/** |
||||
* create channel |
||||
* @param address address |
||||
* @param isSync is sync |
||||
* @return channel |
||||
*/ |
||||
public Channel createChannel(Address address, boolean isSync) { |
||||
ChannelFuture future; |
||||
try { |
||||
synchronized (bootstrap){ |
||||
future = bootstrap.connect(new InetSocketAddress(address.getHost(), address.getPort())); |
||||
} |
||||
if(isSync){ |
||||
future.sync(); |
||||
} |
||||
if (future.isSuccess()) { |
||||
Channel channel = future.channel(); |
||||
channels.put(address, channel); |
||||
return channel; |
||||
} |
||||
} catch (Exception ex) { |
||||
logger.info("connect to {} error {}", address, ex); |
||||
} |
||||
return null; |
||||
} |
||||
|
||||
/** |
||||
* get default thread executor |
||||
* @return thread executor |
||||
*/ |
||||
public ExecutorService getDefaultExecutor() { |
||||
return defaultExecutor; |
||||
} |
||||
|
||||
/** |
||||
* close client |
||||
*/ |
||||
public void close() { |
||||
if(isStarted.compareAndSet(true, false)){ |
||||
try { |
||||
closeChannels(); |
||||
if(workerGroup != null){ |
||||
this.workerGroup.shutdownGracefully(); |
||||
} |
||||
if(defaultExecutor != null){ |
||||
defaultExecutor.shutdown(); |
||||
} |
||||
} catch (Exception ex) { |
||||
logger.error("netty client close exception", ex); |
||||
} |
||||
logger.info("netty client closed"); |
||||
} |
||||
} |
||||
|
||||
/** |
||||
* close channel |
||||
*/ |
||||
private void closeChannels(){ |
||||
for (Channel channel : this.channels.values()) { |
||||
channel.close(); |
||||
} |
||||
this.channels.clear(); |
||||
} |
||||
|
||||
/** |
||||
* remove channel |
||||
* @param address address |
||||
*/ |
||||
public void removeChannel(Address address){ |
||||
Channel channel = this.channels.remove(address); |
||||
if(channel != null){ |
||||
channel.close(); |
||||
} |
||||
} |
||||
} |
||||
@ -0,0 +1,220 @@
|
||||
/* |
||||
* Licensed to the Apache Software Foundation (ASF) under one or more |
||||
* contributor license agreements. See the NOTICE file distributed with |
||||
* this work for additional information regarding copyright ownership. |
||||
* The ASF licenses this file to You under the Apache License, Version 2.0 |
||||
* (the "License"); you may not use this file except in compliance with |
||||
* the License. You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
*/ |
||||
|
||||
package org.apache.dolphinscheduler.remote; |
||||
|
||||
import io.netty.bootstrap.ServerBootstrap; |
||||
import io.netty.channel.ChannelFuture; |
||||
import io.netty.channel.ChannelInitializer; |
||||
import io.netty.channel.ChannelOption; |
||||
import io.netty.channel.ChannelPipeline; |
||||
import io.netty.channel.nio.NioEventLoopGroup; |
||||
import io.netty.channel.socket.nio.NioServerSocketChannel; |
||||
import io.netty.channel.socket.nio.NioSocketChannel; |
||||
import org.apache.dolphinscheduler.remote.codec.NettyDecoder; |
||||
import org.apache.dolphinscheduler.remote.codec.NettyEncoder; |
||||
import org.apache.dolphinscheduler.remote.command.CommandType; |
||||
import org.apache.dolphinscheduler.remote.config.NettyServerConfig; |
||||
import org.apache.dolphinscheduler.remote.handler.NettyServerHandler; |
||||
import org.apache.dolphinscheduler.remote.processor.NettyRequestProcessor; |
||||
import org.apache.dolphinscheduler.remote.utils.Constants; |
||||
import org.slf4j.Logger; |
||||
import org.slf4j.LoggerFactory; |
||||
|
||||
import java.util.concurrent.ExecutorService; |
||||
import java.util.concurrent.Executors; |
||||
import java.util.concurrent.ThreadFactory; |
||||
import java.util.concurrent.atomic.AtomicBoolean; |
||||
import java.util.concurrent.atomic.AtomicInteger; |
||||
|
||||
/** |
||||
* remoting netty server |
||||
*/ |
||||
public class NettyRemotingServer { |
||||
|
||||
private final Logger logger = LoggerFactory.getLogger(NettyRemotingServer.class); |
||||
|
||||
/** |
||||
* server bootstart |
||||
*/ |
||||
private final ServerBootstrap serverBootstrap = new ServerBootstrap(); |
||||
|
||||
/** |
||||
* encoder |
||||
*/ |
||||
private final NettyEncoder encoder = new NettyEncoder(); |
||||
|
||||
/** |
||||
* default executor |
||||
*/ |
||||
private final ExecutorService defaultExecutor = Executors.newFixedThreadPool(Constants.CPUS); |
||||
|
||||
/** |
||||
* boss group |
||||
*/ |
||||
private final NioEventLoopGroup bossGroup; |
||||
|
||||
/** |
||||
* worker group |
||||
*/ |
||||
private final NioEventLoopGroup workGroup; |
||||
|
||||
/** |
||||
* server config |
||||
*/ |
||||
private final NettyServerConfig serverConfig; |
||||
|
||||
/** |
||||
* server handler |
||||
*/ |
||||
private final NettyServerHandler serverHandler = new NettyServerHandler(this); |
||||
|
||||
/** |
||||
* started flag |
||||
*/ |
||||
private final AtomicBoolean isStarted = new AtomicBoolean(false); |
||||
|
||||
/** |
||||
* server init |
||||
* |
||||
* @param serverConfig server config |
||||
*/ |
||||
public NettyRemotingServer(final NettyServerConfig serverConfig){ |
||||
this.serverConfig = serverConfig; |
||||
|
||||
this.bossGroup = new NioEventLoopGroup(1, new ThreadFactory() { |
||||
private AtomicInteger threadIndex = new AtomicInteger(0); |
||||
|
||||
@Override |
||||
public Thread newThread(Runnable r) { |
||||
return new Thread(r, String.format("NettyServerBossThread_%d", this.threadIndex.incrementAndGet())); |
||||
} |
||||
}); |
||||
|
||||
this.workGroup = new NioEventLoopGroup(serverConfig.getWorkerThread(), new ThreadFactory() { |
||||
private AtomicInteger threadIndex = new AtomicInteger(0); |
||||
|
||||
@Override |
||||
public Thread newThread(Runnable r) { |
||||
return new Thread(r, String.format("NettyServerWorkerThread_%d", this.threadIndex.incrementAndGet())); |
||||
} |
||||
}); |
||||
} |
||||
|
||||
/** |
||||
* server start |
||||
*/ |
||||
public void start(){ |
||||
|
||||
if(this.isStarted.get()){ |
||||
return; |
||||
} |
||||
|
||||
this.serverBootstrap |
||||
.group(this.bossGroup, this.workGroup) |
||||
.channel(NioServerSocketChannel.class) |
||||
.option(ChannelOption.SO_REUSEADDR, true) |
||||
.option(ChannelOption.SO_BACKLOG, serverConfig.getSoBacklog()) |
||||
.childOption(ChannelOption.SO_KEEPALIVE, serverConfig.isSoKeepalive()) |
||||
.childOption(ChannelOption.TCP_NODELAY, serverConfig.isTcpNoDelay()) |
||||
.childOption(ChannelOption.SO_SNDBUF, serverConfig.getSendBufferSize()) |
||||
.childOption(ChannelOption.SO_RCVBUF, serverConfig.getReceiveBufferSize()) |
||||
.childHandler(new ChannelInitializer<NioSocketChannel>() { |
||||
|
||||
@Override |
||||
protected void initChannel(NioSocketChannel ch) throws Exception { |
||||
initNettyChannel(ch); |
||||
} |
||||
}); |
||||
|
||||
ChannelFuture future; |
||||
try { |
||||
future = serverBootstrap.bind(serverConfig.getListenPort()).sync(); |
||||
} catch (Exception e) { |
||||
logger.error("NettyRemotingServer bind fail {}, exit", e); |
||||
throw new RuntimeException(String.format("NettyRemotingServer bind %s fail", serverConfig.getListenPort())); |
||||
} |
||||
if (future.isSuccess()) { |
||||
logger.info("NettyRemotingServer bind success at port : {}", serverConfig.getListenPort()); |
||||
} else if (future.cause() != null) { |
||||
throw new RuntimeException(String.format("NettyRemotingServer bind %s fail", serverConfig.getListenPort()), future.cause()); |
||||
} else { |
||||
throw new RuntimeException(String.format("NettyRemotingServer bind %s fail", serverConfig.getListenPort())); |
||||
} |
||||
//
|
||||
isStarted.compareAndSet(false, true); |
||||
} |
||||
|
||||
/** |
||||
* init netty channel |
||||
* @param ch socket channel |
||||
* @throws Exception |
||||
*/ |
||||
private void initNettyChannel(NioSocketChannel ch) throws Exception{ |
||||
ChannelPipeline pipeline = ch.pipeline(); |
||||
pipeline.addLast("encoder", encoder); |
||||
pipeline.addLast("decoder", new NettyDecoder()); |
||||
pipeline.addLast("handler", serverHandler); |
||||
} |
||||
|
||||
/** |
||||
* register processor |
||||
* @param commandType command type |
||||
* @param processor processor |
||||
*/ |
||||
public void registerProcessor(final CommandType commandType, final NettyRequestProcessor processor) { |
||||
this.registerProcessor(commandType, processor, null); |
||||
} |
||||
|
||||
/** |
||||
* register processor |
||||
* |
||||
* @param commandType command type |
||||
* @param processor processor |
||||
* @param executor thread executor |
||||
*/ |
||||
public void registerProcessor(final CommandType commandType, final NettyRequestProcessor processor, final ExecutorService executor) { |
||||
this.serverHandler.registerProcessor(commandType, processor, executor); |
||||
} |
||||
|
||||
/** |
||||
* get default thread executor |
||||
* @return thread executor |
||||
*/ |
||||
public ExecutorService getDefaultExecutor() { |
||||
return defaultExecutor; |
||||
} |
||||
|
||||
public void close() { |
||||
if(isStarted.compareAndSet(true, false)){ |
||||
try { |
||||
if(bossGroup != null){ |
||||
this.bossGroup.shutdownGracefully(); |
||||
} |
||||
if(workGroup != null){ |
||||
this.workGroup.shutdownGracefully(); |
||||
} |
||||
if(defaultExecutor != null){ |
||||
defaultExecutor.shutdown(); |
||||
} |
||||
} catch (Exception ex) { |
||||
logger.error("netty server close exception", ex); |
||||
} |
||||
logger.info("netty server closed"); |
||||
} |
||||
} |
||||
} |
||||
@ -0,0 +1,109 @@
|
||||
/* |
||||
* Licensed to the Apache Software Foundation (ASF) under one or more |
||||
* contributor license agreements. See the NOTICE file distributed with |
||||
* this work for additional information regarding copyright ownership. |
||||
* The ASF licenses this file to You under the Apache License, Version 2.0 |
||||
* (the "License"); you may not use this file except in compliance with |
||||
* the License. You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
*/ |
||||
|
||||
package org.apache.dolphinscheduler.remote.codec; |
||||
|
||||
|
||||
import io.netty.buffer.ByteBuf; |
||||
import io.netty.channel.ChannelHandlerContext; |
||||
import io.netty.handler.codec.ReplayingDecoder; |
||||
import org.apache.dolphinscheduler.remote.command.Command; |
||||
import org.apache.dolphinscheduler.remote.command.CommandHeader; |
||||
import org.apache.dolphinscheduler.remote.command.CommandType; |
||||
|
||||
import java.util.List; |
||||
|
||||
/** |
||||
* netty decoder |
||||
*/ |
||||
public class NettyDecoder extends ReplayingDecoder<NettyDecoder.State> { |
||||
|
||||
public NettyDecoder(){ |
||||
super(State.MAGIC); |
||||
} |
||||
|
||||
private final CommandHeader commandHeader = new CommandHeader(); |
||||
|
||||
/** |
||||
* decode |
||||
* |
||||
* @param ctx channel handler context |
||||
* @param in byte buffer |
||||
* @param out out content |
||||
* @throws Exception |
||||
*/ |
||||
@Override |
||||
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception { |
||||
switch (state()){ |
||||
case MAGIC: |
||||
checkMagic(in.readByte()); |
||||
checkpoint(State.COMMAND); |
||||
case COMMAND: |
||||
commandHeader.setType(in.readByte()); |
||||
checkpoint(State.OPAQUE); |
||||
case OPAQUE: |
||||
commandHeader.setOpaque(in.readLong()); |
||||
checkpoint(State.BODY_LENGTH); |
||||
case BODY_LENGTH: |
||||
commandHeader.setBodyLength(in.readInt()); |
||||
checkpoint(State.BODY); |
||||
case BODY: |
||||
byte[] body = new byte[commandHeader.getBodyLength()]; |
||||
in.readBytes(body); |
||||
//
|
||||
Command packet = new Command(); |
||||
packet.setType(commandType(commandHeader.getType())); |
||||
packet.setOpaque(commandHeader.getOpaque()); |
||||
packet.setBody(body); |
||||
out.add(packet); |
||||
//
|
||||
checkpoint(State.MAGIC); |
||||
} |
||||
} |
||||
|
||||
/** |
||||
* get command type |
||||
* @param type type |
||||
* @return |
||||
*/ |
||||
private CommandType commandType(byte type){ |
||||
for(CommandType ct : CommandType.values()){ |
||||
if(ct.ordinal() == type){ |
||||
return ct; |
||||
} |
||||
} |
||||
return null; |
||||
} |
||||
|
||||
/** |
||||
* check magic |
||||
* @param magic magic |
||||
*/ |
||||
private void checkMagic(byte magic) { |
||||
if (magic != Command.MAGIC) { |
||||
throw new IllegalArgumentException("illegal packet [magic]" + magic); |
||||
} |
||||
} |
||||
|
||||
enum State{ |
||||
MAGIC, |
||||
COMMAND, |
||||
OPAQUE, |
||||
BODY_LENGTH, |
||||
BODY; |
||||
} |
||||
} |
||||
@ -0,0 +1,52 @@
|
||||
/* |
||||
* Licensed to the Apache Software Foundation (ASF) under one or more |
||||
* contributor license agreements. See the NOTICE file distributed with |
||||
* this work for additional information regarding copyright ownership. |
||||
* The ASF licenses this file to You under the Apache License, Version 2.0 |
||||
* (the "License"); you may not use this file except in compliance with |
||||
* the License. You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
*/ |
||||
package org.apache.dolphinscheduler.remote.codec; |
||||
|
||||
import io.netty.buffer.ByteBuf; |
||||
import io.netty.channel.ChannelHandler.Sharable; |
||||
import io.netty.channel.ChannelHandlerContext; |
||||
import io.netty.handler.codec.MessageToByteEncoder; |
||||
import org.apache.dolphinscheduler.remote.command.Command; |
||||
|
||||
/** |
||||
* netty encoder |
||||
*/ |
||||
@Sharable |
||||
public class NettyEncoder extends MessageToByteEncoder<Command> { |
||||
|
||||
/** |
||||
* encode |
||||
* |
||||
* @param ctx channel handler context |
||||
* @param msg command |
||||
* @param out byte buffer |
||||
* @throws Exception |
||||
*/ |
||||
@Override |
||||
protected void encode(ChannelHandlerContext ctx, Command msg, ByteBuf out) throws Exception { |
||||
if(msg == null){ |
||||
throw new Exception("encode msg is null"); |
||||
} |
||||
out.writeByte(Command.MAGIC); |
||||
out.writeByte(msg.getType().ordinal()); |
||||
out.writeLong(msg.getOpaque()); |
||||
out.writeInt(msg.getBody().length); |
||||
out.writeBytes(msg.getBody()); |
||||
} |
||||
|
||||
} |
||||
|
||||
@ -0,0 +1,105 @@
|
||||
/* |
||||
* Licensed to the Apache Software Foundation (ASF) under one or more |
||||
* contributor license agreements. See the NOTICE file distributed with |
||||
* this work for additional information regarding copyright ownership. |
||||
* The ASF licenses this file to You under the Apache License, Version 2.0 |
||||
* (the "License"); you may not use this file except in compliance with |
||||
* the License. You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
*/ |
||||
package org.apache.dolphinscheduler.remote.command; |
||||
|
||||
import java.io.Serializable; |
||||
|
||||
/** |
||||
* receive task log request command and content fill |
||||
* for netty data serializable transfer |
||||
*/ |
||||
public class Command implements Serializable { |
||||
|
||||
private static final long serialVersionUID = 1L; |
||||
|
||||
public static final byte MAGIC = (byte) 0xbabe; |
||||
|
||||
public Command(){ |
||||
} |
||||
|
||||
public Command(long opaque){ |
||||
this.opaque = opaque; |
||||
} |
||||
|
||||
/** |
||||
* command type |
||||
*/ |
||||
private CommandType type; |
||||
|
||||
/** |
||||
* request unique identification |
||||
*/ |
||||
private long opaque; |
||||
|
||||
/** |
||||
* data body |
||||
*/ |
||||
private byte[] body; |
||||
|
||||
public CommandType getType() { |
||||
return type; |
||||
} |
||||
|
||||
public void setType(CommandType type) { |
||||
this.type = type; |
||||
} |
||||
|
||||
public long getOpaque() { |
||||
return opaque; |
||||
} |
||||
|
||||
public void setOpaque(long opaque) { |
||||
this.opaque = opaque; |
||||
} |
||||
|
||||
public byte[] getBody() { |
||||
return body; |
||||
} |
||||
|
||||
public void setBody(byte[] body) { |
||||
this.body = body; |
||||
} |
||||
|
||||
@Override |
||||
public int hashCode() { |
||||
final int prime = 31; |
||||
int result = 1; |
||||
result = prime * result + (int) (opaque ^ (opaque >>> 32)); |
||||
return result; |
||||
} |
||||
|
||||
@Override |
||||
public boolean equals(Object obj) { |
||||
if (this == obj) { |
||||
return true; |
||||
} |
||||
if (obj == null) { |
||||
return false; |
||||
} |
||||
if (getClass() != obj.getClass()) { |
||||
return false; |
||||
} |
||||
Command other = (Command) obj; |
||||
return opaque == other.opaque; |
||||
} |
||||
|
||||
@Override |
||||
public String toString() { |
||||
return "Command [type=" + type + ", opaque=" + opaque + ", bodyLen=" + (body == null ? 0 : body.length) + "]"; |
||||
} |
||||
|
||||
} |
||||
@ -0,0 +1,64 @@
|
||||
/* |
||||
* Licensed to the Apache Software Foundation (ASF) under one or more |
||||
* contributor license agreements. See the NOTICE file distributed with |
||||
* this work for additional information regarding copyright ownership. |
||||
* The ASF licenses this file to You under the Apache License, Version 2.0 |
||||
* (the "License"); you may not use this file except in compliance with |
||||
* the License. You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
*/ |
||||
package org.apache.dolphinscheduler.remote.command; |
||||
|
||||
import java.io.Serializable; |
||||
|
||||
/** |
||||
* command header |
||||
*/ |
||||
public class CommandHeader implements Serializable { |
||||
|
||||
/** |
||||
* type |
||||
*/ |
||||
private byte type; |
||||
|
||||
/** |
||||
* request unique identification |
||||
*/ |
||||
private long opaque; |
||||
|
||||
/** |
||||
* body length |
||||
*/ |
||||
private int bodyLength; |
||||
|
||||
public int getBodyLength() { |
||||
return bodyLength; |
||||
} |
||||
|
||||
public void setBodyLength(int bodyLength) { |
||||
this.bodyLength = bodyLength; |
||||
} |
||||
|
||||
public byte getType() { |
||||
return type; |
||||
} |
||||
|
||||
public void setType(byte type) { |
||||
this.type = type; |
||||
} |
||||
|
||||
public long getOpaque() { |
||||
return opaque; |
||||
} |
||||
|
||||
public void setOpaque(long opaque) { |
||||
this.opaque = opaque; |
||||
} |
||||
} |
||||
@ -0,0 +1 @@
|
||||
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.remote.command;
public enum CommandType {
/**
* roll view log request
*/
ROLL_VIEW_LOG_REQUEST,
/**
* roll view log response
*/
ROLL_VIEW_LOG_RESPONSE,
/**
* view whole log request
*/
VIEW_WHOLE_LOG_REQUEST,
/**
* view whole log response
*/
VIEW_WHOLE_LOG_RESPONSE,
/**
* get log bytes request
*/
GET_LOG_BYTES_REQUEST,
/**
* get log bytes response
*/
GET_LOG_BYTES_RESPONSE,
WORKER_REQUEST,
MASTER_RESPONSE,
/**
* execute task request
*/
EXECUTE_TASK_REQUEST,
/**
* execute task response
*/
EXECUTE_TASK_RESPONSE,
/**
* ping
*/
PING,
/**
* pong
*/
PONG;
} |
||||
@ -0,0 +1 @@
|
||||
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.remote.command;
import org.apache.dolphinscheduler.remote.utils.FastJsonSerializer;
import java.io.Serializable;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;
/**
* execute task request command
*/
public class ExecuteTaskRequestCommand implements Serializable {
private static final AtomicLong REQUEST = new AtomicLong(1);
/**
* task id
*/
private String taskId;
/**
* attempt id
*/
private String attemptId;
/**
* application name
*/
private String applicationName;
/**
* group name
*/
private String groupName;
/**
* task name
*/
private String taskName;
/**
* connect port
*/
private int connectorPort;
/**
* description info
*/
private String description;
/**
* class name
*/
private String className;
/**
* method name
*/
private String methodName;
/**
* params
*/
private String params;
/**
* shard items
*/
private List<Integer> shardItems;
public List<Integer> getShardItems() {
return shardItems;
}
public void setShardItems(List<Integer> shardItems) {
this.shardItems = shardItems;
}
public String getParams() {
return params;
}
public void setParams(String params) {
this.params = params;
}
public String getTaskId() {
return taskId;
}
public void setTaskId(String taskId) {
this.taskId = taskId;
}
public String getApplicationName() {
return applicationName;
}
public void setApplicationName(String applicationName) {
this.applicationName = applicationName;
}
public String getGroupName() {
return groupName;
}
public void setGroupName(String groupName) {
this.groupName = groupName;
}
public String getTaskName() {
return taskName;
}
public void setTaskName(String taskName) {
this.taskName = taskName;
}
public int getConnectorPort() {
return connectorPort;
}
public void setConnectorPort(int connectorPort) {
this.connectorPort = connectorPort;
}
public String getDescription() {
return description;
}
public void setDescription(String description) {
this.description = description;
}
public String getClassName() {
return className;
}
public void setClassName(String className) {
this.className = className;
}
public String getMethodName() {
return methodName;
}
public void setMethodName(String methodName) {
this.methodName = methodName;
}
/**
* package request command
*
* @return command
*/
public Command convert2Command(){
Command command = new Command(REQUEST.getAndIncrement());
command.setType(CommandType.EXECUTE_TASK_REQUEST);
byte[] body = FastJsonSerializer.serialize(this);
command.setBody(body);
return command;
}
} |
||||
@ -0,0 +1 @@
|
||||
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.remote.command;
import org.apache.dolphinscheduler.remote.utils.FastJsonSerializer;
import java.io.Serializable;
import java.util.concurrent.atomic.AtomicLong;
/**
* execute taks response command
*/
public class ExecuteTaskResponseCommand implements Serializable {
private static final AtomicLong REQUEST = new AtomicLong(1);
/**
* task id
*/
private String taskId;
/**
* attempt id
*/
private String attemptId;
/**
* result info
*/
private Object result;
/**
* receive time
*/
private long receivedTime;
/**
* execute count
*/
private int executeCount;
/**
* execute time
*/
private long executeTime;
public String getAttemptId() {
return attemptId;
}
public void setAttemptId(String attemptId) {
this.attemptId = attemptId;
}
public String getTaskId() {
return taskId;
}
public void setTaskId(String taskId) {
this.taskId = taskId;
}
public Object getResult() {
return result;
}
public void setResult(Object result) {
this.result = result;
}
public long getReceivedTime() {
return receivedTime;
}
public void setReceivedTime(long receivedTime) {
this.receivedTime = receivedTime;
}
public int getExecuteCount() {
return executeCount;
}
public void setExecuteCount(int executeCount) {
this.executeCount = executeCount;
}
public long getExecuteTime() {
return executeTime;
}
public void setExecuteTime(long executeTime) {
this.executeTime = executeTime;
}
/**
* package response command
*
* @return command
*/
public Command convert2Command(){
Command command = new Command(REQUEST.getAndIncrement());
command.setType(CommandType.EXECUTE_TASK_RESPONSE);
byte[] body = FastJsonSerializer.serialize(this);
command.setBody(body);
return command;
}
} |
||||
@ -0,0 +1,74 @@
|
||||
/* |
||||
* Licensed to the Apache Software Foundation (ASF) under one or more |
||||
* contributor license agreements. See the NOTICE file distributed with |
||||
* this work for additional information regarding copyright ownership. |
||||
* The ASF licenses this file to You under the Apache License, Version 2.0 |
||||
* (the "License"); you may not use this file except in compliance with |
||||
* the License. You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
*/ |
||||
|
||||
package org.apache.dolphinscheduler.remote.command; |
||||
|
||||
import io.netty.buffer.ByteBuf; |
||||
import io.netty.buffer.Unpooled; |
||||
|
||||
import java.io.Serializable; |
||||
import java.util.concurrent.atomic.AtomicLong; |
||||
|
||||
/** |
||||
* ping machine |
||||
*/ |
||||
public class Ping implements Serializable { |
||||
|
||||
private static final AtomicLong ID = new AtomicLong(1); |
||||
|
||||
/** |
||||
* ping body |
||||
*/ |
||||
protected static ByteBuf EMPTY_BODY = Unpooled.EMPTY_BUFFER; |
||||
|
||||
/** |
||||
* request command body |
||||
*/ |
||||
private static byte[] EMPTY_BODY_ARRAY = new byte[0]; |
||||
|
||||
private static final ByteBuf PING_BUF; |
||||
|
||||
static { |
||||
ByteBuf ping = Unpooled.buffer(); |
||||
ping.writeByte(Command.MAGIC); |
||||
ping.writeByte(CommandType.PING.ordinal()); |
||||
ping.writeLong(0); |
||||
ping.writeInt(0); |
||||
ping.writeBytes(EMPTY_BODY); |
||||
PING_BUF = Unpooled.unreleasableBuffer(ping).asReadOnly(); |
||||
} |
||||
|
||||
/** |
||||
* ping connect |
||||
* @return result |
||||
*/ |
||||
public static ByteBuf pingContent(){ |
||||
return PING_BUF.duplicate(); |
||||
} |
||||
|
||||
/** |
||||
* package ping command |
||||
* |
||||
* @return command |
||||
*/ |
||||
public static Command create(){ |
||||
Command command = new Command(ID.getAndIncrement()); |
||||
command.setType(CommandType.PING); |
||||
command.setBody(EMPTY_BODY_ARRAY); |
||||
return command; |
||||
} |
||||
} |
||||
@ -0,0 +1,75 @@
|
||||
/* |
||||
* Licensed to the Apache Software Foundation (ASF) under one or more |
||||
* contributor license agreements. See the NOTICE file distributed with |
||||
* this work for additional information regarding copyright ownership. |
||||
* The ASF licenses this file to You under the Apache License, Version 2.0 |
||||
* (the "License"); you may not use this file except in compliance with |
||||
* the License. You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
*/ |
||||
|
||||
package org.apache.dolphinscheduler.remote.command; |
||||
|
||||
import io.netty.buffer.ByteBuf; |
||||
import io.netty.buffer.Unpooled; |
||||
|
||||
import java.io.Serializable; |
||||
|
||||
/** |
||||
* Pong return after ping |
||||
*/ |
||||
public class Pong implements Serializable { |
||||
|
||||
/** |
||||
* pong body |
||||
*/ |
||||
protected static ByteBuf EMPTY_BODY = Unpooled.EMPTY_BUFFER; |
||||
|
||||
/** |
||||
* pong command body |
||||
*/ |
||||
private static byte[] EMPTY_BODY_ARRAY = new byte[0]; |
||||
|
||||
/** |
||||
* ping byte buffer |
||||
*/ |
||||
private static final ByteBuf PONG_BUF; |
||||
|
||||
static { |
||||
ByteBuf ping = Unpooled.buffer(); |
||||
ping.writeByte(Command.MAGIC); |
||||
ping.writeByte(CommandType.PONG.ordinal()); |
||||
ping.writeLong(0); |
||||
ping.writeInt(0); |
||||
ping.writeBytes(EMPTY_BODY); |
||||
PONG_BUF = Unpooled.unreleasableBuffer(ping).asReadOnly(); |
||||
} |
||||
|
||||
/** |
||||
* ping content |
||||
* @return result |
||||
*/ |
||||
public static ByteBuf pingContent(){ |
||||
return PONG_BUF.duplicate(); |
||||
} |
||||
|
||||
/** |
||||
* package pong command |
||||
* |
||||
* @param opaque request unique identification |
||||
* @return command |
||||
*/ |
||||
public static Command create(long opaque){ |
||||
Command command = new Command(opaque); |
||||
command.setType(CommandType.PONG); |
||||
command.setBody(EMPTY_BODY_ARRAY); |
||||
return command; |
||||
} |
||||
} |
||||
@ -0,0 +1,69 @@
|
||||
/* |
||||
* Licensed to the Apache Software Foundation (ASF) under one or more |
||||
* contributor license agreements. See the NOTICE file distributed with |
||||
* this work for additional information regarding copyright ownership. |
||||
* The ASF licenses this file to You under the Apache License, Version 2.0 |
||||
* (the "License"); you may not use this file except in compliance with |
||||
* the License. You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
*/ |
||||
|
||||
package org.apache.dolphinscheduler.remote.command.log; |
||||
|
||||
import org.apache.dolphinscheduler.remote.command.Command; |
||||
import org.apache.dolphinscheduler.remote.command.CommandType; |
||||
import org.apache.dolphinscheduler.remote.utils.FastJsonSerializer; |
||||
|
||||
import java.io.Serializable; |
||||
import java.util.concurrent.atomic.AtomicLong; |
||||
|
||||
/** |
||||
* get log bytes request command |
||||
*/ |
||||
public class GetLogBytesRequestCommand implements Serializable { |
||||
|
||||
/** |
||||
* request id |
||||
*/ |
||||
private static final AtomicLong REQUEST = new AtomicLong(1); |
||||
|
||||
/** |
||||
* log path |
||||
*/ |
||||
private String path; |
||||
|
||||
public GetLogBytesRequestCommand() { |
||||
} |
||||
|
||||
public GetLogBytesRequestCommand(String path) { |
||||
this.path = path; |
||||
} |
||||
|
||||
public String getPath() { |
||||
return path; |
||||
} |
||||
|
||||
public void setPath(String path) { |
||||
this.path = path; |
||||
} |
||||
|
||||
/** |
||||
* package request command |
||||
* |
||||
* @return command |
||||
*/ |
||||
public Command convert2Command(){ |
||||
Command command = new Command(REQUEST.getAndIncrement()); |
||||
command.setType(CommandType.GET_LOG_BYTES_REQUEST); |
||||
byte[] body = FastJsonSerializer.serialize(this); |
||||
command.setBody(body); |
||||
return command; |
||||
} |
||||
} |
||||
@ -0,0 +1,65 @@
|
||||
/* |
||||
* Licensed to the Apache Software Foundation (ASF) under one or more |
||||
* contributor license agreements. See the NOTICE file distributed with |
||||
* this work for additional information regarding copyright ownership. |
||||
* The ASF licenses this file to You under the Apache License, Version 2.0 |
||||
* (the "License"); you may not use this file except in compliance with |
||||
* the License. You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
*/ |
||||
|
||||
package org.apache.dolphinscheduler.remote.command.log; |
||||
|
||||
import org.apache.dolphinscheduler.remote.command.Command; |
||||
import org.apache.dolphinscheduler.remote.command.CommandType; |
||||
import org.apache.dolphinscheduler.remote.utils.FastJsonSerializer; |
||||
|
||||
import java.io.Serializable; |
||||
|
||||
/** |
||||
* get log bytes response command |
||||
*/ |
||||
public class GetLogBytesResponseCommand implements Serializable { |
||||
|
||||
/** |
||||
* log byte data |
||||
*/ |
||||
private byte[] data; |
||||
|
||||
public GetLogBytesResponseCommand() { |
||||
} |
||||
|
||||
public GetLogBytesResponseCommand(byte[] data) { |
||||
this.data = data; |
||||
} |
||||
|
||||
public byte[] getData() { |
||||
return data; |
||||
} |
||||
|
||||
public void setData(byte[] data) { |
||||
this.data = data; |
||||
} |
||||
|
||||
/** |
||||
* package response command |
||||
* |
||||
* @param opaque request unique identification |
||||
* @return command |
||||
*/ |
||||
public Command convert2Command(long opaque){ |
||||
Command command = new Command(opaque); |
||||
command.setType(CommandType.GET_LOG_BYTES_RESPONSE); |
||||
byte[] body = FastJsonSerializer.serialize(this); |
||||
command.setBody(body); |
||||
return command; |
||||
} |
||||
|
||||
} |
||||
@ -0,0 +1,97 @@
|
||||
/* |
||||
* Licensed to the Apache Software Foundation (ASF) under one or more |
||||
* contributor license agreements. See the NOTICE file distributed with |
||||
* this work for additional information regarding copyright ownership. |
||||
* The ASF licenses this file to You under the Apache License, Version 2.0 |
||||
* (the "License"); you may not use this file except in compliance with |
||||
* the License. You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
*/ |
||||
|
||||
package org.apache.dolphinscheduler.remote.command.log; |
||||
|
||||
import org.apache.dolphinscheduler.remote.command.Command; |
||||
import org.apache.dolphinscheduler.remote.command.CommandType; |
||||
import org.apache.dolphinscheduler.remote.utils.FastJsonSerializer; |
||||
|
||||
import java.io.Serializable; |
||||
import java.util.concurrent.atomic.AtomicLong; |
||||
|
||||
/** |
||||
* roll view log request command |
||||
*/ |
||||
public class RollViewLogRequestCommand implements Serializable { |
||||
|
||||
/** |
||||
* request id |
||||
*/ |
||||
private static final AtomicLong REQUEST = new AtomicLong(1); |
||||
|
||||
/** |
||||
* log path |
||||
*/ |
||||
private String path; |
||||
|
||||
/** |
||||
* skip line number |
||||
*/ |
||||
private int skipLineNum; |
||||
|
||||
/** |
||||
* query log line number limit |
||||
*/ |
||||
private int limit; |
||||
|
||||
public RollViewLogRequestCommand() { |
||||
} |
||||
|
||||
public RollViewLogRequestCommand(String path, int skipLineNum, int limit) { |
||||
this.path = path; |
||||
this.skipLineNum = skipLineNum; |
||||
this.limit = limit; |
||||
} |
||||
|
||||
public String getPath() { |
||||
return path; |
||||
} |
||||
|
||||
public void setPath(String path) { |
||||
this.path = path; |
||||
} |
||||
|
||||
public int getSkipLineNum() { |
||||
return skipLineNum; |
||||
} |
||||
|
||||
public void setSkipLineNum(int skipLineNum) { |
||||
this.skipLineNum = skipLineNum; |
||||
} |
||||
|
||||
public int getLimit() { |
||||
return limit; |
||||
} |
||||
|
||||
public void setLimit(int limit) { |
||||
this.limit = limit; |
||||
} |
||||
|
||||
/** |
||||
* package request command |
||||
* |
||||
* @return command |
||||
*/ |
||||
public Command convert2Command(){ |
||||
Command command = new Command(REQUEST.getAndIncrement()); |
||||
command.setType(CommandType.ROLL_VIEW_LOG_REQUEST); |
||||
byte[] body = FastJsonSerializer.serialize(this); |
||||
command.setBody(body); |
||||
return command; |
||||
} |
||||
} |
||||
@ -0,0 +1,64 @@
|
||||
/* |
||||
* Licensed to the Apache Software Foundation (ASF) under one or more |
||||
* contributor license agreements. See the NOTICE file distributed with |
||||
* this work for additional information regarding copyright ownership. |
||||
* The ASF licenses this file to You under the Apache License, Version 2.0 |
||||
* (the "License"); you may not use this file except in compliance with |
||||
* the License. You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
*/ |
||||
|
||||
package org.apache.dolphinscheduler.remote.command.log; |
||||
|
||||
import org.apache.dolphinscheduler.remote.command.Command; |
||||
import org.apache.dolphinscheduler.remote.command.CommandType; |
||||
import org.apache.dolphinscheduler.remote.utils.FastJsonSerializer; |
||||
|
||||
import java.io.Serializable; |
||||
|
||||
/** |
||||
* roll view log response command |
||||
*/ |
||||
public class RollViewLogResponseCommand implements Serializable { |
||||
|
||||
/** |
||||
* response data |
||||
*/ |
||||
private String msg; |
||||
|
||||
public RollViewLogResponseCommand() { |
||||
} |
||||
|
||||
public RollViewLogResponseCommand(String msg) { |
||||
this.msg = msg; |
||||
} |
||||
|
||||
public String getMsg() { |
||||
return msg; |
||||
} |
||||
|
||||
public void setMsg(String msg) { |
||||
this.msg = msg; |
||||
} |
||||
|
||||
/** |
||||
* package response command |
||||
* |
||||
* @param opaque request unique identification |
||||
* @return command |
||||
*/ |
||||
public Command convert2Command(long opaque){ |
||||
Command command = new Command(opaque); |
||||
command.setType(CommandType.ROLL_VIEW_LOG_RESPONSE); |
||||
byte[] body = FastJsonSerializer.serialize(this); |
||||
command.setBody(body); |
||||
return command; |
||||
} |
||||
} |
||||
@ -0,0 +1,66 @@
|
||||
/* |
||||
* Licensed to the Apache Software Foundation (ASF) under one or more |
||||
* contributor license agreements. See the NOTICE file distributed with |
||||
* this work for additional information regarding copyright ownership. |
||||
* The ASF licenses this file to You under the Apache License, Version 2.0 |
||||
* (the "License"); you may not use this file except in compliance with |
||||
* the License. You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
*/ |
||||
|
||||
package org.apache.dolphinscheduler.remote.command.log; |
||||
|
||||
import org.apache.dolphinscheduler.remote.command.Command; |
||||
import org.apache.dolphinscheduler.remote.command.CommandType; |
||||
import org.apache.dolphinscheduler.remote.utils.FastJsonSerializer; |
||||
|
||||
import java.io.Serializable; |
||||
import java.util.concurrent.atomic.AtomicLong; |
||||
|
||||
/** |
||||
* view log request command |
||||
*/ |
||||
public class ViewLogRequestCommand implements Serializable { |
||||
|
||||
/** |
||||
* request id |
||||
*/ |
||||
private static final AtomicLong REQUEST = new AtomicLong(1); |
||||
|
||||
private String path; |
||||
|
||||
public ViewLogRequestCommand() { |
||||
} |
||||
|
||||
public ViewLogRequestCommand(String path) { |
||||
this.path = path; |
||||
} |
||||
|
||||
public String getPath() { |
||||
return path; |
||||
} |
||||
|
||||
public void setPath(String path) { |
||||
this.path = path; |
||||
} |
||||
|
||||
/** |
||||
* package request command |
||||
* |
||||
* @return command |
||||
*/ |
||||
public Command convert2Command(){ |
||||
Command command = new Command(REQUEST.getAndIncrement()); |
||||
command.setType(CommandType.VIEW_WHOLE_LOG_REQUEST); |
||||
byte[] body = FastJsonSerializer.serialize(this); |
||||
command.setBody(body); |
||||
return command; |
||||
} |
||||
} |
||||
@ -0,0 +1,64 @@
|
||||
/* |
||||
* Licensed to the Apache Software Foundation (ASF) under one or more |
||||
* contributor license agreements. See the NOTICE file distributed with |
||||
* this work for additional information regarding copyright ownership. |
||||
* The ASF licenses this file to You under the Apache License, Version 2.0 |
||||
* (the "License"); you may not use this file except in compliance with |
||||
* the License. You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
*/ |
||||
|
||||
package org.apache.dolphinscheduler.remote.command.log; |
||||
|
||||
import org.apache.dolphinscheduler.remote.command.Command; |
||||
import org.apache.dolphinscheduler.remote.command.CommandType; |
||||
import org.apache.dolphinscheduler.remote.utils.FastJsonSerializer; |
||||
|
||||
import java.io.Serializable; |
||||
|
||||
/** |
||||
* view log response command |
||||
*/ |
||||
public class ViewLogResponseCommand implements Serializable { |
||||
|
||||
/** |
||||
* response data |
||||
*/ |
||||
private String msg; |
||||
|
||||
public ViewLogResponseCommand() { |
||||
} |
||||
|
||||
public ViewLogResponseCommand(String msg) { |
||||
this.msg = msg; |
||||
} |
||||
|
||||
public String getMsg() { |
||||
return msg; |
||||
} |
||||
|
||||
public void setMsg(String msg) { |
||||
this.msg = msg; |
||||
} |
||||
|
||||
/** |
||||
* package response command |
||||
* |
||||
* @param opaque request unique identification |
||||
* @return command |
||||
*/ |
||||
public Command convert2Command(long opaque){ |
||||
Command command = new Command(opaque); |
||||
command.setType(CommandType.VIEW_WHOLE_LOG_RESPONSE); |
||||
byte[] body = FastJsonSerializer.serialize(this); |
||||
command.setBody(body); |
||||
return command; |
||||
} |
||||
} |
||||
@ -0,0 +1,91 @@
|
||||
/* |
||||
* Licensed to the Apache Software Foundation (ASF) under one or more |
||||
* contributor license agreements. See the NOTICE file distributed with |
||||
* this work for additional information regarding copyright ownership. |
||||
* The ASF licenses this file to You under the Apache License, Version 2.0 |
||||
* (the "License"); you may not use this file except in compliance with |
||||
* the License. You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
*/ |
||||
package org.apache.dolphinscheduler.remote.config; |
||||
|
||||
import org.apache.dolphinscheduler.remote.utils.Constants; |
||||
|
||||
/** |
||||
* netty client config |
||||
*/ |
||||
public class NettyClientConfig { |
||||
|
||||
/** |
||||
* worker threads，default get machine cpus |
||||
*/ |
||||
private int workerThreads = Constants.CPUS; |
||||
|
||||
/** |
||||
* whether tpc delay |
||||
*/ |
||||
private boolean tcpNoDelay = true; |
||||
|
||||
/** |
||||
* whether keep alive |
||||
*/ |
||||
private boolean soKeepalive = true; |
||||
|
||||
/** |
||||
* send buffer size |
||||
*/ |
||||
private int sendBufferSize = 65535; |
||||
|
||||
/** |
||||
* receive buffer size |
||||
*/ |
||||
private int receiveBufferSize = 65535; |
||||
|
||||
public int getWorkerThreads() { |
||||
return workerThreads; |
||||
} |
||||
|
||||
public void setWorkerThreads(int workerThreads) { |
||||
this.workerThreads = workerThreads; |
||||
} |
||||
|
||||
public boolean isTcpNoDelay() { |
||||
return tcpNoDelay; |
||||
} |
||||
|
||||
public void setTcpNoDelay(boolean tcpNoDelay) { |
||||
this.tcpNoDelay = tcpNoDelay; |
||||
} |
||||
|
||||
public boolean isSoKeepalive() { |
||||
return soKeepalive; |
||||
} |
||||
|
||||
public void setSoKeepalive(boolean soKeepalive) { |
||||
this.soKeepalive = soKeepalive; |
||||
} |
||||
|
||||
public int getSendBufferSize() { |
||||
return sendBufferSize; |
||||
} |
||||
|
||||
public void setSendBufferSize(int sendBufferSize) { |
||||
this.sendBufferSize = sendBufferSize; |
||||
} |
||||
|
||||
public int getReceiveBufferSize() { |
||||
return receiveBufferSize; |
||||
} |
||||
|
||||
public void setReceiveBufferSize(int receiveBufferSize) { |
||||
this.receiveBufferSize = receiveBufferSize; |
||||
} |
||||
|
||||
} |
||||
@ -0,0 +1,116 @@
|
||||
/* |
||||
* Licensed to the Apache Software Foundation (ASF) under one or more |
||||
* contributor license agreements. See the NOTICE file distributed with |
||||
* this work for additional information regarding copyright ownership. |
||||
* The ASF licenses this file to You under the Apache License, Version 2.0 |
||||
* (the "License"); you may not use this file except in compliance with |
||||
* the License. You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
*/ |
||||
package org.apache.dolphinscheduler.remote.config; |
||||
|
||||
import org.apache.dolphinscheduler.remote.utils.Constants; |
||||
|
||||
/** |
||||
* netty server config |
||||
*/ |
||||
public class NettyServerConfig { |
||||
|
||||
/** |
||||
* init the server connectable queue |
||||
*/ |
||||
private int soBacklog = 1024; |
||||
|
||||
/** |
||||
* whether tpc delay |
||||
*/ |
||||
private boolean tcpNoDelay = true; |
||||
|
||||
/** |
||||
* whether keep alive |
||||
*/ |
||||
private boolean soKeepalive = true; |
||||
|
||||
/** |
||||
* send buffer size |
||||
*/ |
||||
private int sendBufferSize = 65535; |
||||
|
||||
/** |
||||
* receive buffer size |
||||
*/ |
||||
private int receiveBufferSize = 65535; |
||||
|
||||
/** |
||||
* worker threads，default get machine cpus |
||||
*/ |
||||
private int workerThread = Constants.CPUS; |
||||
|
||||
/** |
||||
* listen port |
||||
*/ |
||||
private int listenPort = 12346; |
||||
|
||||
public int getListenPort() { |
||||
return listenPort; |
||||
} |
||||
|
||||
public void setListenPort(int listenPort) { |
||||
this.listenPort = listenPort; |
||||
} |
||||
|
||||
public int getSoBacklog() { |
||||
return soBacklog; |
||||
} |
||||
|
||||
public void setSoBacklog(int soBacklog) { |
||||
this.soBacklog = soBacklog; |
||||
} |
||||
|
||||
public boolean isTcpNoDelay() { |
||||
return tcpNoDelay; |
||||
} |
||||
|
||||
public void setTcpNoDelay(boolean tcpNoDelay) { |
||||
this.tcpNoDelay = tcpNoDelay; |
||||
} |
||||
|
||||
public boolean isSoKeepalive() { |
||||
return soKeepalive; |
||||
} |
||||
|
||||
public void setSoKeepalive(boolean soKeepalive) { |
||||
this.soKeepalive = soKeepalive; |
||||
} |
||||
|
||||
public int getSendBufferSize() { |
||||
return sendBufferSize; |
||||
} |
||||
|
||||
public void setSendBufferSize(int sendBufferSize) { |
||||
this.sendBufferSize = sendBufferSize; |
||||
} |
||||
|
||||
public int getReceiveBufferSize() { |
||||
return receiveBufferSize; |
||||
} |
||||
|
||||
public void setReceiveBufferSize(int receiveBufferSize) { |
||||
this.receiveBufferSize = receiveBufferSize; |
||||
} |
||||
|
||||
public int getWorkerThread() { |
||||
return workerThread; |
||||
} |
||||
|
||||
public void setWorkerThread(int workerThread) { |
||||
this.workerThread = workerThread; |
||||
} |
||||
} |
||||
@ -0,0 +1,94 @@
|
||||
/* |
||||
* Licensed to the Apache Software Foundation (ASF) under one or more |
||||
* contributor license agreements. See the NOTICE file distributed with |
||||
* this work for additional information regarding copyright ownership. |
||||
* The ASF licenses this file to You under the Apache License, Version 2.0 |
||||
* (the "License"); you may not use this file except in compliance with |
||||
* the License. You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
*/ |
||||
|
||||
package org.apache.dolphinscheduler.remote.exceptions; |
||||
|
||||
/** |
||||
* remote exception |
||||
*/ |
||||
public class RemotingException extends Exception { |
||||
|
||||
public RemotingException() { |
||||
super(); |
||||
} |
||||
|
||||
/** Constructs a new runtime exception with the specified detail message. |
||||
* The cause is not initialized, and may subsequently be initialized by a |
||||
* call to {@link #initCause}. |
||||
* |
||||
* @param message the detail message. The detail message is saved for |
||||
* later retrieval by the {@link #getMessage()} method. |
||||
*/ |
||||
public RemotingException(String message) { |
||||
super(message); |
||||
} |
||||
|
||||
/** |
||||
* Constructs a new runtime exception with the specified detail message and |
||||
* cause. <p>Note that the detail message associated with |
||||
* {@code cause} is <i>not</i> automatically incorporated in |
||||
* this runtime exception's detail message. |
||||
* |
||||
* @param message the detail message (which is saved for later retrieval |
||||
* by the {@link #getMessage()} method). |
||||
* @param cause the cause (which is saved for later retrieval by the |
||||
* {@link #getCause()} method). (A <tt>null</tt> value is |
||||
* permitted, and indicates that the cause is nonexistent or |
||||
* unknown.) |
||||
* @since 1.4 |
||||
*/ |
||||
public RemotingException(String message, Throwable cause) { |
||||
super(message, cause); |
||||
} |
||||
|
||||
/** Constructs a new runtime exception with the specified cause and a |
||||
* detail message of <tt>(cause==null ? null : cause.toString())</tt> |
||||
* (which typically contains the class and detail message of |
||||
* <tt>cause</tt>). This constructor is useful for runtime exceptions |
||||
* that are little more than wrappers for other throwables. |
||||
* |
||||
* @param cause the cause (which is saved for later retrieval by the |
||||
* {@link #getCause()} method). (A <tt>null</tt> value is |
||||
* permitted, and indicates that the cause is nonexistent or |
||||
* unknown.) |
||||
* @since 1.4 |
||||
*/ |
||||
public RemotingException(Throwable cause) { |
||||
super(cause); |
||||
} |
||||
|
||||
/** |
||||
* Constructs a new runtime exception with the specified detail |
||||
* message, cause, suppression enabled or disabled, and writable |
||||
* stack trace enabled or disabled. |
||||
* |
||||
* @param message the detail message. |
||||
* @param cause the cause. (A {@code null} value is permitted, |
||||
* and indicates that the cause is nonexistent or unknown.) |
||||
* @param enableSuppression whether or not suppression is enabled |
||||
* or disabled |
||||
* @param writableStackTrace whether or not the stack trace should |
||||
* be writable |
||||
* |
||||
* @since 1.7 |
||||
*/ |
||||
protected RemotingException(String message, Throwable cause, |
||||
boolean enableSuppression, |
||||
boolean writableStackTrace) { |
||||
super(message, cause, enableSuppression, writableStackTrace); |
||||
} |
||||
} |
||||
@ -0,0 +1,174 @@
|
||||
/* |
||||
* Licensed to the Apache Software Foundation (ASF) under one or more |
||||
* contributor license agreements. See the NOTICE file distributed with |
||||
* this work for additional information regarding copyright ownership. |
||||
* The ASF licenses this file to You under the Apache License, Version 2.0 |
||||
* (the "License"); you may not use this file except in compliance with |
||||
* the License. You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
*/ |
||||
package org.apache.dolphinscheduler.remote.handler; |
||||
|
||||
import io.netty.channel.*; |
||||
import org.apache.dolphinscheduler.remote.NettyRemotingClient; |
||||
import org.apache.dolphinscheduler.remote.command.Command; |
||||
import org.apache.dolphinscheduler.remote.command.CommandType; |
||||
import org.apache.dolphinscheduler.remote.processor.NettyRequestProcessor; |
||||
import org.apache.dolphinscheduler.remote.utils.ChannelUtils; |
||||
import org.apache.dolphinscheduler.remote.utils.Pair; |
||||
import org.slf4j.Logger; |
||||
import org.slf4j.LoggerFactory; |
||||
|
||||
import java.util.concurrent.ConcurrentHashMap; |
||||
import java.util.concurrent.ExecutorService; |
||||
import java.util.concurrent.RejectedExecutionException; |
||||
|
||||
/** |
||||
* netty client request handler |
||||
*/ |
||||
@ChannelHandler.Sharable |
||||
public class NettyClientHandler extends ChannelInboundHandlerAdapter { |
||||
|
||||
private final Logger logger = LoggerFactory.getLogger(NettyClientHandler.class); |
||||
|
||||
/** |
||||
* netty remote client |
||||
*/ |
||||
private final NettyRemotingClient nettyRemotingClient; |
||||
|
||||
/** |
||||
* client processors queue |
||||
*/ |
||||
private final ConcurrentHashMap<CommandType, Pair<NettyRequestProcessor, ExecutorService>> processors = new ConcurrentHashMap(); |
||||
|
||||
public NettyClientHandler(NettyRemotingClient nettyRemotingClient){ |
||||
this.nettyRemotingClient = nettyRemotingClient; |
||||
} |
||||
|
||||
/** |
||||
* When the current channel is not active, |
||||
* the current channel has reached the end of its life cycle |
||||
* |
||||
* @param ctx channel handler context |
||||
* @throws Exception |
||||
*/ |
||||
@Override |
||||
public void channelInactive(ChannelHandlerContext ctx) throws Exception { |
||||
nettyRemotingClient.removeChannel(ChannelUtils.toAddress(ctx.channel())); |
||||
ctx.channel().close(); |
||||
} |
||||
|
||||
/** |
||||
* The current channel reads data from the remote |
||||
* |
||||
* @param ctx channel handler context |
||||
* @param msg message |
||||
* @throws Exception |
||||
*/ |
||||
@Override |
||||
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { |
||||
processReceived(ctx.channel(), (Command)msg); |
||||
} |
||||
|
||||
/** |
||||
* register processor |
||||
* |
||||
* @param commandType command type |
||||
* @param processor processor |
||||
*/ |
||||
public void registerProcessor(final CommandType commandType, final NettyRequestProcessor processor) { |
||||
this.registerProcessor(commandType, processor, nettyRemotingClient.getDefaultExecutor()); |
||||
} |
||||
|
||||
/** |
||||
* register processor |
||||
* |
||||
* @param commandType command type |
||||
* @param processor processor |
||||
* @param executor thread executor |
||||
*/ |
||||
public void registerProcessor(final CommandType commandType, final NettyRequestProcessor processor, final ExecutorService executor) { |
||||
ExecutorService executorRef = executor; |
||||
if(executorRef == null){ |
||||
executorRef = nettyRemotingClient.getDefaultExecutor(); |
||||
} |
||||
this.processors.putIfAbsent(commandType, new Pair<NettyRequestProcessor, ExecutorService>(processor, executorRef)); |
||||
} |
||||
|
||||
/** |
||||
* process received logic |
||||
* |
||||
* @param channel channel |
||||
* @param msg message |
||||
*/ |
||||
private void processReceived(final Channel channel, final Command msg) { |
||||
final CommandType commandType = msg.getType(); |
||||
final Pair<NettyRequestProcessor, ExecutorService> pair = processors.get(commandType); |
||||
if (pair != null) { |
||||
Runnable r = new Runnable() { |
||||
@Override |
||||
public void run() { |
||||
try { |
||||
pair.getLeft().process(channel, msg); |
||||
} catch (Throwable ex) { |
||||
logger.error("process msg {} error : {}", msg, ex); |
||||
} |
||||
} |
||||
}; |
||||
try { |
||||
pair.getRight().submit(r); |
||||
} catch (RejectedExecutionException e) { |
||||
logger.warn("thread pool is full, discard msg {} from {}", msg, ChannelUtils.getRemoteAddress(channel)); |
||||
} |
||||
} else { |
||||
logger.warn("commandType {} not support", commandType); |
||||
} |
||||
} |
||||
|
||||
/** |
||||
* caught exception |
||||
* |
||||
* @param ctx channel handler context |
||||
* @param cause cause |
||||
* @throws Exception |
||||
*/ |
||||
@Override |
||||
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { |
||||
logger.error("exceptionCaught : {}", cause); |
||||
nettyRemotingClient.removeChannel(ChannelUtils.toAddress(ctx.channel())); |
||||
ctx.channel().close(); |
||||
} |
||||
|
||||
/** |
||||
* channel write changed |
||||
* @param ctx channel handler context |
||||
* @throws Exception |
||||
*/ |
||||
@Override |
||||
public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception { |
||||
Channel ch = ctx.channel(); |
||||
ChannelConfig config = ch.config(); |
||||
|
||||
if (!ch.isWritable()) { |
||||
if (logger.isWarnEnabled()) { |
||||
logger.warn("{} is not writable, over high water level : {}", |
||||
new Object[]{ch, config.getWriteBufferHighWaterMark()}); |
||||
} |
||||
|
||||
config.setAutoRead(false); |
||||
} else { |
||||
if (logger.isWarnEnabled()) { |
||||
logger.warn("{} is writable, to low water : {}", |
||||
new Object[]{ch, config.getWriteBufferLowWaterMark()}); |
||||
} |
||||
config.setAutoRead(true); |
||||
} |
||||
} |
||||
} |
||||
@ -0,0 +1,173 @@
|
||||
/* |
||||
* Licensed to the Apache Software Foundation (ASF) under one or more |
||||
* contributor license agreements. See the NOTICE file distributed with |
||||
* this work for additional information regarding copyright ownership. |
||||
* The ASF licenses this file to You under the Apache License, Version 2.0 |
||||
* (the "License"); you may not use this file except in compliance with |
||||
* the License. You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
*/ |
||||
|
||||
package org.apache.dolphinscheduler.remote.handler; |
||||
|
||||
import io.netty.channel.*; |
||||
import org.apache.dolphinscheduler.remote.NettyRemotingServer; |
||||
import org.apache.dolphinscheduler.remote.command.Command; |
||||
import org.apache.dolphinscheduler.remote.command.CommandType; |
||||
import org.apache.dolphinscheduler.remote.processor.NettyRequestProcessor; |
||||
import org.apache.dolphinscheduler.remote.utils.ChannelUtils; |
||||
import org.apache.dolphinscheduler.remote.utils.Pair; |
||||
import org.slf4j.Logger; |
||||
import org.slf4j.LoggerFactory; |
||||
|
||||
import java.util.concurrent.ConcurrentHashMap; |
||||
import java.util.concurrent.ExecutorService; |
||||
import java.util.concurrent.RejectedExecutionException; |
||||
|
||||
/** |
||||
* netty server request handler |
||||
*/ |
||||
@ChannelHandler.Sharable |
||||
public class NettyServerHandler extends ChannelInboundHandlerAdapter { |
||||
|
||||
private final Logger logger = LoggerFactory.getLogger(NettyServerHandler.class); |
||||
|
||||
/** |
||||
* netty remote server |
||||
*/ |
||||
private final NettyRemotingServer nettyRemotingServer; |
||||
|
||||
/** |
||||
* server processors queue |
||||
*/ |
||||
private final ConcurrentHashMap<CommandType, Pair<NettyRequestProcessor, ExecutorService>> processors = new ConcurrentHashMap(); |
||||
|
||||
public NettyServerHandler(NettyRemotingServer nettyRemotingServer){ |
||||
this.nettyRemotingServer = nettyRemotingServer; |
||||
} |
||||
|
||||
/** |
||||
* When the current channel is not active, |
||||
* the current channel has reached the end of its life cycle |
||||
* @param ctx channel handler context |
||||
* @throws Exception |
||||
*/ |
||||
@Override |
||||
public void channelInactive(ChannelHandlerContext ctx) throws Exception { |
||||
ctx.channel().close(); |
||||
} |
||||
|
||||
/** |
||||
* The current channel reads data from the remote end |
||||
* |
||||
* @param ctx channel handler context |
||||
* @param msg message |
||||
* @throws Exception |
||||
*/ |
||||
@Override |
||||
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { |
||||
processReceived(ctx.channel(), (Command)msg); |
||||
} |
||||
|
||||
/** |
||||
* register processor |
||||
* |
||||
* @param commandType command type |
||||
* @param processor processor |
||||
*/ |
||||
public void registerProcessor(final CommandType commandType, final NettyRequestProcessor processor) { |
||||
this.registerProcessor(commandType, processor, null); |
||||
} |
||||
|
||||
/** |
||||
* register processor |
||||
* |
||||
* @param commandType command type |
||||
* @param processor processor |
||||
* @param executor thread executor |
||||
*/ |
||||
public void registerProcessor(final CommandType commandType, final NettyRequestProcessor processor, final ExecutorService executor) { |
||||
ExecutorService executorRef = executor; |
||||
if(executorRef == null){ |
||||
executorRef = nettyRemotingServer.getDefaultExecutor(); |
||||
} |
||||
this.processors.putIfAbsent(commandType, new Pair<NettyRequestProcessor, ExecutorService>(processor, executorRef)); |
||||
} |
||||
|
||||
/** |
||||
* process received logic |
||||
* @param channel channel |
||||
* @param msg message |
||||
*/ |
||||
private void processReceived(final Channel channel, final Command msg) { |
||||
final CommandType commandType = msg.getType(); |
||||
final Pair<NettyRequestProcessor, ExecutorService> pair = processors.get(commandType); |
||||
if (pair != null) { |
||||
Runnable r = new Runnable() { |
||||
|
||||
@Override |
||||
public void run() { |
||||
try { |
||||
pair.getLeft().process(channel, msg); |
||||
} catch (Throwable ex) { |
||||
logger.error("process msg {} error : {}", msg, ex); |
||||
} |
||||
} |
||||
}; |
||||
try { |
||||
pair.getRight().submit(r); |
||||
} catch (RejectedExecutionException e) { |
||||
logger.warn("thread pool is full, discard msg {} from {}", msg, ChannelUtils.getRemoteAddress(channel)); |
||||
} |
||||
} else { |
||||
logger.warn("commandType {} not support", commandType); |
||||
} |
||||
} |
||||
|
||||
/** |
||||
* caught exception |
||||
* |
||||
* @param ctx channel handler context |
||||
* @param cause cause |
||||
* @throws Exception |
||||
*/ |
||||
@Override |
||||
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { |
||||
logger.error("exceptionCaught : {}", cause); |
||||
ctx.channel().close(); |
||||
} |
||||
|
||||
/** |
||||
* channel write changed |
||||
* |
||||
* @param ctx channel handler context |
||||
* @throws Exception |
||||
*/ |
||||
@Override |
||||
public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception { |
||||
Channel ch = ctx.channel(); |
||||
ChannelConfig config = ch.config(); |
||||
|
||||
if (!ch.isWritable()) { |
||||
if (logger.isWarnEnabled()) { |
||||
logger.warn("{} is not writable, over high water level : {}", |
||||
new Object[]{ch, config.getWriteBufferHighWaterMark()}); |
||||
} |
||||
|
||||
config.setAutoRead(false); |
||||
} else { |
||||
if (logger.isWarnEnabled()) { |
||||
logger.warn("{} is writable, to low water : {}", |
||||
new Object[]{ch, config.getWriteBufferLowWaterMark()}); |
||||
} |
||||
config.setAutoRead(true); |
||||
} |
||||
} |
||||
} |
||||
@ -0,0 +1,96 @@
|
||||
/* |
||||
* Licensed to the Apache Software Foundation (ASF) under one or more |
||||
* contributor license agreements. See the NOTICE file distributed with |
||||
* this work for additional information regarding copyright ownership. |
||||
* The ASF licenses this file to You under the Apache License, Version 2.0 |
||||
* (the "License"); you may not use this file except in compliance with |
||||
* the License. You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
*/ |
||||
package org.apache.dolphinscheduler.remote.utils; |
||||
|
||||
import java.io.Serializable; |
||||
|
||||
/** |
||||
* server address |
||||
*/ |
||||
public class Address implements Serializable { |
||||
|
||||
/** |
||||
* host |
||||
*/ |
||||
private String host; |
||||
|
||||
/** |
||||
* port |
||||
*/ |
||||
private int port; |
||||
|
||||
public Address(){ |
||||
//NOP
|
||||
} |
||||
|
||||
public Address(String host, int port){ |
||||
this.host = host; |
||||
this.port = port; |
||||
} |
||||
|
||||
public String getHost() { |
||||
return host; |
||||
} |
||||
|
||||
public void setHost(String host) { |
||||
this.host = host; |
||||
} |
||||
|
||||
public int getPort() { |
||||
return port; |
||||
} |
||||
|
||||
public void setPort(int port) { |
||||
this.port = port; |
||||
} |
||||
|
||||
@Override |
||||
public int hashCode() { |
||||
final int prime = 31; |
||||
int result = 1; |
||||
result = prime * result + ((host == null) ? 0 : host.hashCode()); |
||||
result = prime * result + port; |
||||
return result; |
||||
} |
||||
|
||||
@Override |
||||
public boolean equals(Object obj) { |
||||
if (this == obj) { |
||||
return true; |
||||
} |
||||
if (obj == null) { |
||||
return false; |
||||
} |
||||
if (getClass() != obj.getClass()) { |
||||
return false; |
||||
} |
||||
Address other = (Address) obj; |
||||
if (host == null) { |
||||
if (other.host != null) { |
||||
return false; |
||||
} |
||||
} else if (!host.equals(other.host)) { |
||||
return false; |
||||
} |
||||
return port == other.port; |
||||
} |
||||
|
||||
@Override |
||||
public String toString() { |
||||
return "Address [host=" + host + ", port=" + port + "]"; |
||||
} |
||||
} |
||||
@ -0,0 +1,57 @@
|
||||
/* |
||||
* Licensed to the Apache Software Foundation (ASF) under one or more |
||||
* contributor license agreements. See the NOTICE file distributed with |
||||
* this work for additional information regarding copyright ownership. |
||||
* The ASF licenses this file to You under the Apache License, Version 2.0 |
||||
* (the "License"); you may not use this file except in compliance with |
||||
* the License. You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
*/ |
||||
package org.apache.dolphinscheduler.remote.utils; |
||||
|
||||
import io.netty.channel.Channel; |
||||
|
||||
import java.net.InetSocketAddress; |
||||
|
||||
/** |
||||
* channel utils |
||||
*/ |
||||
public class ChannelUtils { |
||||
|
||||
/** |
||||
* get local address |
||||
* |
||||
* @param channel channel |
||||
* @return local address |
||||
*/ |
||||
public static String getLocalAddress(Channel channel){ |
||||
return ((InetSocketAddress)channel.localAddress()).getAddress().getHostAddress(); |
||||
} |
||||
|
||||
/** |
||||
* get remote address |
||||
* @param channel channel |
||||
* @return remote address |
||||
*/ |
||||
public static String getRemoteAddress(Channel channel){ |
||||
return ((InetSocketAddress)channel.remoteAddress()).getAddress().getHostAddress(); |
||||
} |
||||
|
||||
/** |
||||
* channel to address |
||||
* @param channel channel |
||||
* @return address |
||||
*/ |
||||
public static Address toAddress(Channel channel){ |
||||
InetSocketAddress socketAddress = ((InetSocketAddress)channel.remoteAddress()); |
||||
return new Address(socketAddress.getAddress().getHostAddress(), socketAddress.getPort()); |
||||
} |
||||
|
||||
} |
||||
@ -0,0 +1,60 @@
|
||||
/* |
||||
* Licensed to the Apache Software Foundation (ASF) under one or more |
||||
* contributor license agreements. See the NOTICE file distributed with |
||||
* this work for additional information regarding copyright ownership. |
||||
* The ASF licenses this file to You under the Apache License, Version 2.0 |
||||
* (the "License"); you may not use this file except in compliance with |
||||
* the License. You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
*/ |
||||
package org.apache.dolphinscheduler.remote.utils; |
||||
|
||||
import com.alibaba.fastjson.JSON; |
||||
|
||||
/** |
||||
* json serialize or deserialize |
||||
*/ |
||||
public class FastJsonSerializer { |
||||
|
||||
/** |
||||
* serialize to byte |
||||
* |
||||
* @param obj object |
||||
* @param <T> object type |
||||
* @return byte array |
||||
*/ |
||||
public static <T> byte[] serialize(T obj) { |
||||
String json = JSON.toJSONString(obj); |
||||
return json.getBytes(Constants.UTF8); |
||||
} |
||||
|
||||
/** |
||||
* serialize to string |
||||
* @param obj object |
||||
* @param <T> object type |
||||
* @return string |
||||
*/ |
||||
public static <T> String serializeToString(T obj) { |
||||
return JSON.toJSONString(obj); |
||||
} |
||||
|
||||
/** |
||||
* deserialize |
||||
* |
||||
* @param src byte array |
||||
* @param clazz class
|
||||
* @param <T> deserialize type |
||||
* @return deserialize type |
||||
*/ |
||||
public static <T> T deserialize(byte[] src, Class<T> clazz) { |
||||
return JSON.parseObject(new String(src, Constants.UTF8), clazz); |
||||
} |
||||
|
||||
} |
||||
@ -0,0 +1,53 @@
|
||||
/* |
||||
* Licensed to the Apache Software Foundation (ASF) under one or more |
||||
* contributor license agreements. See the NOTICE file distributed with |
||||
* this work for additional information regarding copyright ownership. |
||||
* The ASF licenses this file to You under the Apache License, Version 2.0 |
||||
* (the "License"); you may not use this file except in compliance with |
||||
* the License. You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
*/ |
||||
|
||||
package org.apache.dolphinscheduler.remote.utils; |
||||
|
||||
|
||||
/** |
||||
* key value pair |
||||
* |
||||
* @param <L> L generic type |
||||
* @param <R> R generic type |
||||
*/ |
||||
public class Pair<L, R> { |
||||
|
||||
private L left; |
||||
|
||||
private R right; |
||||
|
||||
public Pair(L left, R right) { |
||||
this.left = left; |
||||
this.right = right; |
||||
} |
||||
|
||||
public L getLeft() { |
||||
return left; |
||||
} |
||||
|
||||
public void setLeft(L left) { |
||||
this.left = left; |
||||
} |
||||
|
||||
public R getRight() { |
||||
return right; |
||||
} |
||||
|
||||
public void setRight(R right) { |
||||
this.right = right; |
||||
} |
||||
} |
||||
@ -0,0 +1,77 @@
|
||||
package org.apache.dolphinscheduler.remote;/* |
||||
* Licensed to the Apache Software Foundation (ASF) under one or more |
||||
* contributor license agreements. See the NOTICE file distributed with |
||||
* this work for additional information regarding copyright ownership. |
||||
* The ASF licenses this file to You under the Apache License, Version 2.0 |
||||
* (the "License"); you may not use this file except in compliance with |
||||
* the License. You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
*/ |
||||
|
||||
import io.netty.channel.Channel; |
||||
import org.apache.dolphinscheduler.remote.NettyRemotingClient; |
||||
import org.apache.dolphinscheduler.remote.NettyRemotingServer; |
||||
import org.apache.dolphinscheduler.remote.command.Command; |
||||
import org.apache.dolphinscheduler.remote.command.CommandType; |
||||
import org.apache.dolphinscheduler.remote.command.Ping; |
||||
import org.apache.dolphinscheduler.remote.command.Pong; |
||||
import org.apache.dolphinscheduler.remote.config.NettyClientConfig; |
||||
import org.apache.dolphinscheduler.remote.config.NettyServerConfig; |
||||
import org.apache.dolphinscheduler.remote.processor.NettyRequestProcessor; |
||||
import org.apache.dolphinscheduler.remote.utils.Address; |
||||
import org.junit.Assert; |
||||
import org.junit.Test; |
||||
|
||||
import java.util.concurrent.CountDownLatch; |
||||
import java.util.concurrent.atomic.AtomicLong; |
||||
|
||||
/** |
||||
* netty remote client test |
||||
*/ |
||||
public class NettyRemotingClientTest { |
||||
|
||||
|
||||
/** |
||||
* test ping |
||||
*/ |
||||
@Test |
||||
public void testSend(){ |
||||
NettyServerConfig serverConfig = new NettyServerConfig(); |
||||
|
||||
NettyRemotingServer server = new NettyRemotingServer(serverConfig); |
||||
server.registerProcessor(CommandType.PING, new NettyRequestProcessor() { |
||||
@Override |
||||
public void process(Channel channel, Command command) { |
||||
channel.writeAndFlush(Pong.create(command.getOpaque())); |
||||
} |
||||
}); |
||||
server.start(); |
||||
//
|
||||
CountDownLatch latch = new CountDownLatch(1); |
||||
AtomicLong opaque = new AtomicLong(1); |
||||
final NettyClientConfig clientConfig = new NettyClientConfig(); |
||||
NettyRemotingClient client = new NettyRemotingClient(clientConfig); |
||||
client.registerProcessor(CommandType.PONG, new NettyRequestProcessor() { |
||||
@Override |
||||
public void process(Channel channel, Command command) { |
||||
opaque.set(command.getOpaque()); |
||||
latch.countDown(); |
||||
} |
||||
}); |
||||
Command commandPing = Ping.create(); |
||||
try { |
||||
client.send(new Address("127.0.0.1", serverConfig.getListenPort()), commandPing); |
||||
latch.await(); |
||||
} catch (Exception e) { |
||||
e.printStackTrace(); |
||||
} |
||||
Assert.assertEquals(opaque.get(), commandPing.getOpaque()); |
||||
} |
||||
} |
||||
@ -1,113 +0,0 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?> |
||||
<!-- |
||||
~ Licensed to the Apache Software Foundation (ASF) under one or more |
||||
~ contributor license agreements. See the NOTICE file distributed with |
||||
~ this work for additional information regarding copyright ownership. |
||||
~ The ASF licenses this file to You under the Apache License, Version 2.0 |
||||
~ (the "License"); you may not use this file except in compliance with |
||||
~ the License. You may obtain a copy of the License at |
||||
~ |
||||
~ http://www.apache.org/licenses/LICENSE-2.0 |
||||
~ |
||||
~ Unless required by applicable law or agreed to in writing, software |
||||
~ distributed under the License is distributed on an "AS IS" BASIS, |
||||
~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
~ See the License for the specific language governing permissions and |
||||
~ limitations under the License. |
||||
--> |
||||
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> |
||||
<parent> |
||||
<groupId>org.apache.dolphinscheduler</groupId> |
||||
<artifactId>dolphinscheduler</artifactId> |
||||
<version>1.2.1-SNAPSHOT</version> |
||||
</parent> |
||||
<modelVersion>4.0.0</modelVersion> |
||||
|
||||
<artifactId>dolphinscheduler-rpc</artifactId> |
||||
|
||||
<name>dolphinscheduler-rpc</name> |
||||
<url>https://github.com/apache/incubator-dolphinscheduler</url> |
||||
|
||||
<properties> |
||||
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> |
||||
<maven.compiler.source>1.8</maven.compiler.source> |
||||
<maven.compiler.target>1.8</maven.compiler.target> |
||||
|
||||
<protobuf.version>3.5.1</protobuf.version> |
||||
<grpc.version>1.9.0</grpc.version> |
||||
</properties> |
||||
|
||||
<dependencies> |
||||
<dependency> |
||||
<groupId>com.google.protobuf</groupId> |
||||
<artifactId>protobuf-java</artifactId> |
||||
<version>${protobuf.version}</version> |
||||
</dependency> |
||||
<dependency> |
||||
<groupId>io.grpc</groupId> |
||||
<artifactId>grpc-netty</artifactId> |
||||
<version>${grpc.version}</version> |
||||
</dependency> |
||||
<dependency> |
||||
<groupId>io.grpc</groupId> |
||||
<artifactId>grpc-protobuf</artifactId> |
||||
<version>${grpc.version}</version> |
||||
</dependency> |
||||
<dependency> |
||||
<groupId>io.grpc</groupId> |
||||
<artifactId>grpc-stub</artifactId> |
||||
<version>${grpc.version}</version> |
||||
</dependency> |
||||
|
||||
<dependency> |
||||
<groupId>com.google.guava</groupId> |
||||
<artifactId>guava</artifactId> |
||||
</dependency> |
||||
</dependencies> |
||||
|
||||
<build> |
||||
<extensions> |
||||
<extension> |
||||
<groupId>kr.motd.maven</groupId> |
||||
<artifactId>os-maven-plugin</artifactId> |
||||
<version>1.5.0.Final</version> |
||||
</extension> |
||||
</extensions> |
||||
<plugins> |
||||
<plugin> |
||||
<groupId>org.xolstice.maven.plugins</groupId> |
||||
<artifactId>protobuf-maven-plugin</artifactId> |
||||
<version>0.5.0</version> |
||||
<configuration> |
||||
<protocArtifact>com.google.protobuf:protoc:3.5.1-1:exe:${os.detected.classifier}</protocArtifact> |
||||
<pluginId>grpc-java</pluginId> |
||||
<pluginArtifact>io.grpc:protoc-gen-grpc-java:${grpc.version}:exe:${os.detected.classifier}</pluginArtifact> |
||||
</configuration> |
||||
<executions> |
||||
<execution> |
||||
<id>compile</id> |
||||
<goals> |
||||
<goal>compile</goal> |
||||
</goals> |
||||
</execution> |
||||
<execution> |
||||
<id>compile-custom</id> |
||||
<goals> |
||||
<goal>compile-custom</goal> |
||||
</goals> |
||||
</execution> |
||||
</executions> |
||||
</plugin> |
||||
<plugin> |
||||
<groupId>org.apache.maven.plugins</groupId> |
||||
<artifactId>maven-compiler-plugin</artifactId> |
||||
<configuration> |
||||
<source>${java.version}</source> |
||||
<target>${java.version}</target> |
||||
<encoding>${project.build.sourceEncoding}</encoding> |
||||
</configuration> |
||||
</plugin> |
||||
</plugins> |
||||
</build> |
||||
</project> |
||||
@ -1,101 +0,0 @@
|
||||
/* |
||||
* Licensed to the Apache Software Foundation (ASF) under one or more |
||||
* contributor license agreements. See the NOTICE file distributed with |
||||
* this work for additional information regarding copyright ownership. |
||||
* The ASF licenses this file to You under the Apache License, Version 2.0 |
||||
* (the "License"); you may not use this file except in compliance with |
||||
* the License. You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0 |
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
* |
||||
*/ |
||||
|
||||
syntax = "proto3"; |
||||
|
||||
package schduler; |
||||
|
||||
option java_multiple_files = true; |
||||
option java_package = "org.apache.dolphinscheduler.rpc"; |
||||
option java_outer_classname = "SchdulerProto"; |
||||
|
||||
|
||||
/** |
||||
* return str info |
||||
*/ |
||||
message RetStrInfo { |
||||
/** |
||||
* str msg info |
||||
*/ |
||||
string msg = 1 ; |
||||
} |
||||
|
||||
/** |
||||
* return byte info |
||||
*/ |
||||
message RetByteInfo { |
||||
/** |
||||
* byte data info |
||||
*/ |
||||
bytes data = 1; |
||||
} |
||||
|
||||
/** |
||||
* log parameter |
||||
*/ |
||||
message LogParameter { |
||||
|
||||
/** |
||||
* path |
||||
*/ |
||||
string path = 1 ; |
||||
|
||||
/** |
||||
* skip line num |
||||
*/ |
||||
int32 skipLineNum = 2 ; |
||||
|
||||
/** |
||||
* display limt num |
||||
*/ |
||||
int32 limit = 3 ; |
||||
} |
||||
|
||||
|
||||
/** |
||||
* path parameter |
||||
*/ |
||||
message PathParameter { |
||||
|
||||
/** |
||||
* path |
||||
*/ |
||||
string path = 1 ; |
||||
} |
||||
|
||||
/** |
||||
* log view service |
||||
*/ |
||||
service LogViewService { |
||||
|
||||
/** |
||||
* roll view log |
||||
*/ |
||||
rpc rollViewLog(LogParameter) returns (RetStrInfo) {}; |
||||
|
||||
/** |
||||
* view all log |
||||
*/ |
||||
rpc viewLog(PathParameter) returns (RetStrInfo) {}; |
||||
|
||||
/** |
||||
* get log bytes |
||||
*/ |
||||
rpc getLogBytes(PathParameter) returns (RetByteInfo) {}; |
||||
} |
||||
|
||||
@ -0,0 +1,179 @@
|
||||
/* |
||||
* Licensed to the Apache Software Foundation (ASF) under one or more |
||||
* contributor license agreements. See the NOTICE file distributed with |
||||
* this work for additional information regarding copyright ownership. |
||||
* The ASF licenses this file to You under the Apache License, Version 2.0 |
||||
* (the "License"); you may not use this file except in compliance with |
||||
* the License. You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
*/ |
||||
package org.apache.dolphinscheduler.server.log; |
||||
|
||||
import io.netty.channel.Channel; |
||||
import org.apache.dolphinscheduler.remote.command.Command; |
||||
import org.apache.dolphinscheduler.remote.command.CommandType; |
||||
import org.apache.dolphinscheduler.remote.command.log.*; |
||||
import org.apache.dolphinscheduler.remote.processor.NettyRequestProcessor; |
||||
import org.apache.dolphinscheduler.remote.utils.FastJsonSerializer; |
||||
import org.slf4j.Logger; |
||||
import org.slf4j.LoggerFactory; |
||||
|
||||
import java.io.*; |
||||
import java.nio.file.Files; |
||||
import java.nio.file.Paths; |
||||
import java.util.Collections; |
||||
import java.util.List; |
||||
import java.util.concurrent.ExecutorService; |
||||
import java.util.concurrent.LinkedBlockingQueue; |
||||
import java.util.concurrent.ThreadPoolExecutor; |
||||
import java.util.concurrent.TimeUnit; |
||||
import java.util.stream.Collectors; |
||||
import java.util.stream.Stream; |
||||
|
||||
/** |
||||
* logger request process logic |
||||
*/ |
||||
public class LoggerRequestProcessor implements NettyRequestProcessor { |
||||
|
||||
private final Logger logger = LoggerFactory.getLogger(LoggerRequestProcessor.class); |
||||
|
||||
private final ThreadPoolExecutor executor; |
||||
|
||||
public LoggerRequestProcessor(){ |
||||
this.executor = new ThreadPoolExecutor(4, 4, 10, TimeUnit.SECONDS, new LinkedBlockingQueue<>(100)); |
||||
} |
||||
|
||||
@Override |
||||
public void process(Channel channel, Command command) { |
||||
logger.info("received command : {}", command); |
||||
|
||||
/** |
||||
* reuqest task log command type |
||||
*/ |
||||
final CommandType commandType = command.getType(); |
||||
switch (commandType){ |
||||
case GET_LOG_BYTES_REQUEST: |
||||
GetLogBytesRequestCommand getLogRequest = FastJsonSerializer.deserialize( |
||||
command.getBody(), GetLogBytesRequestCommand.class); |
||||
byte[] bytes = getFileContentBytes(getLogRequest.getPath()); |
||||
GetLogBytesResponseCommand getLogResponse = new GetLogBytesResponseCommand(bytes); |
||||
channel.writeAndFlush(getLogResponse.convert2Command(command.getOpaque())); |
||||
break; |
||||
case VIEW_WHOLE_LOG_REQUEST: |
||||
ViewLogRequestCommand viewLogRequest = FastJsonSerializer.deserialize( |
||||
command.getBody(), ViewLogRequestCommand.class); |
||||
String msg = readWholeFileContent(viewLogRequest.getPath()); |
||||
ViewLogResponseCommand viewLogResponse = new ViewLogResponseCommand(msg); |
||||
channel.writeAndFlush(viewLogResponse.convert2Command(command.getOpaque())); |
||||
break; |
||||
case ROLL_VIEW_LOG_REQUEST: |
||||
RollViewLogRequestCommand rollViewLogRequest = FastJsonSerializer.deserialize( |
||||
command.getBody(), RollViewLogRequestCommand.class); |
||||
List<String> lines = readPartFileContent(rollViewLogRequest.getPath(), |
||||
rollViewLogRequest.getSkipLineNum(), rollViewLogRequest.getLimit()); |
||||
StringBuilder builder = new StringBuilder(); |
||||
for (String line : lines){ |
||||
builder.append(line + "\r\n"); |
||||
} |
||||
RollViewLogResponseCommand rollViewLogRequestResponse = new RollViewLogResponseCommand(builder.toString()); |
||||
channel.writeAndFlush(rollViewLogRequestResponse.convert2Command(command.getOpaque())); |
||||
break; |
||||
default: |
||||
throw new IllegalArgumentException("unknown commandType"); |
||||
} |
||||
} |
||||
|
||||
public ExecutorService getExecutor(){ |
||||
return this.executor; |
||||
} |
||||
|
||||
/** |
||||
* get files content bytes，for down load file |
||||
* |
||||
* @param filePath file path |
||||
* @return byte array of file |
||||
* @throws Exception exception |
||||
*/ |
||||
private byte[] getFileContentBytes(String filePath){ |
||||
InputStream in = null; |
||||
ByteArrayOutputStream bos = null; |
||||
try { |
||||
in = new FileInputStream(filePath); |
||||
bos = new ByteArrayOutputStream(); |
||||
byte[] buf = new byte[1024]; |
||||
int len; |
||||
while ((len = in.read(buf)) != -1) { |
||||
bos.write(buf, 0, len); |
||||
} |
||||
return bos.toByteArray(); |
||||
}catch (IOException e){ |
||||
logger.error("get file bytes error",e); |
||||
}finally { |
||||
if (bos != null){ |
||||
try { |
||||
bos.close(); |
||||
} catch (IOException ignore) {} |
||||
} |
||||
if (in != null){ |
||||
try { |
||||
in.close(); |
||||
} catch (IOException ignore) {} |
||||
} |
||||
} |
||||
return new byte[0]; |
||||
} |
||||
|
||||
/** |
||||
* read part file content，can skip any line and read some lines |
||||
* |
||||
* @param filePath file path |
||||
* @param skipLine skip line |
||||
* @param limit read lines limit |
||||
* @return part file content |
||||
*/ |
||||
private List<String> readPartFileContent(String filePath, |
||||
int skipLine, |
||||
int limit){ |
||||
try (Stream<String> stream = Files.lines(Paths.get(filePath))) { |
||||
return stream.skip(skipLine).limit(limit).collect(Collectors.toList()); |
||||
} catch (IOException e) { |
||||
logger.error("read file error",e); |
||||
} |
||||
return Collections.EMPTY_LIST; |
||||
} |
||||
|
||||
/** |
||||
* read whole file content |
||||
* |
||||
* @param filePath file path |
||||
* @return whole file content |
||||
*/ |
||||
private String readWholeFileContent(String filePath){ |
||||
BufferedReader br = null; |
||||
String line; |
||||
StringBuilder sb = new StringBuilder(); |
||||
try { |
||||
br = new BufferedReader(new InputStreamReader(new FileInputStream(filePath))); |
||||
while ((line = br.readLine()) != null){ |
||||
sb.append(line + "\r\n"); |
||||
} |
||||
return sb.toString(); |
||||
}catch (IOException e){ |
||||
logger.error("read file error",e); |
||||
}finally { |
||||
try { |
||||
if (br != null){ |
||||
br.close(); |
||||
} |
||||
} catch (IOException ignore) {} |
||||
} |
||||
return ""; |
||||
} |
||||
} |
||||
@ -0,0 +1,91 @@
|
||||
/* |
||||
* Licensed to the Apache Software Foundation (ASF) under one or more |
||||
* contributor license agreements. See the NOTICE file distributed with |
||||
* this work for additional information regarding copyright ownership. |
||||
* The ASF licenses this file to You under the Apache License, Version 2.0 |
||||
* (the "License"); you may not use this file except in compliance with |
||||
* the License. You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
*/ |
||||
|
||||
package org.apache.dolphinscheduler.server.log; |
||||
|
||||
|
||||
import org.apache.dolphinscheduler.common.Constants; |
||||
import org.apache.dolphinscheduler.remote.NettyRemotingServer; |
||||
import org.apache.dolphinscheduler.remote.command.CommandType; |
||||
import org.apache.dolphinscheduler.remote.config.NettyServerConfig; |
||||
import org.slf4j.Logger; |
||||
import org.slf4j.LoggerFactory; |
||||
|
||||
/** |
||||
* logger server |
||||
*/ |
||||
public class LoggerServer { |
||||
|
||||
private static final Logger logger = LoggerFactory.getLogger(LoggerServer.class); |
||||
|
||||
/** |
||||
* netty server |
||||
*/ |
||||
private final NettyRemotingServer server; |
||||
|
||||
/** |
||||
* netty server config |
||||
*/ |
||||
private final NettyServerConfig serverConfig; |
||||
|
||||
/** |
||||
* loggger request processor |
||||
*/ |
||||
private final LoggerRequestProcessor requestProcessor; |
||||
|
||||
public LoggerServer(){ |
||||
this.serverConfig = new NettyServerConfig(); |
||||
this.serverConfig.setListenPort(Constants.RPC_PORT); |
||||
this.server = new NettyRemotingServer(serverConfig); |
||||
this.requestProcessor = new LoggerRequestProcessor(); |
||||
this.server.registerProcessor(CommandType.GET_LOG_BYTES_REQUEST, requestProcessor, requestProcessor.getExecutor()); |
||||
this.server.registerProcessor(CommandType.ROLL_VIEW_LOG_REQUEST, requestProcessor, requestProcessor.getExecutor()); |
||||
this.server.registerProcessor(CommandType.VIEW_WHOLE_LOG_REQUEST, requestProcessor, requestProcessor.getExecutor()); |
||||
} |
||||
|
||||
/** |
||||
* main launches the server from the command line. |
||||
* @param args arguments |
||||
*/ |
||||
public static void main(String[] args) { |
||||
final LoggerServer server = new LoggerServer(); |
||||
server.start(); |
||||
} |
||||
|
||||
/** |
||||
* server start |
||||
*/ |
||||
public void start() { |
||||
this.server.start(); |
||||
logger.info("logger server started, listening on port : {}" , Constants.RPC_PORT); |
||||
Runtime.getRuntime().addShutdownHook(new Thread() { |
||||
@Override |
||||
public void run() { |
||||
LoggerServer.this.stop(); |
||||
} |
||||
}); |
||||
} |
||||
|
||||
/** |
||||
* stop |
||||
*/ |
||||
public void stop() { |
||||
this.server.close(); |
||||
logger.info("logger server shut down"); |
||||
} |
||||
|
||||
} |
||||
Some files were not shown because too many files have changed in this diff Show More
Loading…
Reference in new issue