You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@asterixdb.apache.org by mh...@apache.org on 2018/11/12 20:57:54 UTC
asterixdb git commit: [ASTERIXDB-2479][API] Add Network Diagnostics
API
Repository: asterixdb
Updated Branches:
refs/heads/master 6eff367d9 -> 1deff42b3
[ASTERIXDB-2479][API] Add Network Diagnostics API
- user model changes: no
- storage format changes: no
- interface changes: no
Details:
- Add a new API that shows the state of node to node
connections and their logical channels to help diagnose
networking issues.
- Add channel details to the waiting thread name in
NetworkOutputChannel.
- Add test case.
Change-Id: Id6fd5a96c56e7078d1404bebcbab8afe93ba8f64
Reviewed-on: https://asterix-gerrit.ics.uci.edu/3025
Sonar-Qube: Jenkins <je...@fulliautomatix.ics.uci.edu>
Tested-by: Jenkins <je...@fulliautomatix.ics.uci.edu>
Integration-Tests: Jenkins <je...@fulliautomatix.ics.uci.edu>
Contrib: Jenkins <je...@fulliautomatix.ics.uci.edu>
Reviewed-by: Murtadha Hubail <mh...@apache.org>
Reviewed-by: Michael Blow <mb...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/asterixdb/repo
Commit: http://git-wip-us.apache.org/repos/asf/asterixdb/commit/1deff42b
Tree: http://git-wip-us.apache.org/repos/asf/asterixdb/tree/1deff42b
Diff: http://git-wip-us.apache.org/repos/asf/asterixdb/diff/1deff42b
Branch: refs/heads/master
Commit: 1deff42b33066acc4c1d8afcd66a3791be0cec02
Parents: 6eff367
Author: Murtadha Hubail <mh...@apache.org>
Authored: Mon Nov 12 16:10:35 2018 +0300
Committer: Murtadha Hubail <mh...@apache.org>
Committed: Mon Nov 12 12:57:25 2018 -0800
----------------------------------------------------------------------
.../http/server/NetDiagnosticsApiServlet.java | 63 ++++++++++++++++++++
.../hyracks/bootstrap/NCApplication.java | 3 +
.../asterix/test/runtime/APIExecutionTest.java | 19 +++++-
.../src/test/resources/runtimets/api.xml | 5 ++
.../net-diagnostics/net-diagnostics.1.get.http | 19 ++++++
.../net-diagnostics/net-diagnostics.1.regexadm | 5 ++
.../apache/asterix/common/utils/Servlets.java | 1 +
.../comm/channels/NetworkOutputChannel.java | 4 +-
.../hyracks/control/nc/net/NetworkManager.java | 4 ++
hyracks-fullstack/hyracks/hyracks-net/pom.xml | 4 ++
.../protocols/muxdemux/ChannelControlBlock.java | 17 ++++++
.../net/protocols/muxdemux/ChannelSet.java | 13 ++++
.../muxdemux/MultiplexedConnection.java | 17 ++++++
.../net/protocols/muxdemux/MuxDemux.java | 42 ++++++++++---
.../java/org/apache/hyracks/util/JSONUtil.java | 10 ++++
15 files changed, 215 insertions(+), 11 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/1deff42b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/NetDiagnosticsApiServlet.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/NetDiagnosticsApiServlet.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/NetDiagnosticsApiServlet.java
new file mode 100644
index 0000000..badb568
--- /dev/null
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/api/http/server/NetDiagnosticsApiServlet.java
@@ -0,0 +1,63 @@
+/*
+ * 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.asterix.api.http.server;
+
+import java.io.IOException;
+import java.io.PrintWriter;
+import java.util.concurrent.ConcurrentMap;
+
+import org.apache.asterix.common.api.INcApplicationContext;
+import org.apache.hyracks.control.nc.NodeControllerService;
+import org.apache.hyracks.control.nc.application.NCServiceContext;
+import org.apache.hyracks.control.nc.net.NetworkManager;
+import org.apache.hyracks.http.api.IServletRequest;
+import org.apache.hyracks.http.api.IServletResponse;
+import org.apache.hyracks.http.server.AbstractServlet;
+import org.apache.hyracks.http.server.utils.HttpUtil;
+import org.apache.hyracks.util.JSONUtil;
+
+import com.fasterxml.jackson.databind.JsonNode;
+
+import io.netty.handler.codec.http.HttpResponseStatus;
+
+public class NetDiagnosticsApiServlet extends AbstractServlet {
+
+ private final INcApplicationContext appCtx;
+
+ public NetDiagnosticsApiServlet(ConcurrentMap<String, Object> ctx, INcApplicationContext appCtx, String... paths) {
+ super(ctx, paths);
+ this.appCtx = appCtx;
+ }
+
+ @Override
+ protected void get(IServletRequest request, IServletResponse response) throws IOException {
+ HttpUtil.setContentType(response, HttpUtil.ContentType.APPLICATION_JSON, HttpUtil.Encoding.UTF8);
+ response.setStatus(HttpResponseStatus.OK);
+ final JsonNode netDiagnostics = getNetDiagnostics();
+ final PrintWriter responseWriter = response.writer();
+ JSONUtil.writeNode(responseWriter, netDiagnostics);
+ }
+
+ private JsonNode getNetDiagnostics() {
+ final NCServiceContext serviceContext = (NCServiceContext) appCtx.getServiceContext();
+ final NodeControllerService controllerService = (NodeControllerService) serviceContext.getControllerService();
+ final NetworkManager networkManager = controllerService.getNetworkManager();
+ return networkManager.getMuxDemux().getState();
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/1deff42b/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/NCApplication.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/NCApplication.java b/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/NCApplication.java
index fbafc2e..4fa86ae 100644
--- a/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/NCApplication.java
+++ b/asterixdb/asterix-app/src/main/java/org/apache/asterix/hyracks/bootstrap/NCApplication.java
@@ -24,6 +24,7 @@ import java.util.Collections;
import java.util.List;
import java.util.Set;
+import org.apache.asterix.api.http.server.NetDiagnosticsApiServlet;
import org.apache.asterix.api.http.server.ServletConstants;
import org.apache.asterix.api.http.server.StorageApiServlet;
import org.apache.asterix.app.io.PersistedResourceRegistry;
@@ -170,6 +171,8 @@ public class NCApplication extends BaseNCApplication {
externalProperties.getNcApiPort(), config);
apiServer.setAttribute(ServletConstants.SERVICE_CONTEXT_ATTR, ncServiceCtx);
apiServer.addServlet(new StorageApiServlet(apiServer.ctx(), getApplicationContext(), Servlets.STORAGE));
+ apiServer.addServlet(
+ new NetDiagnosticsApiServlet(apiServer.ctx(), getApplicationContext(), Servlets.NET_DIAGNOSTICS));
webManager.add(apiServer);
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/1deff42b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/APIExecutionTest.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/APIExecutionTest.java b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/APIExecutionTest.java
index cf62e42..02f17e7 100644
--- a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/APIExecutionTest.java
+++ b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/runtime/APIExecutionTest.java
@@ -19,10 +19,16 @@
package org.apache.asterix.test.runtime;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.asterix.common.api.INcApplicationContext;
import org.apache.asterix.test.common.TestExecutor;
import org.apache.asterix.testframework.context.TestCaseContext;
+import org.apache.hyracks.control.nc.NodeControllerService;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
@@ -36,10 +42,21 @@ import org.junit.runners.Parameterized.Parameters;
@RunWith(Parameterized.class)
public class APIExecutionTest {
protected static final String TEST_CONFIG_FILE_NAME = "src/main/resources/cc.conf";
+ private static final TestExecutor testExecutor = new TestExecutor();
@BeforeClass
public static void setUp() throws Exception {
- LangExecutionUtil.setUp(TEST_CONFIG_FILE_NAME, new TestExecutor());
+ LangExecutionUtil.setUp(TEST_CONFIG_FILE_NAME, testExecutor);
+ final NodeControllerService[] ncs = ExecutionTestUtil.integrationUtil.ncs;
+ final Map<String, InetSocketAddress> ncEndPoints = new HashMap<>();
+ final String ip = InetAddress.getLoopbackAddress().getHostAddress();
+ for (NodeControllerService nc : ncs) {
+ final String nodeId = nc.getId();
+ final INcApplicationContext appCtx = (INcApplicationContext) nc.getApplicationContext();
+ int apiPort = appCtx.getExternalProperties().getNcApiPort();
+ ncEndPoints.put(nodeId, InetSocketAddress.createUnresolved(ip, apiPort));
+ }
+ testExecutor.setNcEndPoints(ncEndPoints);
}
@AfterClass
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/1deff42b/asterixdb/asterix-app/src/test/resources/runtimets/api.xml
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/api.xml b/asterixdb/asterix-app/src/test/resources/runtimets/api.xml
index b686482..8b16889 100644
--- a/asterixdb/asterix-app/src/test/resources/runtimets/api.xml
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/api.xml
@@ -118,5 +118,10 @@
<output-dir compare="Text">feed-stats</output-dir>
</compilation-unit>
</test-case>
+ <test-case FilePath="api">
+ <compilation-unit name="net-diagnostics">
+ <output-dir compare="Text">net-diagnostics</output-dir>
+ </compilation-unit>
+ </test-case>
</test-group>
</test-suite>
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/1deff42b/asterixdb/asterix-app/src/test/resources/runtimets/queries/api/net-diagnostics/net-diagnostics.1.get.http
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries/api/net-diagnostics/net-diagnostics.1.get.http b/asterixdb/asterix-app/src/test/resources/runtimets/queries/api/net-diagnostics/net-diagnostics.1.get.http
new file mode 100644
index 0000000..12288a4
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries/api/net-diagnostics/net-diagnostics.1.get.http
@@ -0,0 +1,19 @@
+/*
+ * 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.
+ */
+nc:asterix_nc1 /admin/net
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/1deff42b/asterixdb/asterix-app/src/test/resources/runtimets/results/api/net-diagnostics/net-diagnostics.1.regexadm
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/results/api/net-diagnostics/net-diagnostics.1.regexadm b/asterixdb/asterix-app/src/test/resources/runtimets/results/api/net-diagnostics/net-diagnostics.1.regexadm
new file mode 100644
index 0000000..be8d03a
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/results/api/net-diagnostics/net-diagnostics.1.regexadm
@@ -0,0 +1,5 @@
+\{
+ .*"incomingConnections" : \[.*\],
+ .*"localAddress" : .*,.*
+ .*"outgoingConnections" : \[.*\]
+\}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/1deff42b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/Servlets.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/Servlets.java b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/Servlets.java
index b135c7f..d5aa5d1 100644
--- a/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/Servlets.java
+++ b/asterixdb/asterix-common/src/main/java/org/apache/asterix/common/utils/Servlets.java
@@ -35,6 +35,7 @@ public class Servlets {
public static final String DIAGNOSTICS = "/admin/diagnostics";
public static final String ACTIVE_STATS = "/admin/active/*";
public static final String STORAGE = "/admin/storage/*";
+ public static final String NET_DIAGNOSTICS = "/admin/net/*";
private Servlets() {
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/1deff42b/hyracks-fullstack/hyracks/hyracks-comm/src/main/java/org/apache/hyracks/comm/channels/NetworkOutputChannel.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-comm/src/main/java/org/apache/hyracks/comm/channels/NetworkOutputChannel.java b/hyracks-fullstack/hyracks/hyracks-comm/src/main/java/org/apache/hyracks/comm/channels/NetworkOutputChannel.java
index 261e7c4..17cdc3e 100644
--- a/hyracks-fullstack/hyracks/hyracks-comm/src/main/java/org/apache/hyracks/comm/channels/NetworkOutputChannel.java
+++ b/hyracks-fullstack/hyracks/hyracks-comm/src/main/java/org/apache/hyracks/comm/channels/NetworkOutputChannel.java
@@ -80,8 +80,8 @@ public class NetworkOutputChannel implements IFrameWriter {
InetSocketAddress remoteAddress = ccb.getRemoteAddress();
String nameBefore = Thread.currentThread().getName();
try {
- Thread.currentThread()
- .setName(nameBefore + ":SendingTo(" + Objects.toString(remoteAddress) + ")");
+ Thread.currentThread().setName(
+ nameBefore + ":SendingTo(" + Objects.toString(remoteAddress) + ") over " + ccb);
wait();
} finally {
Thread.currentThread().setName(nameBefore);
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/1deff42b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/net/NetworkManager.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/net/NetworkManager.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/net/NetworkManager.java
index 8b02f9c..3298b78 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/net/NetworkManager.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/net/NetworkManager.java
@@ -104,6 +104,10 @@ public class NetworkManager implements IChannelConnectionFactory {
return mConn.openChannel();
}
+ public MuxDemux getMuxDemux() {
+ return md;
+ }
+
private class ChannelOpenListener implements IChannelOpenListener {
@Override
public void channelOpened(ChannelControlBlock channel) {
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/1deff42b/hyracks-fullstack/hyracks/hyracks-net/pom.xml
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-net/pom.xml b/hyracks-fullstack/hyracks/hyracks-net/pom.xml
index 192a80a..525543f 100644
--- a/hyracks-fullstack/hyracks/hyracks-net/pom.xml
+++ b/hyracks-fullstack/hyracks/hyracks-net/pom.xml
@@ -60,5 +60,9 @@
<artifactId>hyracks-util</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ </dependency>
</dependencies>
</project>
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/1deff42b/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/muxdemux/ChannelControlBlock.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/muxdemux/ChannelControlBlock.java b/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/muxdemux/ChannelControlBlock.java
index f7ef2aa..3a35212 100644
--- a/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/muxdemux/ChannelControlBlock.java
+++ b/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/muxdemux/ChannelControlBlock.java
@@ -29,6 +29,10 @@ import org.apache.hyracks.api.comm.IChannelReadInterface;
import org.apache.hyracks.api.comm.IChannelWriteInterface;
import org.apache.hyracks.api.exceptions.NetException;
import org.apache.hyracks.net.protocols.muxdemux.MultiplexedConnection.WriterState;
+import org.apache.hyracks.util.JSONUtil;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
/**
* Handle to a channel that represents a logical full-duplex communication end-point.
@@ -168,4 +172,17 @@ public class ChannelControlBlock implements IChannelControlBlock {
public InetSocketAddress getRemoteAddress() {
return cSet.getMultiplexedConnection().getRemoteAddress();
}
+
+ public JsonNode getState() {
+ final ObjectNode state = JSONUtil.createObject();
+ state.put("id", channelId);
+ state.put("localClose", localClose.get());
+ state.put("localCloseAck", localCloseAck.get());
+ state.put("remoteClose", remoteClose.get());
+ state.put("remoteCloseAck", remoteCloseAck.get());
+ state.put("readCredits", ri.getCredits());
+ state.put("writeCredits", wi.getCredits());
+ state.put("completelyClosed", completelyClosed());
+ return state;
+ }
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/1deff42b/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/muxdemux/ChannelSet.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/muxdemux/ChannelSet.java b/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/muxdemux/ChannelSet.java
index f5cdf2c..179f42c 100644
--- a/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/muxdemux/ChannelSet.java
+++ b/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/muxdemux/ChannelSet.java
@@ -24,9 +24,12 @@ import java.util.BitSet;
import org.apache.hyracks.api.comm.IChannelInterfaceFactory;
import org.apache.hyracks.api.comm.MuxDemuxCommand;
import org.apache.hyracks.api.exceptions.NetException;
+import org.apache.hyracks.util.JSONUtil;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+
public class ChannelSet {
private static final Logger LOGGER = LogManager.getLogger();
@@ -243,4 +246,14 @@ public class ChannelSet {
public MultiplexedConnection getMultiplexedConnection() {
return mConn;
}
+
+ public synchronized ArrayNode getState() {
+ final ArrayNode state = JSONUtil.createArray();
+ for (ChannelControlBlock ccb : ccbArray) {
+ if (ccb != null) {
+ state.add(ccb.getState());
+ }
+ }
+ return state;
+ }
}
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/1deff42b/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/muxdemux/MultiplexedConnection.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/muxdemux/MultiplexedConnection.java b/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/muxdemux/MultiplexedConnection.java
index 4c3836a..96ccafb 100644
--- a/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/muxdemux/MultiplexedConnection.java
+++ b/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/muxdemux/MultiplexedConnection.java
@@ -24,6 +24,7 @@ import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.util.BitSet;
+import java.util.Optional;
import org.apache.hyracks.api.comm.IChannelControlBlock;
import org.apache.hyracks.api.comm.IChannelInterfaceFactory;
@@ -32,10 +33,15 @@ import org.apache.hyracks.api.comm.MuxDemuxCommand;
import org.apache.hyracks.api.exceptions.NetException;
import org.apache.hyracks.net.protocols.tcp.ITCPConnectionEventListener;
import org.apache.hyracks.net.protocols.tcp.TCPConnection;
+import org.apache.hyracks.util.JSONUtil;
import org.apache.hyracks.util.annotations.ThreadSafetyGuaranteedBy;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+
/**
* A {@link MultiplexedConnection} can be used by clients to create multiple "channels"
* that can have independent full-duplex conversations.
@@ -442,4 +448,15 @@ public class MultiplexedConnection implements ITCPConnectionEventListener {
public InetSocketAddress getRemoteAddress() {
return tcpConnection == null ? null : tcpConnection.getRemoteAddress();
}
+
+ public synchronized Optional<JsonNode> getState() {
+ if (tcpConnection == null) {
+ return Optional.empty();
+ }
+ final ObjectNode state = JSONUtil.createObject();
+ state.put("remoteAddress", getRemoteAddress().toString());
+ final ArrayNode channels = cSet.getState();
+ state.set("channels", channels);
+ return Optional.of(state);
+ }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/1deff42b/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/muxdemux/MuxDemux.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/muxdemux/MuxDemux.java b/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/muxdemux/MuxDemux.java
index c58cb86..4ee7e83 100644
--- a/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/muxdemux/MuxDemux.java
+++ b/hyracks-fullstack/hyracks/hyracks-net/src/main/java/org/apache/hyracks/net/protocols/muxdemux/MuxDemux.java
@@ -28,6 +28,11 @@ import org.apache.hyracks.api.exceptions.NetException;
import org.apache.hyracks.net.protocols.tcp.ITCPConnectionListener;
import org.apache.hyracks.net.protocols.tcp.TCPConnection;
import org.apache.hyracks.net.protocols.tcp.TCPEndpoint;
+import org.apache.hyracks.util.JSONUtil;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
/**
* Multiplexed Connection Manager.
@@ -43,7 +48,8 @@ public class MuxDemux {
private final int maxConnectionAttempts;
- private final Map<InetSocketAddress, MultiplexedConnection> connectionMap;
+ private final Map<InetSocketAddress, MultiplexedConnection> outgoingConnectionMap;
+ private final Map<InetSocketAddress, MultiplexedConnection> incomingConnectionMap;
private final TCPEndpoint tcpEndpoint;
@@ -69,13 +75,14 @@ public class MuxDemux {
this.channelOpenListener = listener;
this.maxConnectionAttempts = maxConnectionAttempts;
this.channelInterfaceFatory = channelInterfaceFatory;
- connectionMap = new HashMap<>();
+ outgoingConnectionMap = new HashMap<>();
+ incomingConnectionMap = new HashMap<>();
this.tcpEndpoint = new TCPEndpoint(new ITCPConnectionListener() {
@Override
public void connectionEstablished(TCPConnection connection) {
MultiplexedConnection mConn;
synchronized (MuxDemux.this) {
- mConn = connectionMap.get(connection.getRemoteAddress());
+ mConn = outgoingConnectionMap.get(connection.getRemoteAddress());
}
assert mConn != null;
mConn.setTCPConnection(connection);
@@ -89,17 +96,18 @@ public class MuxDemux {
mConn.setTCPConnection(connection);
connection.setEventListener(mConn);
connection.setAttachment(mConn);
+ incomingConnectionMap.put(connection.getRemoteAddress(), mConn);
}
@Override
public void connectionFailure(InetSocketAddress remoteAddress, IOException error) {
MultiplexedConnection mConn;
synchronized (MuxDemux.this) {
- mConn = connectionMap.get(remoteAddress);
+ mConn = outgoingConnectionMap.get(remoteAddress);
assert mConn != null;
int nConnectionAttempts = mConn.getConnectionAttempts();
if (nConnectionAttempts > MuxDemux.this.maxConnectionAttempts) {
- connectionMap.remove(remoteAddress);
+ outgoingConnectionMap.remove(remoteAddress);
mConn.setConnectionFailure(new IOException(remoteAddress.toString() + ": " + error, error));
} else {
mConn.setConnectionAttempts(nConnectionAttempts + 1);
@@ -112,7 +120,9 @@ public class MuxDemux {
public void connectionClosed(TCPConnection connection) {
synchronized (MuxDemux.this) {
if (connection.getType() == TCPConnection.ConnectionType.OUTGOING) {
- connectionMap.remove(connection.getRemoteAddress());
+ outgoingConnectionMap.remove(connection.getRemoteAddress());
+ } else if (connection.getType() == TCPConnection.ConnectionType.INCOMING) {
+ incomingConnectionMap.remove(connection.getRemoteAddress());
}
}
}
@@ -144,10 +154,10 @@ public class MuxDemux {
public MultiplexedConnection connect(InetSocketAddress remoteAddress) throws InterruptedException, NetException {
MultiplexedConnection mConn;
synchronized (this) {
- mConn = connectionMap.get(remoteAddress);
+ mConn = outgoingConnectionMap.get(remoteAddress);
if (mConn == null) {
mConn = new MultiplexedConnection(this);
- connectionMap.put(remoteAddress, mConn);
+ outgoingConnectionMap.put(remoteAddress, mConn);
tcpEndpoint.initiateConnection(remoteAddress);
}
}
@@ -186,4 +196,20 @@ public class MuxDemux {
public IChannelInterfaceFactory getChannelInterfaceFactory() {
return channelInterfaceFatory;
}
+
+ public synchronized JsonNode getState() {
+ final ObjectNode state = JSONUtil.createObject();
+ state.put("localAddress", tcpEndpoint.getLocalAddress().toString());
+ final ArrayNode outgoingConnections = JSONUtil.createArray();
+ state.set("outgoingConnections", outgoingConnections);
+ for (MultiplexedConnection connection : outgoingConnectionMap.values()) {
+ connection.getState().ifPresent(outgoingConnections::add);
+ }
+ final ArrayNode incomingConnections = JSONUtil.createArray();
+ state.set("incomingConnections", incomingConnections);
+ for (MultiplexedConnection connection : incomingConnectionMap.values()) {
+ connection.getState().ifPresent(incomingConnections::add);
+ }
+ return state;
+ }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/1deff42b/hyracks-fullstack/hyracks/hyracks-util/src/main/java/org/apache/hyracks/util/JSONUtil.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/hyracks/hyracks-util/src/main/java/org/apache/hyracks/util/JSONUtil.java b/hyracks-fullstack/hyracks/hyracks-util/src/main/java/org/apache/hyracks/util/JSONUtil.java
index baa3174..006659b 100644
--- a/hyracks-fullstack/hyracks/hyracks-util/src/main/java/org/apache/hyracks/util/JSONUtil.java
+++ b/hyracks-fullstack/hyracks/hyracks-util/src/main/java/org/apache/hyracks/util/JSONUtil.java
@@ -37,11 +37,13 @@ import com.fasterxml.jackson.databind.MapperFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.ObjectWriter;
import com.fasterxml.jackson.databind.SerializationFeature;
+import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
public class JSONUtil {
private static final Logger LOGGER = LogManager.getLogger();
+ private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
private static final String INDENT = "\t";
@@ -276,4 +278,12 @@ public class JSONUtil {
public static void put(ObjectNode o, String name, List<String> elements) {
elements.forEach(o.putArray(name)::add);
}
+
+ public static ObjectNode createObject() {
+ return OBJECT_MAPPER.createObjectNode();
+ }
+
+ public static ArrayNode createArray() {
+ return OBJECT_MAPPER.createArrayNode();
+ }
}