-
Notifications
You must be signed in to change notification settings - Fork 137
New gRPC channel manager #1400
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Draft
rcameronm
wants to merge
7
commits into
NationalSecurityAgency:main
Choose a base branch
from
rcameronm:Create-different-gRPC-channel-managers
base: main
Could not load branches
Branch not found: {{ refName }}
Loading
Could not load tags
Nothing to show
Loading
Are you sure you want to change the base?
Some commits from the old base branch may be removed from the timeline,
and old review comments may become outdated.
Draft
New gRPC channel manager #1400
Changes from all commits
Commits
Show all changes
7 commits
Select commit
Hold shift + click to select a range
586b900
New gRPC channel managers
rcameronm 471b621
Use modern gRPC channel builder API
rcameronm 451bb4e
Overridable gRPC channel credential creation method
rcameronm 47d40a6
Endpoint URI validation
rcameronm 7f505c2
Updated channel validation exception messages
rcameronm 5ddda8b
Align target creation logic with existing logic
rcameronm 3a1b340
replace reflection with a factory
fbruton File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
165 changes: 165 additions & 0 deletions
165
src/main/java/emissary/grpc/channel/ChannelManager.java
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,165 @@ | ||
| package emissary.grpc.channel; | ||
|
|
||
| import emissary.config.Configurator; | ||
|
|
||
| import io.grpc.ChannelCredentials; | ||
| import io.grpc.Grpc; | ||
| import io.grpc.ManagedChannel; | ||
| import org.apache.commons.lang3.StringUtils; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| import java.util.concurrent.TimeUnit; | ||
|
|
||
| /** | ||
| * Abstract base class for managing gRPC {@link io.grpc.ManagedChannel} connections. | ||
| * <p> | ||
| * Configuration Keys: | ||
| * <ul> | ||
| * <li>{@code GRPC_CHANNEL_KEEP_ALIVE_MILLIS} - Time to wait before sending a ping on idle, default={@code 60000L}</li> | ||
| * <li>{@code GRPC_CHANNEL_KEEP_ALIVE_TIMEOUT_MILLIS} - Timeout for receiving ping ACKs, default={@code 30000L}</li> | ||
| * <li>{@code GRPC_CHANNEL_KEEP_ALIVE_WITHOUT_CALLS} - Send pings even when no RPCs are active if {@code true}, | ||
| * default={@code false}</li> | ||
| * <li>{@code GRPC_CHANNEL_LOAD_BALANCING_POLICY} - gRPC load balancing policy, default={@code "round_robin"}</li> | ||
| * <li>{@code GRPC_CHANNEL_MAX_INBOUND_MESSAGE_BYTE_SIZE} - Max inbound gRPC message size, default={@code 4194304}</li> | ||
| * <li>{@code GRPC_CHANNEL_MAX_INBOUND_METADATA_BYTE_SIZE} - Max inbound gRPC metadata size, default={@code 8192}</li> | ||
| * </ul> | ||
| */ | ||
| public abstract class ChannelManager implements AutoCloseable { | ||
| public static final String GRPC_CHANNEL_PREFIX = "GRPC_CHANNEL_"; | ||
| public static final String KEEP_ALIVE_MILLIS = GRPC_CHANNEL_PREFIX + "KEEP_ALIVE_MILLIS"; | ||
| public static final String KEEP_ALIVE_TIMEOUT_MILLIS = GRPC_CHANNEL_PREFIX + "KEEP_ALIVE_TIMEOUT_MILLIS"; | ||
| public static final String KEEP_ALIVE_WITHOUT_CALLS = GRPC_CHANNEL_PREFIX + "KEEP_ALIVE_WITHOUT_CALLS"; | ||
| public static final String LOAD_BALANCING_POLICY = GRPC_CHANNEL_PREFIX + "LOAD_BALANCING_POLICY"; | ||
| public static final String MAX_INBOUND_MESSAGE_BYTE_SIZE = GRPC_CHANNEL_PREFIX + "MAX_INBOUND_MESSAGE_BYTE_SIZE"; | ||
| public static final String MAX_INBOUND_METADATA_BYTE_SIZE = GRPC_CHANNEL_PREFIX + "MAX_INBOUND_METADATA_BYTE_SIZE"; | ||
|
|
||
| protected static final int MAX_PORT_NUMBER = 0xFFFF; | ||
|
|
||
| protected final Logger logger; | ||
|
|
||
| protected final String host; | ||
| protected final int port; | ||
| protected final String target; | ||
| protected final long keepAliveMillis; | ||
| protected final long keepAliveTimeoutMillis; | ||
| protected final boolean keepAliveWithoutCalls; | ||
| protected final int maxInboundMessageByteSize; | ||
| protected final int maxInboundMetadataByteSize; | ||
| protected final String loadBalancingPolicy; | ||
| protected final ChannelCredentials channelCredentials; | ||
|
|
||
| /** | ||
| * Constructs a new gRPC connection manager using the provided host, port, and configuration. Initializes gRPC channel | ||
| * properties from the given configuration source. | ||
| * | ||
| * @param host gRPC service hostname or DNS target | ||
| * @param port gRPC service port | ||
| * @param configG configuration provider for channel parameters | ||
| * @param credentials transport-level credentials for the gRPC server | ||
| * @see ChannelManager | ||
| * @see <a href="https://docs.microsoft.com/en-us/aspnet/core/grpc/performance?view=aspnetcore-5.0">Source</a> for | ||
| * default gRPC configurations. | ||
| */ | ||
| protected ChannelManager(String host, int port, Configurator configG, ChannelCredentials credentials) { | ||
| this.logger = LoggerFactory.getLogger(this.getClass().getName()); | ||
|
|
||
| this.host = host; | ||
| this.port = port; | ||
| this.target = createTarget(); | ||
|
|
||
| // How often (in milliseconds) to send pings when the connection is idle | ||
| this.keepAliveMillis = configG.findLongEntry(KEEP_ALIVE_MILLIS, 60000L); | ||
|
|
||
| // Time to wait (in milliseconds) for a ping ACK before closing the connection | ||
| this.keepAliveTimeoutMillis = configG.findLongEntry(KEEP_ALIVE_TIMEOUT_MILLIS, 30000L); | ||
|
|
||
| // Whether to send pings when no RPCs are active | ||
| // Note: Seme gRPC services have this set to false and will be noisy if not adjusted | ||
| this.keepAliveWithoutCalls = configG.findBooleanEntry(KEEP_ALIVE_WITHOUT_CALLS, false); | ||
|
|
||
| // Specifies how the client chooses between multiple backend addresses | ||
| // e.g. "pick_first" uses the first address only, "round_robin" cycles through all of them for client-side balancing | ||
| this.loadBalancingPolicy = configG.findObjectEntry( | ||
| LOAD_BALANCING_POLICY, LoadBalancingPolicy::valueOf, LoadBalancingPolicy.ROUND_ROBIN).toString(); | ||
|
|
||
| // Max size (in bytes) for incoming messages and message metadata from the server | ||
| this.maxInboundMessageByteSize = configG.findIntEntry(MAX_INBOUND_MESSAGE_BYTE_SIZE, 4 << 20); // 4 MiB | ||
| this.maxInboundMetadataByteSize = configG.findIntEntry(MAX_INBOUND_METADATA_BYTE_SIZE, 8 << 10); // 8 KiB | ||
|
|
||
| this.channelCredentials = credentials; | ||
| } | ||
|
|
||
| private String createTarget() { | ||
| if (StringUtils.isEmpty(host)) { | ||
| throw new IllegalArgumentException("Missing required gRPC host configuration"); | ||
| } | ||
| if (port <= 0 || port > MAX_PORT_NUMBER) { | ||
| throw new IllegalArgumentException( | ||
| String.format("Port \"%d\" is outside valid range [1, %d]", port, MAX_PORT_NUMBER)); | ||
| } | ||
| return host + ":" + port; | ||
| } | ||
|
|
||
| public static ChannelManager ofSubClass( | ||
| String className, String host, int port, Configurator configG, ChannelCredentials credentials) { | ||
| try { | ||
| return ofSubClass(Class.forName(className).asSubclass(ChannelManager.class), host, port, configG, credentials); | ||
| } catch (ClassNotFoundException e) { | ||
| throw new IllegalArgumentException("Cannot find class: " + className, e); | ||
| } | ||
| } | ||
|
|
||
| public static ChannelManager ofSubClass( | ||
| Class<? extends ChannelManager> clz, String host, int port, Configurator configG, ChannelCredentials credentials) { | ||
| return ChannelManagerRegistry.ofSubClass(clz, host, port, configG, credentials); | ||
| } | ||
|
|
||
| /** | ||
| * Creates a new {@link ManagedChannel} instance configured with the current factory settings. | ||
| * | ||
| * @return a new gRPC channel | ||
| */ | ||
| protected final ManagedChannel create() { | ||
| return Grpc.newChannelBuilder(this.target, this.channelCredentials) | ||
| .keepAliveTime(this.keepAliveMillis, TimeUnit.MILLISECONDS) | ||
| .keepAliveTimeout(this.keepAliveTimeoutMillis, TimeUnit.MILLISECONDS) | ||
| .keepAliveWithoutCalls(this.keepAliveWithoutCalls) | ||
| .defaultLoadBalancingPolicy(this.loadBalancingPolicy) | ||
| .maxInboundMessageSize(this.maxInboundMessageByteSize) | ||
| .maxInboundMetadataSize(this.maxInboundMetadataByteSize) | ||
| .build(); | ||
| } | ||
|
|
||
| public abstract ManagedChannel acquire(); | ||
|
|
||
| public abstract void release(ManagedChannel channel); | ||
|
|
||
| public abstract void shutdown(ManagedChannel channel); | ||
|
|
||
| @Override | ||
| public abstract void close(); | ||
|
|
||
| public String getHost() { | ||
| return host; | ||
| } | ||
|
|
||
| public int getPort() { | ||
| return port; | ||
| } | ||
|
|
||
| public enum LoadBalancingPolicy { | ||
| ROUND_ROBIN("round_robin"), PICK_FIRST("pick_first"); | ||
|
|
||
| private final String policy; | ||
|
|
||
| LoadBalancingPolicy(String policy) { | ||
| this.policy = policy; | ||
| } | ||
|
|
||
| @Override | ||
| public String toString() { | ||
| return policy; | ||
| } | ||
| } | ||
| } | ||
10 changes: 10 additions & 0 deletions
10
src/main/java/emissary/grpc/channel/ChannelManagerFactory.java
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,10 @@ | ||
| package emissary.grpc.channel; | ||
|
|
||
| import emissary.config.Configurator; | ||
|
|
||
| import io.grpc.ChannelCredentials; | ||
|
|
||
| @FunctionalInterface | ||
| public interface ChannelManagerFactory { | ||
| ChannelManager create(String host, int port, Configurator configG, ChannelCredentials credentials); | ||
| } |
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Attached is a patch to replace reflection with a factory design pattern. Any class subclass without the proper method signature will throw a compile error.
0001-replace-reflection-with-a-factory.patch
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I've been using this reflection pattern for awhile. Should I make some new tickets to change them to this factory design pattern as well?
Uh oh!
There was an error while loading. Please reload this page.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I recommend it. It will improve maintainability.