Repository: tinkerpop Updated Branches: refs/heads/tp32 3df6c5806 -> d97056443
TINKERPOP-1726 Added idleReadLimit and idleWriteLimit for Gremlin Server This enables Gremlin Server to periodically ping clients and auto-close zombie connections when a client disappears without issuing a close. Project: http://git-wip-us.apache.org/repos/asf/tinkerpop/repo Commit: http://git-wip-us.apache.org/repos/asf/tinkerpop/commit/10ab3341 Tree: http://git-wip-us.apache.org/repos/asf/tinkerpop/tree/10ab3341 Diff: http://git-wip-us.apache.org/repos/asf/tinkerpop/diff/10ab3341 Branch: refs/heads/tp32 Commit: 10ab33410862ccb935e911b028c27f4cd835b70f Parents: bcffaad Author: Stephen Mallette <sp...@genoprime.com> Authored: Fri Feb 16 16:19:56 2018 -0500 Committer: Stephen Mallette <sp...@genoprime.com> Committed: Sat Feb 17 06:50:46 2018 -0500 ---------------------------------------------------------------------- CHANGELOG.asciidoc | 1 + .../src/reference/gremlin-applications.asciidoc | 13 ++++--- .../upgrade/release-3.2.x-incubating.asciidoc | 15 ++++++++ .../driver/handler/WebSocketClientHandler.java | 5 ++- gremlin-server/conf/gremlin-server-classic.yaml | 2 ++ .../conf/gremlin-server-modern-py.yaml | 2 ++ .../conf/gremlin-server-modern-readonly.yaml | 2 ++ gremlin-server/conf/gremlin-server-modern.yaml | 2 ++ gremlin-server/conf/gremlin-server-neo4j.yaml | 2 ++ gremlin-server/conf/gremlin-server-secure.yaml | 2 ++ gremlin-server/conf/gremlin-server-spark.yaml | 2 ++ gremlin-server/conf/gremlin-server.yaml | 2 ++ .../gremlin/server/AbstractChannelizer.java | 11 ++++-- .../tinkerpop/gremlin/server/Channelizer.java | 17 +++++++++ .../tinkerpop/gremlin/server/Settings.java | 16 ++++++++- .../server/channel/WebSocketChannelizer.java | 11 ++++++ .../server/channel/WsAndHttpChannelizer.java | 28 +++++++-------- .../server/handler/OpSelectorHandler.java | 35 +++++++++++++++++++ .../handler/WsAndHttpChannelizerHandler.java | 11 ++++-- .../server/GremlinServerIntegrateTest.java | 36 +++++++++++++++++++- .../gremlin/util/Log4jRecordingAppender.java | 5 +-- .../util/Log4jRecordingAppenderTest.java | 2 +- .../remote/gremlin-server-integration.yaml | 2 ++ .../server/gremlin-server-integration.yaml | 2 ++ .../server/gremlin-server-performance.yaml | 2 ++ 25 files changed, 198 insertions(+), 30 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/10ab3341/CHANGELOG.asciidoc ---------------------------------------------------------------------- diff --git a/CHANGELOG.asciidoc b/CHANGELOG.asciidoc index f1519b6..f516f7b 100644 --- a/CHANGELOG.asciidoc +++ b/CHANGELOG.asciidoc @@ -25,6 +25,7 @@ image::https://raw.githubusercontent.com/apache/tinkerpop/master/docs/static/ima * Modified `GremlinDslProcessor` so that it generated the `getAnonymousTraversalClass()` method to return the DSL version of `__`. * Added the "Kitchen Sink" test data set. +* Added `idleReadLimit` and `idleWriteLimit` to Gremlin Server that enables a "ping" and auto-close for seemingly dead clients. * Fixed a bug in `NumberHelper` that led to wrong min/max results if numbers exceeded the Integer limits. * Delayed setting of the request identifier until `RequestMessage` construction by the builder. * Improved error messaging for failed serialization and deserialization of request/response messages. http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/10ab3341/docs/src/reference/gremlin-applications.asciidoc ---------------------------------------------------------------------- diff --git a/docs/src/reference/gremlin-applications.asciidoc b/docs/src/reference/gremlin-applications.asciidoc index 4e4f190..2413bdb 100644 --- a/docs/src/reference/gremlin-applications.asciidoc +++ b/docs/src/reference/gremlin-applications.asciidoc @@ -1073,7 +1073,8 @@ The following table describes the various configuration options that Gremlin Ser |graphs |A `Map` of `Graph` configuration files where the key of the `Map` becomes the name to which the `Graph` will be bound and the value is the file name of a `Graph` configuration file. |_none_ |gremlinPool |The number of "Gremlin" threads available to execute actual scripts in a `ScriptEngine`. This pool represents the workers available to handle blocking operations in Gremlin Server. When set to `0`, Gremlin Server will use the value provided by `Runtime.availableProcessors()`. |0 |host |The name of the host to bind the server to. |localhost -|useEpollEventLoop |try to use epoll event loops (works only on Linux os) instead of netty NIO. |false +|idleReadLimit |Time in milliseconds that the server will allow a channel to not receive requests from a client before it automatically closes. If enabled, the value provided should typically exceed the amount of time given to `idleWriteLimit`. Set this value to `0` to disable this feature. |0 +|idleWriteLimit |Time in milliseconds that the server will allow a channel to not send responses to a client before it sends a "ping" to see if it is still present. If it is present, the client should respond with a "pong" which will thus reset the {@link #idleReadLimit} and keep the channel open. If enabled, this number should be smaller than the value provided to the `idleReadLimit`. Set this value to `0` to disable this feature. |0 |maxAccumulationBufferComponents |Maximum number of request components that can be aggregated for a message. |1024 |maxChunkSize |The maximum length of the content or each chunk. If the content length exceeds this value, the transfer encoding of the decoded request will be converted to 'chunked' and the content will be split into multiple `HttpContent` objects. If the transfer encoding of the HTTP request is 'chunked' already, each chunk will be split into smaller chunks if the length of the chunk exceeds this value. |8192 |maxContentLength |The maximum length of the aggregated content for a message. Works in concert with `maxChunkSize` where chunked requests are accumulated back into a single message. A request exceeding this size will return a `413 - Request Entity Too Large` status code. A response exceeding this size will raise an internal exception. |65536 @@ -1121,6 +1122,7 @@ The following table describes the various configuration options that Gremlin Ser |strictTransactionManagement |Set to `true` to require `aliases` to be submitted on every requests, where the `aliases` become the scope of transaction management. |false |threadPoolBoss |The number of threads available to Gremlin Server for accepting connections. Should always be set to `1`. |1 |threadPoolWorker |The number of threads available to Gremlin Server for processing non-blocking reads and writes. |1 +|useEpollEventLoop |try to use epoll event loops (works only on Linux os) instead of netty NIO. |false |writeBufferHighWaterMark | If the number of bytes in the network send buffer exceeds this value then the channel is no longer writeable, accepting no additional writes until buffer is drained and the `writeBufferLowWaterMark` is met. |65536 |writeBufferLowWaterMark | Once the number of bytes queued in the network send buffer exceeds the `writeBufferHighWaterMark`, the channel will not become writeable again until the buffer is drained and it drops below this value. |65536 |========================================================= @@ -1160,8 +1162,7 @@ The `SessionOpProcessor` provides a way to interact with Gremlin Server over a < |sessionTimeout |Time in milliseconds before a session will time out. |28800000 |========================================================= -StandardOpProcessor -++++++++++++++++++ +===== StandardOpProcessor The `StandardOpProcessor` provides a way to interact with Gremlin Server without use of sessions and is the default method for processing script evaluation requests. @@ -1593,8 +1594,10 @@ The following sections define best practices for working with Gremlin Server. image:gremlin-handdrawn.png[width=120,float=right] Tuning Gremlin Server for a particular environment may require some simple trial-and-error, but the following represent some basic guidelines that might be useful: -* Gremlin Server defaults to a very modest maximum heap size. Consider increasing this value for non-trivial uses. Maximum heap size (`-Xmx`) is defined with the `JAVA_OPTIONS` setting in `gremlin-server.sh`. -* When configuring the size of `threadPoolWorker` start with the default of `1` and increment by one as needed to a maximum of `2*number of cores`. +* Gremlin Server defaults to a very modest maximum heap size. Consider increasing this value for non-trivial uses. +Maximum heap size (`-Xmx`) is defined with the `JAVA_OPTIONS` setting in `gremlin-server.sh`. +* When configuring the size of `threadPoolWorker` start with the default of `1` and increment by one as needed to a +maximum of `2*number of cores`. * The "right" size of the `gremlinPool` setting is somewhat dependent on the type of scripts that will be processed by Gremlin Server. As requests arrive to Gremlin Server they are decoded and queued to be processed by threads in this pool. When this pool is exhausted of threads, Gremlin Server will continue to accept incoming requests, but http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/10ab3341/docs/src/upgrade/release-3.2.x-incubating.asciidoc ---------------------------------------------------------------------- diff --git a/docs/src/upgrade/release-3.2.x-incubating.asciidoc b/docs/src/upgrade/release-3.2.x-incubating.asciidoc index 00a6e83..f623127 100644 --- a/docs/src/upgrade/release-3.2.x-incubating.asciidoc +++ b/docs/src/upgrade/release-3.2.x-incubating.asciidoc @@ -27,6 +27,21 @@ image::https://raw.githubusercontent.com/apache/tinkerpop/master/docs/static/ima Please see the link:https://github.com/apache/tinkerpop/blob/3.2.8/CHANGELOG.asciidoc#release-3-2-8[changelog] for a complete list of all the modifications that are part of this release. +=== Upgrading for Users + +==== Improved Connection Monitoring + +Gremlin Server now has two new settings: `idleReadLimit` and `idleWriteLimit`. The `idleWriteLimit` tells Gremlin +Server how long it should wait between writes to a client before it issues a "ping" to that client to see if it is +still present. The `idleReadLimit` represents how long Gremlin Server should wait between requests from a client before +it closes the connection on the server side. By default, these two configurations are set to zero, meaning that they +are both disabled. + +This change should help to alleviate issues where connections are left open on the server longer than they should be +by clients that might mysteriously disappear without properly closing their connections. + +See: link:https://issues.apache.org/jira/browse/TINKERPOP-1726[TINKERPOP-1726] + === Upgrading for Providers ==== Graph System Providers http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/10ab3341/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/WebSocketClientHandler.java ---------------------------------------------------------------------- diff --git a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/WebSocketClientHandler.java b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/WebSocketClientHandler.java index c63d790..0a1f2f7 100644 --- a/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/WebSocketClientHandler.java +++ b/gremlin-driver/src/main/java/org/apache/tinkerpop/gremlin/driver/handler/WebSocketClientHandler.java @@ -26,6 +26,7 @@ import io.netty.channel.SimpleChannelInboundHandler; import io.netty.handler.codec.http.FullHttpResponse; import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame; import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame; +import io.netty.handler.codec.http.websocketx.PingWebSocketFrame; import io.netty.handler.codec.http.websocketx.PongWebSocketFrame; import io.netty.handler.codec.http.websocketx.TextWebSocketFrame; import io.netty.handler.codec.http.websocketx.WebSocketClientHandshaker; @@ -85,7 +86,9 @@ public final class WebSocketClientHandler extends SimpleChannelInboundHandler<Ob final WebSocketFrame frame = (WebSocketFrame) msg; if (frame instanceof TextWebSocketFrame) { ctx.fireChannelRead(frame.retain(2)); - } else if (frame instanceof PongWebSocketFrame) { + } else if (frame instanceof PingWebSocketFrame) { + ctx.writeAndFlush(new PongWebSocketFrame()); + }else if (frame instanceof PongWebSocketFrame) { logger.debug("Received response from keep-alive request"); } else if (frame instanceof BinaryWebSocketFrame) { ctx.fireChannelRead(frame.retain(2)); http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/10ab3341/gremlin-server/conf/gremlin-server-classic.yaml ---------------------------------------------------------------------- diff --git a/gremlin-server/conf/gremlin-server-classic.yaml b/gremlin-server/conf/gremlin-server-classic.yaml index b56960d..6a7f3f4 100644 --- a/gremlin-server/conf/gremlin-server-classic.yaml +++ b/gremlin-server/conf/gremlin-server-classic.yaml @@ -36,6 +36,8 @@ serializers: metrics: { slf4jReporter: {enabled: true, interval: 180000}} strictTransactionManagement: false +idleReadLimit: 0 +idleWriteLimit: 0 maxInitialLineLength: 4096 maxHeaderSize: 8192 maxChunkSize: 8192 http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/10ab3341/gremlin-server/conf/gremlin-server-modern-py.yaml ---------------------------------------------------------------------- diff --git a/gremlin-server/conf/gremlin-server-modern-py.yaml b/gremlin-server/conf/gremlin-server-modern-py.yaml index a04681f..726890d 100644 --- a/gremlin-server/conf/gremlin-server-modern-py.yaml +++ b/gremlin-server/conf/gremlin-server-modern-py.yaml @@ -51,6 +51,8 @@ serializers: metrics: { slf4jReporter: {enabled: true, interval: 180000}} strictTransactionManagement: false +idleReadLimit: 0 +idleWriteLimit: 0 threadPoolBoss: 1 maxInitialLineLength: 4096 maxHeaderSize: 8192 http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/10ab3341/gremlin-server/conf/gremlin-server-modern-readonly.yaml ---------------------------------------------------------------------- diff --git a/gremlin-server/conf/gremlin-server-modern-readonly.yaml b/gremlin-server/conf/gremlin-server-modern-readonly.yaml index c2073ea..6840e18 100644 --- a/gremlin-server/conf/gremlin-server-modern-readonly.yaml +++ b/gremlin-server/conf/gremlin-server-modern-readonly.yaml @@ -36,6 +36,8 @@ serializers: metrics: { slf4jReporter: {enabled: true, interval: 180000}} strictTransactionManagement: false +idleReadLimit: 0 +idleWriteLimit: 0 maxInitialLineLength: 4096 maxHeaderSize: 8192 maxChunkSize: 8192 http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/10ab3341/gremlin-server/conf/gremlin-server-modern.yaml ---------------------------------------------------------------------- diff --git a/gremlin-server/conf/gremlin-server-modern.yaml b/gremlin-server/conf/gremlin-server-modern.yaml index 16eba68..6ec3e23 100644 --- a/gremlin-server/conf/gremlin-server-modern.yaml +++ b/gremlin-server/conf/gremlin-server-modern.yaml @@ -36,6 +36,8 @@ serializers: metrics: { slf4jReporter: {enabled: true, interval: 180000}} strictTransactionManagement: false +idleReadLimit: 0 +idleWriteLimit: 0 maxInitialLineLength: 4096 maxHeaderSize: 8192 maxChunkSize: 8192 http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/10ab3341/gremlin-server/conf/gremlin-server-neo4j.yaml ---------------------------------------------------------------------- diff --git a/gremlin-server/conf/gremlin-server-neo4j.yaml b/gremlin-server/conf/gremlin-server-neo4j.yaml index 8d91374..a1a32ca 100644 --- a/gremlin-server/conf/gremlin-server-neo4j.yaml +++ b/gremlin-server/conf/gremlin-server-neo4j.yaml @@ -55,6 +55,8 @@ metrics: { gangliaReporter: {enabled: false, interval: 180000, addressingMode: MULTICAST}, graphiteReporter: {enabled: false, interval: 180000}} strictTransactionManagement: false +idleReadLimit: 0 +idleWriteLimit: 0 maxInitialLineLength: 4096 maxHeaderSize: 8192 maxChunkSize: 8192 http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/10ab3341/gremlin-server/conf/gremlin-server-secure.yaml ---------------------------------------------------------------------- diff --git a/gremlin-server/conf/gremlin-server-secure.yaml b/gremlin-server/conf/gremlin-server-secure.yaml index 197e3a7..771f31b 100644 --- a/gremlin-server/conf/gremlin-server-secure.yaml +++ b/gremlin-server/conf/gremlin-server-secure.yaml @@ -58,6 +58,8 @@ metrics: { gangliaReporter: {enabled: false, interval: 180000, addressingMode: MULTICAST}, graphiteReporter: {enabled: false, interval: 180000}} strictTransactionManagement: false +idleReadLimit: 0 +idleWriteLimit: 0 maxInitialLineLength: 4096 maxHeaderSize: 8192 maxChunkSize: 8192 http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/10ab3341/gremlin-server/conf/gremlin-server-spark.yaml ---------------------------------------------------------------------- diff --git a/gremlin-server/conf/gremlin-server-spark.yaml b/gremlin-server/conf/gremlin-server-spark.yaml index 5ff7cc8..747845a 100644 --- a/gremlin-server/conf/gremlin-server-spark.yaml +++ b/gremlin-server/conf/gremlin-server-spark.yaml @@ -68,6 +68,8 @@ metrics: { gangliaReporter: {enabled: false, interval: 180000, addressingMode: MULTICAST}, graphiteReporter: {enabled: false, interval: 180000}} strictTransactionManagement: false +idleReadLimit: 0 +idleWriteLimit: 0 maxInitialLineLength: 4096 maxHeaderSize: 8192 maxChunkSize: 8192 http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/10ab3341/gremlin-server/conf/gremlin-server.yaml ---------------------------------------------------------------------- diff --git a/gremlin-server/conf/gremlin-server.yaml b/gremlin-server/conf/gremlin-server.yaml index 71d1d91..6b61d4b 100644 --- a/gremlin-server/conf/gremlin-server.yaml +++ b/gremlin-server/conf/gremlin-server.yaml @@ -45,6 +45,8 @@ metrics: { gangliaReporter: {enabled: false, interval: 180000, addressingMode: MULTICAST}, graphiteReporter: {enabled: false, interval: 180000}} strictTransactionManagement: false +idleReadLimit: 0 +idleWriteLimit: 0 maxInitialLineLength: 4096 maxHeaderSize: 8192 maxChunkSize: 8192 http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/10ab3341/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/AbstractChannelizer.java ---------------------------------------------------------------------- diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/AbstractChannelizer.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/AbstractChannelizer.java index c4d8398..0bc4e00 100644 --- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/AbstractChannelizer.java +++ b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/AbstractChannelizer.java @@ -23,6 +23,7 @@ import io.netty.handler.ssl.SslContext; import io.netty.handler.ssl.SslContextBuilder; import io.netty.handler.ssl.SslProvider; import io.netty.handler.ssl.util.SelfSignedCertificate; +import io.netty.handler.timeout.IdleStateHandler; import org.apache.tinkerpop.gremlin.driver.MessageSerializer; import org.apache.tinkerpop.gremlin.driver.ser.AbstractGryoMessageSerializerV1d0; import org.apache.tinkerpop.gremlin.driver.ser.GraphSONMessageSerializerV1d0; @@ -98,6 +99,7 @@ public abstract class AbstractChannelizer extends ChannelInitializer<SocketChann protected final Map<String, MessageSerializer> serializers = new HashMap<>(); + private IdleStateHandler idleStateHandler; private OpSelectorHandler opSelectorHandler; private OpExecutorHandler opExecutorHandler; private IteratorHandler iteratorHandler; @@ -138,7 +140,7 @@ public abstract class AbstractChannelizer extends ChannelInitializer<SocketChann authenticator = createAuthenticator(settings.authentication); // these handlers don't share any state and can thus be initialized once per pipeline - opSelectorHandler = new OpSelectorHandler(settings, graphManager, gremlinExecutor, scheduledExecutorService); + opSelectorHandler = new OpSelectorHandler(settings, graphManager, gremlinExecutor, scheduledExecutorService, this); opExecutorHandler = new OpExecutorHandler(settings, graphManager, gremlinExecutor, scheduledExecutorService); iteratorHandler = new IteratorHandler(settings); } @@ -147,7 +149,12 @@ public abstract class AbstractChannelizer extends ChannelInitializer<SocketChann public void initChannel(final SocketChannel ch) throws Exception { final ChannelPipeline pipeline = ch.pipeline(); - if (sslContext.isPresent()) pipeline.addLast(PIPELINE_SSL, sslContext.get().newHandler(ch.alloc())); + sslContext.ifPresent(sslContext -> pipeline.addLast(PIPELINE_SSL, sslContext.newHandler(ch.alloc()))); + + // checks for no activity on a channel and triggers an event that is consumed by the OpSelectorHandler + // and either closes the connection or sends a ping to see if the client is still alive + if (supportsIdleMonitor()) + pipeline.addLast(new IdleStateHandler((int) (settings.idleReadLimit / 1000), (int) (settings.idleWriteLimit / 1000),0)); // the implementation provides the method by which Gremlin Server will process requests. the end of the // pipeline must decode to an incoming RequestMessage instances and encode to a outgoing ResponseMessage http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/10ab3341/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/Channelizer.java ---------------------------------------------------------------------- diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/Channelizer.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/Channelizer.java index fd7821e..15fb7d3 100644 --- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/Channelizer.java +++ b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/Channelizer.java @@ -36,4 +36,21 @@ public interface Channelizer extends ChannelHandler { * This method is called just after the {@code Channelizer} is initialized. */ public void init(final ServerGremlinExecutor<EventLoopGroup> serverGremlinExecutor); + + /** + * Create a message to send to seemingly dead clients to see if they respond back. The message sent will be + * dependent on the implementation. For example, a websocket implementation would create a "ping" message. + * This method will only be used if {@link #supportsIdleMonitor()} is {@code true}. + */ + public default Object createIdleDetectionMessage() { + return null; + } + + /** + * Determines if the channelizer supports a method for keeping the connection alive and auto-closing zombie + * channels. + */ + public default boolean supportsIdleMonitor() { + return false; + } } http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/10ab3341/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/Settings.java ---------------------------------------------------------------------- diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/Settings.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/Settings.java index d0bd8fd..75466ca 100644 --- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/Settings.java +++ b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/Settings.java @@ -31,7 +31,6 @@ import org.apache.tinkerpop.gremlin.server.channel.WebSocketChannelizer; import org.apache.tinkerpop.gremlin.server.handler.AbstractAuthenticationHandler; import org.apache.tinkerpop.gremlin.server.util.DefaultGraphManager; import info.ganglia.gmetric4j.gmetric.GMetric; -import org.apache.tinkerpop.gremlin.server.op.session.SessionOpProcessor; import org.apache.tinkerpop.gremlin.server.util.LifeCycleHook; import org.apache.tinkerpop.gremlin.structure.Graph; import org.yaml.snakeyaml.TypeDescription; @@ -165,6 +164,21 @@ public class Settings { public int writeBufferLowWaterMark = 1024 * 32; /** + * Time in milliseconds that the server will allow a channel to not receive requests from a client before it + * automatically closes. If enabled, the value provided should typically exceed the amount of time given to + * {@link #idleWriteLimit}. Set this value to 0 to disable this feature. + */ + public long idleReadLimit = 0; + + /** + * Time in milliseconds that the server will allow a channel to not send responses to a client before it sends + * a "ping" to see if it is still present. If it is present, the client should respond with a "pong" which will + * thus reset the {@link #idleReadLimit} and keep the channel open. If enabled, this number should be smaller than + * the value provided to the {@link #idleReadLimit}. Set this value to 0 to disable this feature. + */ + public long idleWriteLimit = 0; + + /** * If set to {@code true} the {@code aliases} option is required on requests and Gremlin Server will use that * information to control which {@link Graph} instances are transaction managed for that request. If this * setting is false (which is the default), specification of {@code aliases} is not required and Gremlin http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/10ab3341/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/channel/WebSocketChannelizer.java ---------------------------------------------------------------------- diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/channel/WebSocketChannelizer.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/channel/WebSocketChannelizer.java index af41cc6..a6ec20b 100644 --- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/channel/WebSocketChannelizer.java +++ b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/channel/WebSocketChannelizer.java @@ -19,6 +19,7 @@ package org.apache.tinkerpop.gremlin.server.channel; import io.netty.channel.EventLoopGroup; +import io.netty.handler.codec.http.websocketx.PingWebSocketFrame; import org.apache.tinkerpop.gremlin.server.AbstractChannelizer; import org.apache.tinkerpop.gremlin.server.auth.AllowAllAuthenticator; import org.apache.tinkerpop.gremlin.server.handler.AbstractAuthenticationHandler; @@ -113,6 +114,16 @@ public class WebSocketChannelizer extends AbstractChannelizer { pipeline.addLast(PIPELINE_AUTHENTICATOR, authenticationHandler); } + @Override + public boolean supportsIdleMonitor() { + return true; + } + + @Override + public Object createIdleDetectionMessage() { + return new PingWebSocketFrame(); + } + private AbstractAuthenticationHandler instantiateAuthenticationHandler(final Settings.AuthenticationSettings authSettings) { final String authenticationHandler = authSettings.authenticationHandler; if (authenticationHandler == null) { http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/10ab3341/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/channel/WsAndHttpChannelizer.java ---------------------------------------------------------------------- diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/channel/WsAndHttpChannelizer.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/channel/WsAndHttpChannelizer.java index 58885fb..e415887 100644 --- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/channel/WsAndHttpChannelizer.java +++ b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/channel/WsAndHttpChannelizer.java @@ -18,32 +18,20 @@ */ package org.apache.tinkerpop.gremlin.server.channel; -import io.netty.channel.ChannelInboundHandlerAdapter; import io.netty.channel.ChannelPipeline; import io.netty.channel.EventLoopGroup; -import io.netty.handler.codec.http.HttpObjectAggregator; -import io.netty.handler.codec.http.HttpRequestDecoder; -import io.netty.handler.codec.http.HttpResponseEncoder; -import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler; -import io.netty.handler.logging.LogLevel; -import io.netty.handler.logging.LoggingHandler; import org.apache.tinkerpop.gremlin.server.AbstractChannelizer; import org.apache.tinkerpop.gremlin.server.handler.HttpGremlinEndpointHandler; import org.apache.tinkerpop.gremlin.server.handler.WsAndHttpChannelizerHandler; import org.apache.tinkerpop.gremlin.server.util.ServerGremlinExecutor; -import org.apache.tinkerpop.gremlin.server.channel.WebSocketChannelizer; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; /** - *A channelizer for port unification with websockets and http - *@author Keith Lohnes lohn...@gmail.com + * A channelizer for port unification with websockets and http + * + * @author Keith Lohnes lohn...@gmail.com */ - public class WsAndHttpChannelizer extends AbstractChannelizer { - private static final Logger logger = LoggerFactory.getLogger(WsAndHttpChannelizer.class); - private WsAndHttpChannelizerHandler handler; @Override @@ -58,4 +46,14 @@ public class WsAndHttpChannelizer extends AbstractChannelizer { handler.configure(pipeline); pipeline.addAfter(PIPELINE_HTTP_REQUEST_DECODER, "WsAndHttpChannelizerHandler", handler); } + + @Override + public boolean supportsIdleMonitor() { + return handler.getWsChannelizer().supportsIdleMonitor(); + } + + @Override + public Object createIdleDetectionMessage() { + return handler.getWsChannelizer().createIdleDetectionMessage(); + } } http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/10ab3341/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/OpSelectorHandler.java ---------------------------------------------------------------------- diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/OpSelectorHandler.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/OpSelectorHandler.java index 30055bf..8bb9c44 100644 --- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/OpSelectorHandler.java +++ b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/OpSelectorHandler.java @@ -19,10 +19,13 @@ package org.apache.tinkerpop.gremlin.server.handler; import com.codahale.metrics.Meter; +import io.netty.handler.timeout.IdleState; +import io.netty.handler.timeout.IdleStateEvent; import org.apache.tinkerpop.gremlin.driver.message.RequestMessage; import org.apache.tinkerpop.gremlin.driver.message.ResponseMessage; import org.apache.tinkerpop.gremlin.driver.message.ResponseStatusCode; import org.apache.tinkerpop.gremlin.groovy.engine.GremlinExecutor; +import org.apache.tinkerpop.gremlin.server.Channelizer; import org.apache.tinkerpop.gremlin.server.Context; import org.apache.tinkerpop.gremlin.server.GraphManager; import org.apache.tinkerpop.gremlin.server.GremlinServer; @@ -66,13 +69,24 @@ public class OpSelectorHandler extends MessageToMessageDecoder<RequestMessage> { private final GremlinExecutor gremlinExecutor; private final ScheduledExecutorService scheduledExecutorService; + private final Channelizer channelizer; + /** + * @deprecated As of release 3.2.8, replaced by {@link #OpSelectorHandler(Settings, GraphManager, GremlinExecutor, ScheduledExecutorService, Channelizer)} + */ + @Deprecated public OpSelectorHandler(final Settings settings, final GraphManager graphManager, final GremlinExecutor gremlinExecutor, final ScheduledExecutorService scheduledExecutorService) { + this(settings, graphManager, gremlinExecutor, scheduledExecutorService, null); + } + + public OpSelectorHandler(final Settings settings, final GraphManager graphManager, final GremlinExecutor gremlinExecutor, + final ScheduledExecutorService scheduledExecutorService, final Channelizer channelizer) { this.settings = settings; this.graphManager = graphManager; this.gremlinExecutor = gremlinExecutor; this.scheduledExecutorService = scheduledExecutorService; + this.channelizer = channelizer; } @Override @@ -99,4 +113,25 @@ public class OpSelectorHandler extends MessageToMessageDecoder<RequestMessage> { ctx.writeAndFlush(ope.getResponseMessage()); } } + + @Override + public void userEventTriggered(final ChannelHandlerContext ctx, final Object evt) throws Exception { + // only need to handle this event if the idle monitor is on + if (!channelizer.supportsIdleMonitor()) return; + + if (evt instanceof IdleStateEvent) { + final IdleStateEvent e = (IdleStateEvent) evt; + + // if no requests (reader) then close, if no writes from server to client then ping. clients should + // periodically ping the server, but coming from this direction allows the server to kill channels that + // have dead clients on the other end + if (e.state() == IdleState.READER_IDLE) { + logger.info("Closing channel - client is disconnected after idle period of " + settings.idleReadLimit + " " + ctx.channel()); + ctx.close(); + } else if (e.state() == IdleState.WRITER_IDLE) { + logger.info("Checking channel - sending ping to client after idle period of " + settings.idleWriteLimit + " " + ctx.channel()); + ctx.writeAndFlush(channelizer.createIdleDetectionMessage()); + } + } + } } http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/10ab3341/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/WsAndHttpChannelizerHandler.java ---------------------------------------------------------------------- diff --git a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/WsAndHttpChannelizerHandler.java b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/WsAndHttpChannelizerHandler.java index 328a34b..6d32194 100644 --- a/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/WsAndHttpChannelizerHandler.java +++ b/gremlin-server/src/main/java/org/apache/tinkerpop/gremlin/server/handler/WsAndHttpChannelizerHandler.java @@ -26,8 +26,10 @@ import io.netty.channel.EventLoopGroup; import io.netty.handler.codec.http.HttpMessage; import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler; import io.netty.handler.codec.http.HttpServerCodec; +import org.apache.tinkerpop.gremlin.server.Channelizer; import org.apache.tinkerpop.gremlin.server.channel.HttpChannelizer; import org.apache.tinkerpop.gremlin.server.channel.WebSocketChannelizer; +import org.apache.tinkerpop.gremlin.server.channel.WsAndHttpChannelizer; import org.apache.tinkerpop.gremlin.server.handler.HttpGremlinEndpointHandler; import org.apache.tinkerpop.gremlin.server.handler.WsAndHttpChannelizerHandler; import org.apache.tinkerpop.gremlin.server.handler.WebSocketHandlerUtil; @@ -39,9 +41,10 @@ import static org.apache.tinkerpop.gremlin.server.channel.WebSocketChannelizer.P import static org.apache.tinkerpop.gremlin.server.channel.WebSocketChannelizer.PIPELINE_REQUEST_HANDLER; import static org.apache.tinkerpop.gremlin.server.channel.WebSocketChannelizer.PIPELINE_HTTP_RESPONSE_ENCODER; -/* +/** * A ChannelInboundHandlerAdapter for use with {@link WsAndHttpChannelizer} that toggles between WebSockets - * and http + * and http. + * * @author Keith Lohnes lohn...@gmail.com */ @ChannelHandler.Sharable @@ -56,6 +59,10 @@ public class WsAndHttpChannelizerHandler extends ChannelInboundHandlerAdapter { this.httpGremlinEndpointHandler = httpGremlinEndpointHandler; } + public Channelizer getWsChannelizer() { + return wsChannelizer; + } + public void configure(final ChannelPipeline pipeline) { wsChannelizer.configure(pipeline); } http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/10ab3341/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerIntegrateTest.java ---------------------------------------------------------------------- diff --git a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerIntegrateTest.java b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerIntegrateTest.java index c401de6..55912e1 100644 --- a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerIntegrateTest.java +++ b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/server/GremlinServerIntegrateTest.java @@ -27,6 +27,7 @@ import io.netty.handler.ssl.util.SelfSignedCertificate; import org.apache.commons.configuration.BaseConfiguration; import org.apache.commons.configuration.Configuration; import org.apache.commons.lang.exception.ExceptionUtils; +import org.apache.log4j.Level; import org.apache.log4j.Logger; import org.apache.tinkerpop.gremlin.TestHelper; import org.apache.tinkerpop.gremlin.driver.Client; @@ -54,6 +55,7 @@ import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversal; import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversalSource; import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__; import org.apache.tinkerpop.gremlin.process.traversal.step.util.BulkSet; +import org.apache.tinkerpop.gremlin.server.handler.OpSelectorHandler; import org.apache.tinkerpop.gremlin.server.op.AbstractEvalOpProcessor; import org.apache.tinkerpop.gremlin.server.op.standard.StandardOpProcessor; import org.apache.tinkerpop.gremlin.structure.Graph; @@ -82,7 +84,6 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; -import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Supplier; import java.util.stream.Collectors; import java.util.stream.IntStream; @@ -112,6 +113,7 @@ public class GremlinServerIntegrateTest extends AbstractGremlinServerIntegration private static final String KEY_PASS = "changeit"; private static final String CLIENT_KEY = "src/test/resources/client.key.pk8"; private static final String CLIENT_CRT = "src/test/resources/client.crt"; + private Level previousLogLevel; private Log4jRecordingAppender recordingAppender = null; private final Supplier<Graph> graphGetter = () -> server.getServerGremlinExecutor().getGraphManager().getGraph("graph"); @@ -128,12 +130,26 @@ public class GremlinServerIntegrateTest extends AbstractGremlinServerIntegration public void setupForEachTest() { recordingAppender = new Log4jRecordingAppender(); final Logger rootLogger = Logger.getRootLogger(); + + if (name.getMethodName().equals("shouldPingChannelIfClientDies")) { + final org.apache.log4j.Logger webSocketClientHandlerLogger = org.apache.log4j.Logger.getLogger(OpSelectorHandler.class); + previousLogLevel = webSocketClientHandlerLogger.getLevel(); + webSocketClientHandlerLogger.setLevel(Level.INFO); + } + rootLogger.addAppender(recordingAppender); } @After public void teardownForEachTest() { final Logger rootLogger = Logger.getRootLogger(); + + if (name.getMethodName().equals("shouldPingChannelIfClientDies")) { + final org.apache.log4j.Logger webSocketClientHandlerLogger = org.apache.log4j.Logger.getLogger(OpSelectorHandler.class); + previousLogLevel = webSocketClientHandlerLogger.getLevel(); + webSocketClientHandlerLogger.setLevel(previousLogLevel); + } + rootLogger.removeAppender(recordingAppender); } @@ -234,6 +250,9 @@ public class GremlinServerIntegrateTest extends AbstractGremlinServerIntegration case "shouldTimeOutRemoteTraversal": settings.scriptEvaluationTimeout = 500; break; + case "shouldPingChannelIfClientDies": + settings.idleWriteLimit = 1000; + break; } return settings; @@ -294,6 +313,21 @@ public class GremlinServerIntegrateTest extends AbstractGremlinServerIntegration } @Test + public void shouldPingChannelIfClientDies() throws Exception { + final Client client = TestClientFactory.build().maxConnectionPoolSize(1).minConnectionPoolSize(1).keepAliveInterval(0).create().connect(); + client.submit("1+1").all().get(); + + // since we do nothing for 3 seconds and the time limit for ping is 1 second we should get *about* 3 pings - + // i don't think the assertion needs to be too accurate. just need to make sure there's a ping message out + // there record + Thread.sleep(3000); + + assertThat(recordingAppender.logContainsAny(".*Checking channel - sending ping to client after idle period of .*$"), is(true)); + + client.close(); + } + + @Test public void shouldTimeOutRemoteTraversal() throws Exception { final Graph graph = EmptyGraph.instance(); final GraphTraversalSource g = graph.traversal().withRemote(conf); http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/10ab3341/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/util/Log4jRecordingAppender.java ---------------------------------------------------------------------- diff --git a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/util/Log4jRecordingAppender.java b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/util/Log4jRecordingAppender.java index e44f72f..82c14ac 100644 --- a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/util/Log4jRecordingAppender.java +++ b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/util/Log4jRecordingAppender.java @@ -60,7 +60,8 @@ public class Log4jRecordingAppender extends AppenderSkeleton { messages.clear(); } - public boolean logContainsAny(final String fragment) { - return messages.stream().anyMatch(m -> m.contains(fragment)); + public boolean logContainsAny(final String regex) { + // chop off the line feed so that the regex doesn't have to account for that + return messages.stream().anyMatch(m -> m.substring(0,m.length() - 1).matches(regex)); } } \ No newline at end of file http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/10ab3341/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/util/Log4jRecordingAppenderTest.java ---------------------------------------------------------------------- diff --git a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/util/Log4jRecordingAppenderTest.java b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/util/Log4jRecordingAppenderTest.java index dd1ea63..624dfc4 100644 --- a/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/util/Log4jRecordingAppenderTest.java +++ b/gremlin-server/src/test/java/org/apache/tinkerpop/gremlin/util/Log4jRecordingAppenderTest.java @@ -69,7 +69,7 @@ public class Log4jRecordingAppenderTest { @Test public void shouldMatchAnyMessages() { - assertTrue(recordingAppender.logContainsAny("ERROR")); + assertTrue(recordingAppender.logContainsAny("ERROR.*")); } @Test http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/10ab3341/gremlin-server/src/test/resources/org/apache/tinkerpop/gremlin/driver/remote/gremlin-server-integration.yaml ---------------------------------------------------------------------- diff --git a/gremlin-server/src/test/resources/org/apache/tinkerpop/gremlin/driver/remote/gremlin-server-integration.yaml b/gremlin-server/src/test/resources/org/apache/tinkerpop/gremlin/driver/remote/gremlin-server-integration.yaml index 28ea1ac..e08973f 100644 --- a/gremlin-server/src/test/resources/org/apache/tinkerpop/gremlin/driver/remote/gremlin-server-integration.yaml +++ b/gremlin-server/src/test/resources/org/apache/tinkerpop/gremlin/driver/remote/gremlin-server-integration.yaml @@ -44,6 +44,8 @@ processors: metrics: { slf4jReporter: {enabled: true, interval: 180000}} strictTransactionManagement: false +idleReadLimit: 0 +idleWriteLimit: 0 maxInitialLineLength: 4096 maxHeaderSize: 8192 maxChunkSize: 8192 http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/10ab3341/gremlin-server/src/test/resources/org/apache/tinkerpop/gremlin/server/gremlin-server-integration.yaml ---------------------------------------------------------------------- diff --git a/gremlin-server/src/test/resources/org/apache/tinkerpop/gremlin/server/gremlin-server-integration.yaml b/gremlin-server/src/test/resources/org/apache/tinkerpop/gremlin/server/gremlin-server-integration.yaml index f80c38a..b03e057 100644 --- a/gremlin-server/src/test/resources/org/apache/tinkerpop/gremlin/server/gremlin-server-integration.yaml +++ b/gremlin-server/src/test/resources/org/apache/tinkerpop/gremlin/server/gremlin-server-integration.yaml @@ -39,6 +39,8 @@ processors: metrics: { slf4jReporter: {enabled: true, interval: 180000}} strictTransactionManagement: false +idleReadLimit: 0 +idleWriteLimit: 0 maxInitialLineLength: 4096 maxHeaderSize: 8192 maxChunkSize: 8192 http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/10ab3341/gremlin-server/src/test/resources/org/apache/tinkerpop/gremlin/server/gremlin-server-performance.yaml ---------------------------------------------------------------------- diff --git a/gremlin-server/src/test/resources/org/apache/tinkerpop/gremlin/server/gremlin-server-performance.yaml b/gremlin-server/src/test/resources/org/apache/tinkerpop/gremlin/server/gremlin-server-performance.yaml index d30635d..a88b5a7 100644 --- a/gremlin-server/src/test/resources/org/apache/tinkerpop/gremlin/server/gremlin-server-performance.yaml +++ b/gremlin-server/src/test/resources/org/apache/tinkerpop/gremlin/server/gremlin-server-performance.yaml @@ -36,6 +36,8 @@ processors: metrics: { slf4jReporter: {enabled: true, interval: 180000}} strictTransactionManagement: false +idleReadLimit: 0 +idleWriteLimit: 0 maxInitialLineLength: 4096 maxHeaderSize: 8192 maxChunkSize: 8192
