Skip to content

Commit 7372360

Browse files
finnegancarrollfinnegancarroll
authored
Additional settings for gRPC experimental transport. (#17973)
'grpc.netty.max_concurrent_connection_calls' Limit concurrent in flight requests per client connection. 'grpc.netty.max_connection_age' Set timeout for persistent connections. 'grpc.netty.max_connection_idle' Set timeout for idle connections. 'grpc.netty.keepalive_timeout' Set timeout for connection keepalive pings. Signed-off-by: Finn Carroll <[email protected]>
1 parent 3a7d8fa commit 7372360

File tree

4 files changed

+105
-14
lines changed

4 files changed

+105
-14
lines changed

‎plugins/transport-grpc/README.md

+12-8
Original file line numberDiff line numberDiff line change
@@ -17,14 +17,18 @@ setting 'aux.transport.types', '[experiment
1717
setting 'aux.transport.experimental-secure-transport-grpc.port', '9400-9500' //optional
1818
```
1919

20-
Other settings are agnostic as to the gRPC transport type:
21-
22-
```
23-
setting 'grpc.publish_port', '9400'
24-
setting 'grpc.host', '["0.0.0.0"]'
25-
setting 'grpc.bind_host', '["0.0.0.0", "::", "10.0.0.1"]'
26-
setting 'grpc.publish_host', '["thisnode.example.com"]'
27-
setting 'grpc.netty.worker_count', '2'
20+
Other gRPC settings:
21+
22+
```
23+
setting 'grpc.publish_port', '9400'
24+
setting 'grpc.host', '["0.0.0.0"]'
25+
setting 'grpc.bind_host', '["0.0.0.0", "::", "10.0.0.1"]'
26+
setting 'grpc.publish_host', '["thisnode.example.com"]'
27+
setting 'grpc.netty.worker_count', '2'
28+
setting 'grpc.netty.max_concurrent_connection_calls', '200'
29+
setting 'grpc.netty.max_connection_age', '500ms'
30+
setting 'grpc.netty.max_connection_idle', '2m'
31+
setting 'grpc.netty.keepalive_timeout', '1s'
2832
```
2933

3034
## Testing

‎plugins/transport-grpc/src/main/java/org/opensearch/plugin/transport/grpc/GrpcPlugin.java

+9-1
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,10 @@
4242
import static org.opensearch.plugin.transport.grpc.Netty4GrpcServerTransport.GRPC_TRANSPORT_SETTING_KEY;
4343
import static org.opensearch.plugin.transport.grpc.Netty4GrpcServerTransport.SETTING_GRPC_BIND_HOST;
4444
import static org.opensearch.plugin.transport.grpc.Netty4GrpcServerTransport.SETTING_GRPC_HOST;
45+
import static org.opensearch.plugin.transport.grpc.Netty4GrpcServerTransport.SETTING_GRPC_KEEPALIVE_TIMEOUT;
46+
import static org.opensearch.plugin.transport.grpc.Netty4GrpcServerTransport.SETTING_GRPC_MAX_CONCURRENT_CONNECTION_CALLS;
47+
import static org.opensearch.plugin.transport.grpc.Netty4GrpcServerTransport.SETTING_GRPC_MAX_CONNECTION_AGE;
48+
import static org.opensearch.plugin.transport.grpc.Netty4GrpcServerTransport.SETTING_GRPC_MAX_CONNECTION_IDLE;
4549
import static org.opensearch.plugin.transport.grpc.Netty4GrpcServerTransport.SETTING_GRPC_PORT;
4650
import static org.opensearch.plugin.transport.grpc.Netty4GrpcServerTransport.SETTING_GRPC_PUBLISH_HOST;
4751
import static org.opensearch.plugin.transport.grpc.Netty4GrpcServerTransport.SETTING_GRPC_PUBLISH_PORT;
@@ -145,12 +149,16 @@ private List<BindableService> registerGRPCServices(BindableService... services)
145149
public List<Setting<?>> getSettings() {
146150
return List.of(
147151
SETTING_GRPC_PORT,
152+
SETTING_GRPC_PUBLISH_PORT,
148153
SETTING_GRPC_SECURE_PORT,
149154
SETTING_GRPC_HOST,
150155
SETTING_GRPC_PUBLISH_HOST,
151156
SETTING_GRPC_BIND_HOST,
152157
SETTING_GRPC_WORKER_COUNT,
153-
SETTING_GRPC_PUBLISH_PORT
158+
SETTING_GRPC_MAX_CONCURRENT_CONNECTION_CALLS,
159+
SETTING_GRPC_MAX_CONNECTION_AGE,
160+
SETTING_GRPC_MAX_CONNECTION_IDLE,
161+
SETTING_GRPC_KEEPALIVE_TIMEOUT
154162
);
155163
}
156164

‎plugins/transport-grpc/src/main/java/org/opensearch/plugin/transport/grpc/Netty4GrpcServerTransport.java

+72-4
Original file line numberDiff line numberDiff line change
@@ -14,10 +14,13 @@
1414
import org.opensearch.common.settings.Setting;
1515
import org.opensearch.common.settings.Settings;
1616
import org.opensearch.common.transport.PortsRange;
17+
import org.opensearch.common.unit.TimeValue;
1718
import org.opensearch.common.util.concurrent.OpenSearchExecutors;
1819
import org.opensearch.core.common.Strings;
1920
import org.opensearch.core.common.transport.BoundTransportAddress;
2021
import org.opensearch.core.common.transport.TransportAddress;
22+
import org.opensearch.core.common.unit.ByteSizeUnit;
23+
import org.opensearch.core.common.unit.ByteSizeValue;
2124
import org.opensearch.plugins.NetworkPlugin;
2225
import org.opensearch.transport.BindTransportException;
2326

@@ -115,6 +118,60 @@ public class Netty4GrpcServerTransport extends NetworkPlugin.AuxTransport {
115118
Setting.Property.NodeScope
116119
);
117120

121+
/**
122+
* Controls the number of allowed simultaneous in flight requests a single client connection may send.
123+
*/
124+
public static final Setting<Integer> SETTING_GRPC_MAX_CONCURRENT_CONNECTION_CALLS = Setting.intSetting(
125+
"grpc.netty.max_concurrent_connection_calls",
126+
100,
127+
1,
128+
Integer.MAX_VALUE,
129+
Setting.Property.NodeScope
130+
);
131+
132+
/**
133+
* Configure maximum inbound message size in bytes.
134+
*/
135+
public static final Setting<ByteSizeValue> SETTING_GRPC_MAX_MSG_SIZE = Setting.byteSizeSetting(
136+
"grpc.netty.max_msg_size",
137+
new ByteSizeValue(10, ByteSizeUnit.MB),
138+
new ByteSizeValue(0, ByteSizeUnit.MB),
139+
new ByteSizeValue(Integer.MAX_VALUE, ByteSizeUnit.BYTES),
140+
Setting.Property.NodeScope
141+
);
142+
143+
/**
144+
* Connections lasting longer than configured age will be gracefully terminated.
145+
* No max connection age by default.
146+
*/
147+
public static final Setting<TimeValue> SETTING_GRPC_MAX_CONNECTION_AGE = Setting.timeSetting(
148+
"grpc.netty.max_connection_age",
149+
new TimeValue(Long.MAX_VALUE),
150+
new TimeValue(0),
151+
Setting.Property.NodeScope
152+
);
153+
154+
/**
155+
* Idle connections lasting longer than configured value will be gracefully terminated.
156+
* No max idle time by default.
157+
*/
158+
public static final Setting<TimeValue> SETTING_GRPC_MAX_CONNECTION_IDLE = Setting.timeSetting(
159+
"grpc.netty.max_connection_idle",
160+
new TimeValue(Long.MAX_VALUE),
161+
new TimeValue(0),
162+
Setting.Property.NodeScope
163+
);
164+
165+
/**
166+
* Timeout for keepalive ping requests of an established connection.
167+
*/
168+
public static final Setting<TimeValue> SETTING_GRPC_KEEPALIVE_TIMEOUT = Setting.timeSetting(
169+
"grpc.netty.keepalive_timeout",
170+
new TimeValue(Long.MAX_VALUE),
171+
new TimeValue(0),
172+
Setting.Property.NodeScope
173+
);
174+
118175
/**
119176
* Port range on which servers bind.
120177
*/
@@ -136,6 +193,11 @@ public class Netty4GrpcServerTransport extends NetworkPlugin.AuxTransport {
136193
private final String[] bindHosts;
137194
private final String[] publishHosts;
138195
private final int nettyEventLoopThreads;
196+
private final long maxInboundMessageSize;
197+
private final long maxConcurrentConnectionCalls;
198+
private final TimeValue maxConnectionAge;
199+
private final TimeValue maxConnectionIdle;
200+
private final TimeValue keepAliveTimeout;
139201
private final CopyOnWriteArrayList<Server> servers = new CopyOnWriteArrayList<>();
140202
private final List<UnaryOperator<NettyServerBuilder>> serverBuilderConfigs = new ArrayList<>();
141203

@@ -153,18 +215,20 @@ public Netty4GrpcServerTransport(Settings settings, List<BindableService> servic
153215
this.settings = Objects.requireNonNull(settings);
154216
this.services = Objects.requireNonNull(services);
155217
this.networkService = Objects.requireNonNull(networkService);
156-
157218
final List<String> grpcBindHost = SETTING_GRPC_BIND_HOST.get(settings);
158219
this.bindHosts = (grpcBindHost.isEmpty() ? NetworkService.GLOBAL_NETWORK_BIND_HOST_SETTING.get(settings) : grpcBindHost).toArray(
159220
Strings.EMPTY_ARRAY
160221
);
161-
162222
final List<String> grpcPublishHost = SETTING_GRPC_PUBLISH_HOST.get(settings);
163223
this.publishHosts = (grpcPublishHost.isEmpty() ? NetworkService.GLOBAL_NETWORK_PUBLISH_HOST_SETTING.get(settings) : grpcPublishHost)
164224
.toArray(Strings.EMPTY_ARRAY);
165-
166225
this.port = SETTING_GRPC_PORT.get(settings);
167226
this.nettyEventLoopThreads = SETTING_GRPC_WORKER_COUNT.get(settings);
227+
this.maxInboundMessageSize = SETTING_GRPC_MAX_MSG_SIZE.get(settings).getBytes();
228+
this.maxConcurrentConnectionCalls = SETTING_GRPC_MAX_CONCURRENT_CONNECTION_CALLS.get(settings);
229+
this.maxConnectionAge = SETTING_GRPC_MAX_CONNECTION_AGE.get(settings);
230+
this.maxConnectionIdle = SETTING_GRPC_MAX_CONNECTION_IDLE.get(settings);
231+
this.keepAliveTimeout = SETTING_GRPC_KEEPALIVE_TIMEOUT.get(settings);
168232
this.portSettingKey = SETTING_GRPC_PORT.getKey();
169233
}
170234

@@ -285,12 +349,16 @@ private TransportAddress bindAddress(InetAddress hostAddress, PortsRange portRan
285349

286350
boolean success = portRange.iterate(portNumber -> {
287351
try {
288-
289352
final InetSocketAddress address = new InetSocketAddress(hostAddress, portNumber);
290353
final NettyServerBuilder serverBuilder = NettyServerBuilder.forAddress(address)
291354
.directExecutor()
292355
.bossEventLoopGroup(eventLoopGroup)
293356
.workerEventLoopGroup(eventLoopGroup)
357+
.maxInboundMessageSize((int) maxInboundMessageSize)
358+
.maxConcurrentCallsPerConnection((int) maxConcurrentConnectionCalls)
359+
.maxConnectionAge(maxConnectionAge.duration(), maxConnectionAge.timeUnit())
360+
.maxConnectionIdle(maxConnectionIdle.duration(), maxConnectionIdle.timeUnit())
361+
.keepAliveTimeout(keepAliveTimeout.duration(), keepAliveTimeout.timeUnit())
294362
.channelType(NioServerSocketChannel.class)
295363
.addService(new HealthStatusManager().getHealthService())
296364
.addService(ProtoReflectionService.newInstance());

‎plugins/transport-grpc/src/test/java/org/opensearch/plugin/transport/grpc/GrpcPluginTests.java

+12-1
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,10 @@
3131
import static org.opensearch.plugin.transport.grpc.Netty4GrpcServerTransport.GRPC_TRANSPORT_SETTING_KEY;
3232
import static org.opensearch.plugin.transport.grpc.Netty4GrpcServerTransport.SETTING_GRPC_BIND_HOST;
3333
import static org.opensearch.plugin.transport.grpc.Netty4GrpcServerTransport.SETTING_GRPC_HOST;
34+
import static org.opensearch.plugin.transport.grpc.Netty4GrpcServerTransport.SETTING_GRPC_KEEPALIVE_TIMEOUT;
35+
import static org.opensearch.plugin.transport.grpc.Netty4GrpcServerTransport.SETTING_GRPC_MAX_CONCURRENT_CONNECTION_CALLS;
36+
import static org.opensearch.plugin.transport.grpc.Netty4GrpcServerTransport.SETTING_GRPC_MAX_CONNECTION_AGE;
37+
import static org.opensearch.plugin.transport.grpc.Netty4GrpcServerTransport.SETTING_GRPC_MAX_CONNECTION_IDLE;
3438
import static org.opensearch.plugin.transport.grpc.Netty4GrpcServerTransport.SETTING_GRPC_PORT;
3539
import static org.opensearch.plugin.transport.grpc.Netty4GrpcServerTransport.SETTING_GRPC_PUBLISH_HOST;
3640
import static org.opensearch.plugin.transport.grpc.Netty4GrpcServerTransport.SETTING_GRPC_PUBLISH_PORT;
@@ -97,9 +101,16 @@ public void testGetSettings() {
97101
assertTrue("SETTING_GRPC_BIND_HOST should be included", settings.contains(SETTING_GRPC_BIND_HOST));
98102
assertTrue("SETTING_GRPC_WORKER_COUNT should be included", settings.contains(SETTING_GRPC_WORKER_COUNT));
99103
assertTrue("SETTING_GRPC_PUBLISH_PORT should be included", settings.contains(SETTING_GRPC_PUBLISH_PORT));
104+
assertTrue(
105+
"SETTING_GRPC_MAX_CONCURRENT_CONNECTION_CALLS should be included",
106+
settings.contains(SETTING_GRPC_MAX_CONCURRENT_CONNECTION_CALLS)
107+
);
108+
assertTrue("SETTING_GRPC_MAX_CONNECTION_AGE should be included", settings.contains(SETTING_GRPC_MAX_CONNECTION_AGE));
109+
assertTrue("SETTING_GRPC_MAX_CONNECTION_IDLE should be included", settings.contains(SETTING_GRPC_MAX_CONNECTION_IDLE));
110+
assertTrue("SETTING_GRPC_KEEPALIVE_TIMEOUT should be included", settings.contains(SETTING_GRPC_KEEPALIVE_TIMEOUT));
100111

101112
// Verify the number of settings
102-
assertEquals("Should return 7 settings", 7, settings.size());
113+
assertEquals("Should return 11 settings", 11, settings.size());
103114
}
104115

105116
public void testGetAuxTransports() {

0 commit comments

Comments
 (0)