Allow sharing the same gRPC channel between clients
This change introduces a refactoring of the gRPC protocol subsystem that
allows the creation of a gRPC chanel independently of the client, while
allowing multiple clients to share the same channel (e.g. as in Stratum
where we use 3 clients).
Moreover, we refactor the P4RuntimeClient API to support multiple
P4Runtime-internal device ID using the same client. While before the
client was associated to one of such ID.
Finally, we provide an abstract implementation for gRPC-based driver
behaviors, reducing code duplication in P4Runtime, gNMI and gNOI drivers.
Change-Id: I1a46352bbbef1e0d24042f169ae8ba580202944f
diff --git a/protocols/grpc/BUILD b/protocols/grpc/BUILD
index f82e1f0..3dd1084 100644
--- a/protocols/grpc/BUILD
+++ b/protocols/grpc/BUILD
@@ -3,9 +3,9 @@
load("//tools/build/bazel:variables.bzl", "PROTOBUF_VERSION")
BUNDLES = [
- "//protocols/grpc/proto:onos-protocols-grpc-proto",
"//protocols/grpc/api:onos-protocols-grpc-api",
"//protocols/grpc/ctl:onos-protocols-grpc-ctl",
+ "//protocols/grpc/utils:onos-protocols-grpc-utils",
# gRPC dependencies (with patched core)
":grpc-core",
":grpc-stub",
diff --git a/protocols/grpc/api/src/main/java/org/onosproject/grpc/api/GrpcChannelController.java b/protocols/grpc/api/src/main/java/org/onosproject/grpc/api/GrpcChannelController.java
index ad4324b..42a9248 100644
--- a/protocols/grpc/api/src/main/java/org/onosproject/grpc/api/GrpcChannelController.java
+++ b/protocols/grpc/api/src/main/java/org/onosproject/grpc/api/GrpcChannelController.java
@@ -19,75 +19,84 @@
import com.google.common.annotations.Beta;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
-import io.grpc.StatusRuntimeException;
-import java.util.Map;
+import java.net.URI;
import java.util.Optional;
-import java.util.concurrent.CompletableFuture;
/**
- * Abstraction of a gRPC controller that stores and manages gRPC channels.
+ * Abstraction of a gRPC controller that creates, stores, and manages gRPC
+ * channels.
*/
@Beta
public interface GrpcChannelController {
- int CONNECTION_TIMEOUT_SECONDS = 20;
+ /**
+ * Creates a gRPC managed channel to the server identified by the given
+ * channel URI. The channel is created using the information contained in the
+ * URI, as such, the URI is expected to have absolute server-based form,
+ * where the scheme can be either {@code grpc:} or {@code grpcs:}, to
+ * indicated respectively a plaintext or secure channel.
+ * <p>
+ * Example of valid URIs are: <pre> {@code
+ * grpc://10.0.0.1:50001
+ * grpcs://10.0.0.1:50001
+ * grpcs://myserver.local:50001
+ * }</pre>
+ * <p>
+ * This method creates and stores the channel instance associating it to the
+ * passed URI, but it does not make any attempt to connect the channel or
+ * verify server reachability.
+ * <p>
+ * If another channel with the same URI already exists, an {@link
+ * IllegalArgumentException} is thrown. To create multiple channels to the
+ * same server-port combination, URI file or query parameters can be used.
+ * For example: <pre> {@code
+ * grpc://10.0.0.1:50001/foo
+ * grpc://10.0.0.1:50001/bar
+ * grpc://10.0.0.1:50001/bar?param=1
+ * grpc://10.0.0.1:50001/bar?param=2
+ * }</pre>
+ * <p>
+ * When creating secure channels (i.e., {@code grpcs:)}, the current
+ * implementation provides encryption but not authentication, any server
+ * certificate, even if insecure, will be accepted.
+ *
+ * @param channelUri channel URI
+ * @return the managed channel created
+ * @throws IllegalArgumentException if a channel with the same channel URI
+ * already exists
+ */
+ ManagedChannel create(URI channelUri);
/**
- * Creates a gRPC managed channel from the given builder and opens the
- * connection. If the connection is successful, returns the managed channel
- * object and stores the channel internally, associated with the given
- * channel ID.
- * <p>
- * This method blocks until the channel is open or a timeout expires. By
- * default the timeout is {@link #CONNECTION_TIMEOUT_SECONDS} seconds. If
- * the timeout expires, a {@link StatusRuntimeException} is thrown. If
- * another channel with the same ID already exists, an {@link
- * IllegalArgumentException} is thrown.
+ * Similar to {@link #create(URI)} but does not create the chanel instance,
+ * instead, it uses the given channel builder to create it. As such, there
+ * is no requirement on the format of the URI, any URI can be used. The
+ * implementation might modify the passed builder for purposes specific to
+ * this controller, such as to enable gRPC message logging.
*
- * @param channelId ID of the channel
+ * @param channelUri URI identifying the channel
* @param channelBuilder builder of the managed channel
* @return the managed channel created
- * @throws StatusRuntimeException if the channel cannot be opened
* @throws IllegalArgumentException if a channel with the same ID already
* exists
*/
- ManagedChannel connectChannel(GrpcChannelId channelId,
- ManagedChannelBuilder<?> channelBuilder);
+ ManagedChannel create(URI channelUri,
+ ManagedChannelBuilder<?> channelBuilder);
/**
- * Closes the gRPC managed channel (i.e., disconnects from the gRPC server)
- * and removes any internal state associated to it.
+ * Closes and destroys the gRPC channel associated to the given URI and
+ * removes any internal state associated to it.
*
- * @param channelId ID of the channel to remove
+ * @param channelUri URI of the channel to remove
*/
- void disconnectChannel(GrpcChannelId channelId);
+ void destroy(URI channelUri);
/**
- * Returns all channels known by this controller, each one mapped to the ID
- * passed at creation time.
+ * If present, returns the channel associated with the given URI.
*
- * @return map of all the channels with their ID as stored in this
- * controller
- */
- Map<GrpcChannelId, ManagedChannel> getChannels();
-
- /**
- * If present, returns the channel associated with the given ID.
- *
- * @param channelId channel ID
+ * @param channelUri channel URI
* @return optional channel
*/
- Optional<ManagedChannel> getChannel(GrpcChannelId channelId);
-
- /**
- * Probes the server at the endpoint of the given channel. Returns true if
- * the server responded to the probe, false otherwise or if the channel does
- * not exist.
- *
- * @param channelId channel ID
- * @return completable future eventually true if the gRPC server responded
- * to the probe; false otherwise
- */
- CompletableFuture<Boolean> probeChannel(GrpcChannelId channelId);
+ Optional<ManagedChannel> get(URI channelUri);
}
diff --git a/protocols/grpc/api/src/main/java/org/onosproject/grpc/api/GrpcChannelId.java b/protocols/grpc/api/src/main/java/org/onosproject/grpc/api/GrpcChannelId.java
deleted file mode 100644
index 6db331a..0000000
--- a/protocols/grpc/api/src/main/java/org/onosproject/grpc/api/GrpcChannelId.java
+++ /dev/null
@@ -1,41 +0,0 @@
-/*
- * Copyright 2017-present Open Networking Foundation
- *
- * Licensed 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.onosproject.grpc.api;
-
-import com.google.common.annotations.Beta;
-import org.onlab.util.Identifier;
-
-/**
- * gRPC channel identifier, unique in the scope of an ONOS node.
- */
-@Beta
-public final class GrpcChannelId extends Identifier<String> {
-
- private GrpcChannelId(String channelName) {
- super(channelName);
- }
-
- /**
- * Instantiates a new channel ID.
- *
- * @param channelName name of the channel
- * @return channel ID
- */
- public static GrpcChannelId of(String channelName) {
- return new GrpcChannelId(channelName);
- }
-}
diff --git a/protocols/grpc/api/src/main/java/org/onosproject/grpc/api/GrpcClientController.java b/protocols/grpc/api/src/main/java/org/onosproject/grpc/api/GrpcClientController.java
index b47d499..3edb36d 100644
--- a/protocols/grpc/api/src/main/java/org/onosproject/grpc/api/GrpcClientController.java
+++ b/protocols/grpc/api/src/main/java/org/onosproject/grpc/api/GrpcClientController.java
@@ -17,6 +17,7 @@
package org.onosproject.grpc.api;
import com.google.common.annotations.Beta;
+import io.grpc.ManagedChannel;
import org.onosproject.net.DeviceId;
import org.onosproject.net.device.DeviceAgentListener;
import org.onosproject.net.provider.ProviderId;
@@ -24,76 +25,54 @@
/**
* Abstraction of controller that manages gRPC clients.
*
- * @param <K> the gRPC client key
* @param <C> the gRPC client type
*/
@Beta
-public interface GrpcClientController<K extends GrpcClientKey, C extends GrpcClient> {
+public interface GrpcClientController<C extends GrpcClient> {
/**
- * Instantiates a new client to operate on a gRPC server identified by the
- * given information. As a result of this method, a client can be later
- * obtained by invoking {@link #getClient(DeviceId)}.
+ * Instantiates a new client to operate on the given gRPC channel. Returns
+ * true if the client was created successfully, false otherwise. Clients
+ * are identified by device IDs and once created they can be obtained by
+ * invoking {@link #get(DeviceId)}.
* <p>
- * Upon creation, a connection to the server is automatically started, which
- * blocks execution. If the connection is successful, the client is created
- * and this method returns true, otherwise (e.g., socket error) any state
- * associated with this client is destroyed and returns false.
- * <p>
- * Only one client can exist for the same device ID. Calls to this method
- * are idempotent fot the same client key, i.e. returns true if such client
- * already exists. Otherwise, if a client for the same device ID but
- * different client key already exists, throws an exception.
+ * Only one client can exist for the same device ID. If a client for the
+ * given device ID already exists, throws an exception.
*
- * @param clientKey the client key
- * @return true if the client was created and the channel to the server is
- * open; false otherwise
- * @throws IllegalArgumentException if a client for the same device ID but
- * different client key already exists.
+ * @param deviceId device ID
+ * @param channel gRPC managed channel
+ * @return true if the client was created, false otherwise
+ * @throws IllegalArgumentException if a client for the same device ID
+ * already exists.
*/
- boolean createClient(K clientKey);
+ boolean create(DeviceId deviceId, ManagedChannel channel);
/**
- * Returns the gRPC client previously created for the given device, or null
- * if such client does not exist.
- *
- * @param deviceId the device identifier
- * @return the gRPC client of the device if exists; null otherwise
- */
- C getClient(DeviceId deviceId);
-
- /**
- * Returns the gRPC client previously created for the given client key, or
+ * Returns the gRPC client previously created for the given device ID, or
* null if such client does not exist.
*
- * @param clientKey client key
+ * @param deviceId the device ID
* @return the gRPC client of the device if exists; null otherwise
*/
- C getClient(K clientKey);
+ C get(DeviceId deviceId);
/**
* Removes the gRPC client for the given device and any gRPC channel state
* associated to it. If no client exists for the given device, the result is
* a no-op.
*
- * @param deviceId the device identifier
+ * @param deviceId the device ID
*/
- void removeClient(DeviceId deviceId);
+ void remove(DeviceId deviceId);
/**
- * Similar to {@link #removeClient(DeviceId)} but uses the client key to
- * identify the client to remove.
+ * Adds a listener for device agent events for the given provider. If a
+ * listener already exists for the given device ID and provider ID, then it
+ * will be replaced by the new one.
*
- * @param clientKey the client key
- */
- void removeClient(K clientKey);
-
- /**
- * Adds a listener for device agent events for the given provider.
- *
- * @param deviceId device identifier
+ * @param deviceId device ID
* @param providerId provider ID
- * @param listener the device agent listener
+ * @param listener the device agent listener
*/
void addDeviceAgentListener(DeviceId deviceId, ProviderId providerId,
DeviceAgentListener listener);
@@ -102,7 +81,7 @@
* Removes the listener for device agent events that was previously
* registered for the given provider.
*
- * @param deviceId device identifier
+ * @param deviceId device ID
* @param providerId the provider ID
*/
void removeDeviceAgentListener(DeviceId deviceId, ProviderId providerId);
diff --git a/protocols/grpc/api/src/main/java/org/onosproject/grpc/api/GrpcClientKey.java b/protocols/grpc/api/src/main/java/org/onosproject/grpc/api/GrpcClientKey.java
deleted file mode 100644
index d1d0b0f..0000000
--- a/protocols/grpc/api/src/main/java/org/onosproject/grpc/api/GrpcClientKey.java
+++ /dev/null
@@ -1,126 +0,0 @@
-/*
- * Copyright 2018-present Open Networking Foundation
- *
- * Licensed 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.onosproject.grpc.api;
-
-import com.google.common.annotations.Beta;
-import com.google.common.base.Objects;
-import org.onosproject.net.DeviceId;
-
-import java.net.URI;
-
-import static com.google.common.base.Preconditions.checkArgument;
-import static com.google.common.base.Preconditions.checkNotNull;
-import static com.google.common.base.Strings.isNullOrEmpty;
-import static java.lang.String.format;
-
-/**
- * Key that uniquely identifies a gRPC client.
- */
-@Beta
-public class GrpcClientKey {
-
- private static final String GRPC = "grpc";
- private static final String GRPCS = "grpcs";
-
- private final String serviceName;
- private final DeviceId deviceId;
- private final URI serverUri;
-
- /**
- * Creates a new client key.
- *
- * @param serviceName gRPC service name of the client
- * @param deviceId ONOS device ID
- * @param serverUri gRPC server URI
- */
- public GrpcClientKey(String serviceName, DeviceId deviceId, URI serverUri) {
- checkNotNull(serviceName);
- checkNotNull(deviceId);
- checkNotNull(serverUri);
- checkArgument(!serviceName.isEmpty(),
- "Service name can not be null");
- checkArgument(serverUri.getScheme().equals(GRPC)
- || serverUri.getScheme().equals(GRPCS),
- format("Server URI scheme must be %s or %s", GRPC, GRPCS));
- checkArgument(!isNullOrEmpty(serverUri.getHost()),
- "Server host address should not be empty");
- checkArgument(serverUri.getPort() > 0 && serverUri.getPort() <= 65535, "Invalid server port");
- this.serviceName = serviceName;
- this.deviceId = deviceId;
- this.serverUri = serverUri;
- }
-
- /**
- * Gets the gRPC service name of the client.
- *
- * @return the service name
- */
- public String serviceName() {
- return serviceName;
- }
-
- /**
- * Gets the device ID.
- *
- * @return the device ID
- */
- public DeviceId deviceId() {
- return deviceId;
- }
-
- /**
- * Returns the gRPC server URI.
- *
- * @return the gRPC server URI.
- */
- public URI serveUri() {
- return serverUri;
- }
-
- /**
- * Returns true if the client requires TLS/SSL, false otherwise.
- *
- * @return boolean
- */
- public boolean requiresSecureChannel() {
- return serverUri.getScheme().equals(GRPCS);
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
- GrpcClientKey that = (GrpcClientKey) o;
- return Objects.equal(serviceName, that.serviceName) &&
- Objects.equal(deviceId, that.deviceId) &&
- Objects.equal(serverUri, that.serverUri);
- }
-
- @Override
- public int hashCode() {
- return Objects.hashCode(serviceName, deviceId, serverUri);
- }
-
- @Override
- public String toString() {
- return format("%s/%s@%s", deviceId, serviceName, serverUri);
- }
-}
diff --git a/protocols/grpc/ctl/BUILD b/protocols/grpc/ctl/BUILD
index a467398..fa10ff3 100644
--- a/protocols/grpc/ctl/BUILD
+++ b/protocols/grpc/ctl/BUILD
@@ -1,6 +1,5 @@
COMPILE_DEPS = CORE_DEPS + [
"//protocols/grpc/api:onos-protocols-grpc-api",
- "//protocols/grpc/proto:onos-protocols-grpc-proto",
"@io_grpc_grpc_java//core",
"@io_grpc_grpc_java//netty",
"@io_grpc_grpc_java//protobuf-lite",
diff --git a/protocols/grpc/ctl/src/main/java/org/onosproject/grpc/ctl/AbstractGrpcClient.java b/protocols/grpc/ctl/src/main/java/org/onosproject/grpc/ctl/AbstractGrpcClient.java
index befa334..933c7ce 100644
--- a/protocols/grpc/ctl/src/main/java/org/onosproject/grpc/ctl/AbstractGrpcClient.java
+++ b/protocols/grpc/ctl/src/main/java/org/onosproject/grpc/ctl/AbstractGrpcClient.java
@@ -21,7 +21,6 @@
import io.grpc.ManagedChannel;
import io.grpc.StatusRuntimeException;
import org.onosproject.grpc.api.GrpcClient;
-import org.onosproject.grpc.api.GrpcClientKey;
import org.onosproject.net.DeviceId;
import org.onosproject.net.device.DeviceAgentEvent;
import org.slf4j.Logger;
@@ -49,25 +48,26 @@
private final AtomicBoolean channelOpen = new AtomicBoolean(false);
/**
- * Creates an new client for the given key and channel. Setting persistent
- * to true avoids the gRPC channel to stay IDLE. The controller instance is
- * needed to propagate channel events.
+ * Creates an new client for the given device and channel. Setting
+ * persistent to true avoids the gRPC channel to go {@link
+ * ConnectivityState#IDLE}. The controller instance is needed to propagate
+ * channel events.
*
- * @param clientKey client key
+ * @param deviceId device ID
* @param channel channel
* @param persistent true if the gRPC should never stay IDLE
* @param controller controller
*/
- protected AbstractGrpcClient(GrpcClientKey clientKey, ManagedChannel channel,
+ protected AbstractGrpcClient(DeviceId deviceId, ManagedChannel channel,
boolean persistent, AbstractGrpcClientController controller) {
- checkNotNull(clientKey);
+ checkNotNull(deviceId);
checkNotNull(channel);
- this.deviceId = clientKey.deviceId();
+ this.deviceId = deviceId;
this.channel = channel;
this.persistent = persistent;
this.controller = controller;
- setChannelCallback(clientKey.deviceId(), channel, ConnectivityState.CONNECTING);
+ setChannelCallback(ConnectivityState.CONNECTING);
}
@Override
@@ -94,7 +94,7 @@
"ignoring request to shutdown for {}...", deviceId);
return;
}
- log.warn("Shutting down client for {}...", deviceId);
+ log.debug("Shutting down client for {}...", deviceId);
cancellableContext.cancel(new InterruptedException(
"Requested client shutdown"));
}
@@ -140,84 +140,70 @@
opDescription, deviceId), throwable);
}
- private void setChannelCallback(DeviceId deviceId, ManagedChannel channel,
- ConnectivityState sourceState) {
+ private void setChannelCallback(ConnectivityState sourceState) {
if (log.isTraceEnabled()) {
log.trace("Setting channel callback for {} with source state {}...",
deviceId, sourceState);
}
channel.notifyWhenStateChanged(
- sourceState, new ChannelConnectivityCallback(deviceId, channel));
+ sourceState, this::channelStateCallback);
}
/**
- * Runnable task invoked at each change of the channel connectivity state.
- * New callbacks are created as long as the channel is not shut down.
+ * Invoked at each change of the channel connectivity state. New callbacks
+ * are created as long as the channel is not shut down.
*/
- private final class ChannelConnectivityCallback implements Runnable {
-
- private final DeviceId deviceId;
- private final ManagedChannel channel;
-
- private ChannelConnectivityCallback(
- DeviceId deviceId, ManagedChannel channel) {
- this.deviceId = deviceId;
- this.channel = channel;
+ private void channelStateCallback() {
+ final ConnectivityState newState = channel.getState(false);
+ final DeviceAgentEvent.Type eventType;
+ switch (newState) {
+ // On gRPC connectivity states:
+ // https://github.com/grpc/grpc/blob/master/doc/connectivity-semantics-and-api.md
+ case READY:
+ eventType = DeviceAgentEvent.Type.CHANNEL_OPEN;
+ break;
+ case TRANSIENT_FAILURE:
+ eventType = DeviceAgentEvent.Type.CHANNEL_ERROR;
+ break;
+ case SHUTDOWN:
+ eventType = DeviceAgentEvent.Type.CHANNEL_CLOSED;
+ break;
+ case IDLE:
+ // IDLE and CONNECTING are transient states that will
+ // eventually move to READY or TRANSIENT_FAILURE. Do not
+ // generate an event for now.
+ if (persistent) {
+ log.debug("Forcing channel for {} to exist state IDLE...", deviceId);
+ channel.getState(true);
+ }
+ eventType = null;
+ break;
+ case CONNECTING:
+ eventType = null;
+ break;
+ default:
+ log.error("Unrecognized connectivity state {}", newState);
+ eventType = null;
}
- @Override
- public void run() {
- final ConnectivityState newState = channel.getState(false);
- final DeviceAgentEvent.Type eventType;
- switch (newState) {
- // On gRPC connectivity states:
- // https://github.com/grpc/grpc/blob/master/doc/connectivity-semantics-and-api.md
- case READY:
- eventType = DeviceAgentEvent.Type.CHANNEL_OPEN;
- break;
- case TRANSIENT_FAILURE:
- eventType = DeviceAgentEvent.Type.CHANNEL_ERROR;
- break;
- case SHUTDOWN:
- eventType = DeviceAgentEvent.Type.CHANNEL_CLOSED;
- break;
- case IDLE:
- // IDLE and CONNECTING are transient states that will
- // eventually move to READY or TRANSIENT_FAILURE. Do not
- // generate an event for now.
- if (persistent) {
- log.debug("Forcing channel for {} to exist state IDLE...", deviceId);
- channel.getState(true);
- }
- eventType = null;
- break;
- case CONNECTING:
- eventType = null;
- break;
- default:
- log.error("Unrecognized connectivity state {}", newState);
- eventType = null;
- }
+ if (log.isTraceEnabled()) {
+ log.trace("Detected channel connectivity change for {}, new state is {}",
+ deviceId, newState);
+ }
- if (log.isTraceEnabled()) {
- log.trace("Detected channel connectivity change for {}, new state is {}",
- deviceId, newState);
+ if (eventType != null) {
+ // Avoid sending consecutive duplicate events.
+ final boolean present = eventType == DeviceAgentEvent.Type.CHANNEL_OPEN;
+ final boolean past = channelOpen.getAndSet(present);
+ if (present != past) {
+ log.debug("Notifying event {} for {}", eventType, deviceId);
+ controller.postEvent(new DeviceAgentEvent(eventType, deviceId));
}
+ }
- if (eventType != null) {
- // Avoid sending consecutive duplicate events.
- final boolean present = eventType == DeviceAgentEvent.Type.CHANNEL_OPEN;
- final boolean past = channelOpen.getAndSet(present);
- if (present != past) {
- log.debug("Notifying event {} for {}", eventType, deviceId);
- controller.postEvent(new DeviceAgentEvent(eventType, deviceId));
- }
- }
-
- if (newState != ConnectivityState.SHUTDOWN) {
- // Channels never leave SHUTDOWN state, no need for a new callback.
- setChannelCallback(deviceId, channel, newState);
- }
+ if (newState != ConnectivityState.SHUTDOWN) {
+ // Channels never leave SHUTDOWN state, no need for a new callback.
+ setChannelCallback(newState);
}
}
}
diff --git a/protocols/grpc/ctl/src/main/java/org/onosproject/grpc/ctl/AbstractGrpcClientController.java b/protocols/grpc/ctl/src/main/java/org/onosproject/grpc/ctl/AbstractGrpcClientController.java
index 3c38209..9fb3625 100644
--- a/protocols/grpc/ctl/src/main/java/org/onosproject/grpc/ctl/AbstractGrpcClientController.java
+++ b/protocols/grpc/ctl/src/main/java/org/onosproject/grpc/ctl/AbstractGrpcClientController.java
@@ -19,29 +19,19 @@
import com.google.common.collect.Maps;
import com.google.common.util.concurrent.Striped;
import io.grpc.ManagedChannel;
-import io.grpc.netty.GrpcSslContexts;
-import io.grpc.netty.NettyChannelBuilder;
-import io.netty.handler.ssl.SslContext;
-import io.netty.handler.ssl.util.InsecureTrustManagerFactory;
import org.onosproject.event.AbstractListenerManager;
import org.onosproject.event.Event;
import org.onosproject.event.EventListener;
-import org.onosproject.grpc.api.GrpcChannelController;
-import org.onosproject.grpc.api.GrpcChannelId;
import org.onosproject.grpc.api.GrpcClient;
import org.onosproject.grpc.api.GrpcClientController;
-import org.onosproject.grpc.api.GrpcClientKey;
import org.onosproject.net.DeviceId;
import org.onosproject.net.device.DeviceAgentEvent;
import org.onosproject.net.device.DeviceAgentListener;
import org.onosproject.net.provider.ProviderId;
import org.osgi.service.component.annotations.Activate;
import org.osgi.service.component.annotations.Deactivate;
-import org.osgi.service.component.annotations.Reference;
-import org.osgi.service.component.annotations.ReferenceCardinality;
import org.slf4j.Logger;
-import javax.net.ssl.SSLException;
import java.util.Map;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.locks.Lock;
@@ -52,40 +42,35 @@
import static org.slf4j.LoggerFactory.getLogger;
/**
- * Abstract class of a gRPC based client controller for specific gRPC client
- * which provides basic gRPC client management and thread safe mechanism.
+ * Abstract class of a controller for gRPC clients which provides means to
+ * create clients, associate device agent listeners to them and register other
+ * event listeners.
*
* @param <C> the gRPC client type
- * @param <K> the key type of the gRPC client
* @param <E> the event type of the gRPC client
* @param <L> the event listener of event {@link E}
*/
public abstract class AbstractGrpcClientController
- <K extends GrpcClientKey, C extends GrpcClient, E extends Event, L extends EventListener<E>>
+ <C extends GrpcClient, E extends Event, L extends EventListener<E>>
extends AbstractListenerManager<E, L>
- implements GrpcClientController<K, C> {
+ implements GrpcClientController<C> {
/**
* The default max inbound message size (MB).
*/
- private static final int DEFAULT_MAX_INBOUND_MSG_SIZE = 256; // Megabytes.
- private static final int MEGABYTES = 1024 * 1024;
private static final int DEFAULT_DEVICE_LOCK_SIZE = 30;
private final Logger log = getLogger(getClass());
- private final Map<DeviceId, K> clientKeys = Maps.newHashMap();
- private final Map<K, C> clients = Maps.newHashMap();
- private final Map<DeviceId, GrpcChannelId> channelIds = Maps.newHashMap();
+ private final Map<DeviceId, C> clients = Maps.newHashMap();
private final ConcurrentMap<DeviceId, ConcurrentMap<ProviderId, DeviceAgentListener>>
deviceAgentListeners = Maps.newConcurrentMap();
private final Class<E> eventClass;
+ private final String serviceName;
private final Striped<Lock> stripedLocks = Striped.lock(DEFAULT_DEVICE_LOCK_SIZE);
- @Reference(cardinality = ReferenceCardinality.MANDATORY)
- protected GrpcChannelController grpcChannelController;
-
- public AbstractGrpcClientController(Class<E> eventClass) {
+ public AbstractGrpcClientController(Class<E> eventClass, String serviceName) {
this.eventClass = eventClass;
+ this.serviceName = serviceName;
}
@Activate
@@ -97,162 +82,67 @@
@Deactivate
public void deactivate() {
eventDispatcher.removeSink(eventClass);
- clientKeys.keySet().forEach(this::removeClient);
- clientKeys.clear();
clients.clear();
- channelIds.clear();
deviceAgentListeners.clear();
log.info("Stopped");
}
@Override
- public boolean createClient(K clientKey) {
- checkNotNull(clientKey);
- return withDeviceLock(() -> doCreateClient(clientKey),
- clientKey.deviceId());
+ public boolean create(DeviceId deviceId, ManagedChannel channel) {
+ checkNotNull(deviceId);
+ checkNotNull(channel);
+ return withDeviceLock(() -> doCreateClient(deviceId, channel), deviceId);
}
- private boolean doCreateClient(K clientKey) {
- DeviceId deviceId = clientKey.deviceId();
+ private boolean doCreateClient(DeviceId deviceId, ManagedChannel channel) {
- if (clientKeys.containsKey(deviceId)) {
- final GrpcClientKey existingKey = clientKeys.get(deviceId);
- if (clientKey.equals(existingKey)) {
- log.debug("Not creating {} as it already exists... (key={})",
- clientName(clientKey), clientKey);
- return true;
- } else {
- throw new IllegalArgumentException(format(
- "A client already exists for device %s (%s)",
- deviceId, clientKey));
- }
- }
-
- final String method = clientKey.requiresSecureChannel()
- ? "TLS" : "plaintext TCP";
-
- log.info("Connecting {} client for {} to server at {} using {}...",
- clientKey.serviceName(), deviceId, clientKey.serveUri(), method);
-
- SslContext sslContext = null;
- if (clientKey.requiresSecureChannel()) {
- try {
- // FIXME: Accept any server certificate; this is insecure and
- // should not be used in production
- sslContext = GrpcSslContexts.forClient().trustManager(
- InsecureTrustManagerFactory.INSTANCE).build();
- } catch (SSLException e) {
- log.error("Failed to build SSL Context", e);
- return false;
- }
- }
-
- GrpcChannelId channelId = GrpcChannelId.of(clientKey.toString());
- NettyChannelBuilder channelBuilder = NettyChannelBuilder
- .forAddress(clientKey.serveUri().getHost(),
- clientKey.serveUri().getPort())
- .maxInboundMessageSize(DEFAULT_MAX_INBOUND_MSG_SIZE * MEGABYTES);
-
- if (sslContext != null) {
- channelBuilder
- .sslContext(sslContext)
- .useTransportSecurity();
- } else {
- channelBuilder.usePlaintext();
- }
-
- final ManagedChannel channel;
-
- try {
- channel = grpcChannelController.connectChannel(channelId, channelBuilder);
- } catch (Throwable e) {
- log.warn("Failed to connect to {} ({}) using {}: {}",
- deviceId, clientKey.serveUri(), method, e.toString());
- log.debug("gRPC client connection exception", e);
- return false;
+ if (clients.containsKey(deviceId)) {
+ throw new IllegalArgumentException(format(
+ "A %s client already exists for %s", serviceName, deviceId));
}
final C client;
try {
- client = createClientInstance(clientKey, channel);
+ client = createClientInstance(deviceId, channel);
} catch (Throwable e) {
- log.error("Exception while creating {}", clientName(clientKey), e);
- grpcChannelController.disconnectChannel(channelId);
+ log.error("Exception while creating {}", clientName(deviceId), e);
return false;
}
if (client == null) {
- log.error("Unable to create {}, implementation returned null... (key={})",
- clientName(clientKey), clientKey);
- grpcChannelController.disconnectChannel(channelId);
+ log.error("Unable to create {}, implementation returned null...",
+ clientName(deviceId));
return false;
}
- clientKeys.put(deviceId, clientKey);
- clients.put(clientKey, client);
- channelIds.put(deviceId, channelId);
-
+ clients.put(deviceId, client);
return true;
}
- protected abstract C createClientInstance(K clientKey, ManagedChannel channel);
+ protected abstract C createClientInstance(DeviceId deviceId, ManagedChannel channel);
@Override
- public C getClient(DeviceId deviceId) {
+ public C get(DeviceId deviceId) {
checkNotNull(deviceId);
- return withDeviceLock(() -> doGetClient(deviceId), deviceId);
+ return withDeviceLock(() -> clients.get(deviceId), deviceId);
}
- private C doGetClient(DeviceId deviceId) {
- if (!clientKeys.containsKey(deviceId)) {
+ @Override
+ public void remove(DeviceId deviceId) {
+ checkNotNull(deviceId);
+ withDeviceLock(() -> {
+ final C client = clients.remove(deviceId);
+ if (client != null) {
+ client.shutdown();
+ }
return null;
- }
- return clients.get(clientKeys.get(deviceId));
- }
-
- @Override
- public C getClient(K clientKey) {
- checkNotNull(clientKey);
- return clients.get(clientKey);
- }
-
- @Override
- public void removeClient(DeviceId deviceId) {
- checkNotNull(deviceId);
- withDeviceLock(() -> doRemoveClient(deviceId), deviceId);
- }
-
- @Override
- public void removeClient(K clientKey) {
- checkNotNull(clientKey);
- withDeviceLock(() -> doRemoveClient(clientKey), clientKey.deviceId());
- }
-
- private Void doRemoveClient(DeviceId deviceId) {
- if (clientKeys.containsKey(deviceId)) {
- doRemoveClient(clientKeys.get(deviceId));
- }
- return null;
- }
-
- private Void doRemoveClient(K clientKey) {
- if (clients.containsKey(clientKey)) {
- clients.get(clientKey).shutdown();
- }
- if (channelIds.containsKey(clientKey.deviceId())) {
- grpcChannelController.disconnectChannel(
- channelIds.get(clientKey.deviceId()));
- }
- clientKeys.remove(clientKey.deviceId());
- clients.remove(clientKey);
- channelIds.remove(clientKey.deviceId());
- return null;
+ }, deviceId);
}
@Override
public void addDeviceAgentListener(DeviceId deviceId, ProviderId providerId, DeviceAgentListener listener) {
checkNotNull(deviceId, "deviceId cannot be null");
- checkNotNull(deviceId, "providerId cannot be null");
+ checkNotNull(providerId, "providerId cannot be null");
checkNotNull(listener, "listener cannot be null");
deviceAgentListeners.putIfAbsent(deviceId, Maps.newConcurrentMap());
deviceAgentListeners.get(deviceId).put(providerId, listener);
@@ -277,7 +167,7 @@
// We should have only one event delivery mechanism. We have two now
// because we have two different types of events, DeviceAgentEvent and
// controller/protocol specific ones (e.g. P4Runtime or gNMI).
- // TODO: extend device agent event to allow delivery protocol-specific
+ // TODO: extend device agent event to allow delivery of protocol-specific
// events, e.g. packet-in
checkNotNull(event);
if (deviceAgentListeners.containsKey(event.subject())) {
@@ -296,7 +186,7 @@
}
}
- private String clientName(GrpcClientKey key) {
- return format("%s client for %s", key.serviceName(), key.deviceId());
+ private String clientName(DeviceId deviceId) {
+ return format("%s client for %s", serviceName, deviceId);
}
}
diff --git a/protocols/grpc/ctl/src/main/java/org/onosproject/grpc/ctl/GrpcChannelControllerImpl.java b/protocols/grpc/ctl/src/main/java/org/onosproject/grpc/ctl/GrpcChannelControllerImpl.java
index 96a1671..3be7706 100644
--- a/protocols/grpc/ctl/src/main/java/org/onosproject/grpc/ctl/GrpcChannelControllerImpl.java
+++ b/protocols/grpc/ctl/src/main/java/org/onosproject/grpc/ctl/GrpcChannelControllerImpl.java
@@ -16,18 +16,16 @@
package org.onosproject.grpc.ctl;
-import com.google.common.collect.ImmutableMap;
import com.google.common.util.concurrent.Striped;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
-import io.grpc.Status;
-import io.grpc.StatusRuntimeException;
+import io.grpc.netty.GrpcSslContexts;
+import io.grpc.netty.NettyChannelBuilder;
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.util.InsecureTrustManagerFactory;
import org.onlab.util.Tools;
import org.onosproject.cfg.ComponentConfigService;
import org.onosproject.grpc.api.GrpcChannelController;
-import org.onosproject.grpc.api.GrpcChannelId;
-import org.onosproject.grpc.proto.dummy.Dummy;
-import org.onosproject.grpc.proto.dummy.DummyServiceGrpc;
import org.osgi.service.component.ComponentContext;
import org.osgi.service.component.annotations.Activate;
import org.osgi.service.component.annotations.Component;
@@ -38,16 +36,19 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import javax.net.ssl.SSLException;
+import java.net.URI;
import java.util.Dictionary;
import java.util.Map;
import java.util.Optional;
-import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Lock;
+import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;
+import static com.google.common.base.Strings.isNullOrEmpty;
import static java.lang.String.format;
import static org.onosproject.grpc.ctl.OsgiPropertyConstants.ENABLE_MESSAGE_LOG;
import static org.onosproject.grpc.ctl.OsgiPropertyConstants.ENABLE_MESSAGE_LOG_DEFAULT;
@@ -61,6 +62,12 @@
})
public class GrpcChannelControllerImpl implements GrpcChannelController {
+ private static final String GRPC = "grpc";
+ private static final String GRPCS = "grpcs";
+
+ private static final int DEFAULT_MAX_INBOUND_MSG_SIZE = 256; // Megabytes.
+ private static final int MEGABYTES = 1024 * 1024;
+
@Reference(cardinality = ReferenceCardinality.MANDATORY)
protected ComponentConfigService componentConfigService;
@@ -72,8 +79,8 @@
private final Logger log = LoggerFactory.getLogger(getClass());
- private Map<GrpcChannelId, ManagedChannel> channels;
- private Map<GrpcChannelId, GrpcLoggingInterceptor> interceptors;
+ private Map<URI, ManagedChannel> channels;
+ private Map<URI, GrpcLoggingInterceptor> interceptors;
private final Striped<Lock> channelLocks = Striped.lock(30);
@@ -109,129 +116,109 @@
}
@Override
- public ManagedChannel connectChannel(GrpcChannelId channelId,
- ManagedChannelBuilder<?> channelBuilder) {
- checkNotNull(channelId);
- checkNotNull(channelBuilder);
-
- Lock lock = channelLocks.get(channelId);
- lock.lock();
-
- try {
- if (channels.containsKey(channelId)) {
- throw new IllegalArgumentException(format(
- "A channel with ID '%s' already exists", channelId));
- }
-
- final GrpcLoggingInterceptor interceptor = new GrpcLoggingInterceptor(
- channelId, enableMessageLog);
- channelBuilder.intercept(interceptor);
-
- ManagedChannel channel = channelBuilder.build();
- // Forced connection API is still experimental. Use workaround...
- // channel.getState(true);
- try {
- doDummyMessage(channel);
- } catch (StatusRuntimeException e) {
- interceptor.close();
- shutdownNowAndWait(channel, channelId);
- throw e;
- }
- // If here, channel is open.
- channels.put(channelId, channel);
- interceptors.put(channelId, interceptor);
- return channel;
- } finally {
- lock.unlock();
- }
- }
-
- private void doDummyMessage(ManagedChannel channel) throws StatusRuntimeException {
- DummyServiceGrpc.DummyServiceBlockingStub dummyStub = DummyServiceGrpc
- .newBlockingStub(channel)
- .withDeadlineAfter(CONNECTION_TIMEOUT_SECONDS, TimeUnit.SECONDS);
- try {
- //noinspection ResultOfMethodCallIgnored
- dummyStub.sayHello(Dummy.DummyMessageThatNoOneWouldReallyUse
- .getDefaultInstance());
- } catch (StatusRuntimeException e) {
- if (!e.getStatus().equals(Status.UNIMPLEMENTED)) {
- // UNIMPLEMENTED means that the server received our message but
- // doesn't know how to handle it. Hence, channel is open.
- throw e;
- }
- }
+ public ManagedChannel create(URI channelUri) {
+ return create(channelUri, makeChannelBuilder(channelUri));
}
@Override
- public void disconnectChannel(GrpcChannelId channelId) {
- checkNotNull(channelId);
+ public ManagedChannel create(URI channelUri, ManagedChannelBuilder<?> channelBuilder) {
+ checkNotNull(channelUri);
+ checkNotNull(channelBuilder);
- Lock lock = channelLocks.get(channelId);
- lock.lock();
+ channelLocks.get(channelUri).lock();
try {
- final ManagedChannel channel = channels.remove(channelId);
- if (channel != null) {
- shutdownNowAndWait(channel, channelId);
+ if (channels.containsKey(channelUri)) {
+ throw new IllegalArgumentException(format(
+ "A channel with ID '%s' already exists", channelUri));
}
- final GrpcLoggingInterceptor interceptor = interceptors.remove(channelId);
+
+ log.info("Creating new gRPC channel {}...", channelUri);
+
+ final GrpcLoggingInterceptor interceptor = new GrpcLoggingInterceptor(
+ channelUri, enableMessageLog);
+ channelBuilder.intercept(interceptor);
+
+ final ManagedChannel channel = channelBuilder.build();
+
+ channels.put(channelUri, channelBuilder.build());
+ interceptors.put(channelUri, interceptor);
+
+ return channel;
+ } finally {
+ channelLocks.get(channelUri).unlock();
+ }
+ }
+
+ private NettyChannelBuilder makeChannelBuilder(URI channelUri) {
+
+ checkArgument(channelUri.getScheme().equals(GRPC)
+ || channelUri.getScheme().equals(GRPCS),
+ format("Server URI scheme must be %s or %s", GRPC, GRPCS));
+ checkArgument(!isNullOrEmpty(channelUri.getHost()),
+ "Server host address should not be empty");
+ checkArgument(channelUri.getPort() > 0 && channelUri.getPort() <= 65535,
+ "Invalid server port");
+
+ final boolean useTls = channelUri.getScheme().equals(GRPCS);
+
+ final NettyChannelBuilder channelBuilder = NettyChannelBuilder
+ .forAddress(channelUri.getHost(),
+ channelUri.getPort())
+ .maxInboundMessageSize(DEFAULT_MAX_INBOUND_MSG_SIZE * MEGABYTES);
+
+ if (useTls) {
+ try {
+ // Accept any server certificate; this is insecure and
+ // should not be used in production.
+ final SslContext sslContext = GrpcSslContexts.forClient().trustManager(
+ InsecureTrustManagerFactory.INSTANCE).build();
+ channelBuilder.sslContext(sslContext).useTransportSecurity();
+ } catch (SSLException e) {
+ log.error("Failed to build SSL context", e);
+ return null;
+ }
+ } else {
+ channelBuilder.usePlaintext();
+ }
+
+ return channelBuilder;
+ }
+
+ @Override
+ public void destroy(URI channelUri) {
+ checkNotNull(channelUri);
+
+ channelLocks.get(channelUri).lock();
+ try {
+ final ManagedChannel channel = channels.remove(channelUri);
+ if (channel != null) {
+ shutdownNowAndWait(channel, channelUri);
+ }
+ final GrpcLoggingInterceptor interceptor = interceptors.remove(channelUri);
if (interceptor != null) {
interceptor.close();
}
} finally {
- lock.unlock();
+ channelLocks.get(channelUri).unlock();
}
}
- private void shutdownNowAndWait(ManagedChannel channel, GrpcChannelId channelId) {
+ private void shutdownNowAndWait(ManagedChannel channel, URI channelUri) {
try {
if (!channel.shutdownNow()
.awaitTermination(5, TimeUnit.SECONDS)) {
- log.error("Channel '{}' didn't terminate, although we " +
- "triggered a shutdown and waited",
- channelId);
+ log.error("Channel {} did not terminate properly",
+ channelUri);
}
} catch (InterruptedException e) {
- log.warn("Channel {} didn't shutdown in time", channelId);
+ log.warn("Channel {} didn't shutdown in time", channelUri);
Thread.currentThread().interrupt();
}
}
@Override
- public Map<GrpcChannelId, ManagedChannel> getChannels() {
- return ImmutableMap.copyOf(channels);
- }
-
- @Override
- public Optional<ManagedChannel> getChannel(GrpcChannelId channelId) {
- checkNotNull(channelId);
-
- Lock lock = channelLocks.get(channelId);
- lock.lock();
- try {
- return Optional.ofNullable(channels.get(channelId));
- } finally {
- lock.unlock();
- }
- }
-
- @Override
- public CompletableFuture<Boolean> probeChannel(GrpcChannelId channelId) {
- final ManagedChannel channel = channels.get(channelId);
- if (channel == null) {
- log.warn("Unable to find any channel with ID {}, cannot send probe",
- channelId);
- return CompletableFuture.completedFuture(false);
- }
- return CompletableFuture.supplyAsync(() -> {
- try {
- doDummyMessage(channel);
- return true;
- } catch (StatusRuntimeException e) {
- log.debug("Probe for {} failed", channelId);
- log.debug("", e);
- return false;
- }
- });
+ public Optional<ManagedChannel> get(URI channelUri) {
+ checkNotNull(channelUri);
+ return Optional.ofNullable(channels.get(channelUri));
}
}
diff --git a/protocols/grpc/ctl/src/main/java/org/onosproject/grpc/ctl/GrpcLoggingInterceptor.java b/protocols/grpc/ctl/src/main/java/org/onosproject/grpc/ctl/GrpcLoggingInterceptor.java
index 45ec5ba..6f06c11 100644
--- a/protocols/grpc/ctl/src/main/java/org/onosproject/grpc/ctl/GrpcLoggingInterceptor.java
+++ b/protocols/grpc/ctl/src/main/java/org/onosproject/grpc/ctl/GrpcLoggingInterceptor.java
@@ -26,12 +26,12 @@
import io.grpc.MethodDescriptor;
import io.grpc.Status;
import io.grpc.protobuf.lite.ProtoLiteUtils;
-import org.onosproject.grpc.api.GrpcChannelId;
import org.slf4j.Logger;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
+import java.net.URI;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.StringJoiner;
@@ -55,13 +55,13 @@
private static final Logger log = getLogger(GrpcLoggingInterceptor.class);
private final AtomicLong callIdGenerator = new AtomicLong();
- private final GrpcChannelId channelId;
+ private final URI channelUri;
private final AtomicBoolean enabled;
private FileWriter writer;
- GrpcLoggingInterceptor(GrpcChannelId channelId, AtomicBoolean enabled) {
- this.channelId = channelId;
+ GrpcLoggingInterceptor(URI channelUri, AtomicBoolean enabled) {
+ this.channelUri = channelUri;
this.enabled = enabled;
}
@@ -69,14 +69,13 @@
if (writer != null) {
return true;
}
- final String safeChName = channelId.id()
- .replaceAll("[^A-Za-z0-9]", "_");
- final String fileName = format("grpc_%s_", safeChName).toLowerCase();
+ final String safeChName = channelUri.toString()
+ .replaceAll("[^A-Za-z0-9]", "_").toLowerCase();
try {
- final File tmpFile = File.createTempFile(fileName, ".log");
+ final File tmpFile = File.createTempFile(safeChName + "_", ".log");
this.writer = new FileWriter(tmpFile);
log.info("Created gRPC call log file for channel {}: {}",
- channelId, tmpFile.getAbsolutePath());
+ channelUri, tmpFile.getAbsolutePath());
return true;
} catch (IOException e) {
log.error("Unable to initialize gRPC call log writer", e);
@@ -90,7 +89,7 @@
return;
}
try {
- log.info("Closing log writer for {}...", channelId);
+ log.info("Closing log writer for {}...", channelUri);
writer.close();
} catch (IOException e) {
log.error("Unable to close gRPC call log writer", e);
diff --git a/protocols/grpc/proto/BUILD b/protocols/grpc/proto/BUILD
deleted file mode 100644
index 48249c0..0000000
--- a/protocols/grpc/proto/BUILD
+++ /dev/null
@@ -1,11 +0,0 @@
-load("//tools/build/bazel:osgi_java_library.bzl", "osgi_proto_jar")
-
-osgi_proto_jar(
- grpc_proto_lib = ":dummy_proto",
- proto_libs = [":dummy_proto"],
-)
-
-proto_library(
- name = "dummy_proto",
- srcs = ["dummy.proto"],
-)
diff --git a/protocols/grpc/proto/dummy.proto b/protocols/grpc/proto/dummy.proto
deleted file mode 100644
index 003f403..0000000
--- a/protocols/grpc/proto/dummy.proto
+++ /dev/null
@@ -1,13 +0,0 @@
-syntax = "proto3";
-
-option java_package = "org.onosproject.grpc.proto.dummy";
-
-package dummy;
-
-service DummyService {
- rpc SayHello (DummyMessageThatNoOneWouldReallyUse) returns (DummyMessageThatNoOneWouldReallyUse) {
- }
-}
-
-message DummyMessageThatNoOneWouldReallyUse {
-}
diff --git a/protocols/grpc/utils/BUILD b/protocols/grpc/utils/BUILD
new file mode 100644
index 0000000..7baaf52
--- /dev/null
+++ b/protocols/grpc/utils/BUILD
@@ -0,0 +1,8 @@
+COMPILE_DEPS = CORE_DEPS + [
+ "//protocols/grpc/api:onos-protocols-grpc-api",
+ "@io_grpc_grpc_java//core",
+]
+
+osgi_jar(
+ deps = COMPILE_DEPS,
+)
diff --git a/protocols/grpc/utils/src/main/java/org/onosproject/grpc/utils/AbstractGrpcHandlerBehaviour.java b/protocols/grpc/utils/src/main/java/org/onosproject/grpc/utils/AbstractGrpcHandlerBehaviour.java
new file mode 100644
index 0000000..c66dede
--- /dev/null
+++ b/protocols/grpc/utils/src/main/java/org/onosproject/grpc/utils/AbstractGrpcHandlerBehaviour.java
@@ -0,0 +1,100 @@
+/*
+ * Copyright 2019-present Open Networking Foundation
+ *
+ * Licensed 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.onosproject.grpc.utils;
+
+import com.google.common.base.Strings;
+import com.google.common.collect.Maps;
+import org.onosproject.grpc.api.GrpcClient;
+import org.onosproject.grpc.api.GrpcClientController;
+import org.onosproject.net.DeviceId;
+import org.onosproject.net.config.NetworkConfigService;
+import org.onosproject.net.config.basics.BasicDeviceConfig;
+import org.onosproject.net.device.DeviceService;
+import org.onosproject.net.driver.AbstractHandlerBehaviour;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.concurrent.ConcurrentMap;
+
+/**
+ * Abstract implementation of HandlerBehaviour for gNMI-based devices.
+ *
+ * @param <CLIENT> gRPC client class
+ * @param <CTRL> gRPC controller class
+ */
+public class AbstractGrpcHandlerBehaviour
+ <CLIENT extends GrpcClient, CTRL extends GrpcClientController<CLIENT>>
+ extends AbstractHandlerBehaviour {
+
+ static final ConcurrentMap<DeviceId, URI> CHANNEL_URIS = Maps.newConcurrentMap();
+
+ protected final Logger log = LoggerFactory.getLogger(getClass());
+
+ final Class<CTRL> controllerClass;
+ protected DeviceId deviceId;
+ protected DeviceService deviceService;
+ protected CLIENT client;
+
+ public AbstractGrpcHandlerBehaviour(Class<CTRL> controllerClass) {
+ this.controllerClass = controllerClass;
+ }
+
+ protected boolean setupBehaviour(String opName) {
+ deviceId = handler().data().deviceId();
+ deviceService = handler().get(DeviceService.class);
+ client = getClientByNetcfg();
+ if (client == null) {
+ log.warn("Missing client for {}, aborting {}", deviceId, opName);
+ return false;
+ }
+
+ return true;
+ }
+
+ private CLIENT getClientByNetcfg() {
+ // Check if there's a channel for this device and if we created it with
+ // the same URI of that derived from the current netcfg. This makes sure
+ // we return null if the netcfg changed after we created the channel.
+ if (!CHANNEL_URIS.containsKey(data().deviceId()) ||
+ !CHANNEL_URIS.get(data().deviceId()).equals(mgmtUriFromNetcfg())) {
+ return null;
+ }
+ return handler().get(controllerClass).get(data().deviceId());
+ }
+
+ protected URI mgmtUriFromNetcfg() {
+ deviceId = handler().data().deviceId();
+
+ final BasicDeviceConfig cfg = handler().get(NetworkConfigService.class)
+ .getConfig(deviceId, BasicDeviceConfig.class);
+ if (cfg == null || Strings.isNullOrEmpty(cfg.managementAddress())) {
+ log.error("Missing or invalid config for {}, cannot derive " +
+ "gRPC server endpoints", deviceId);
+ return null;
+ }
+
+ try {
+ return new URI(cfg.managementAddress());
+ } catch (URISyntaxException e) {
+ log.error("Management address of {} is not a valid URI: {}",
+ deviceId, cfg.managementAddress());
+ return null;
+ }
+ }
+}
diff --git a/protocols/grpc/utils/src/main/java/org/onosproject/grpc/utils/AbstractGrpcHandshaker.java b/protocols/grpc/utils/src/main/java/org/onosproject/grpc/utils/AbstractGrpcHandshaker.java
new file mode 100644
index 0000000..0183133
--- /dev/null
+++ b/protocols/grpc/utils/src/main/java/org/onosproject/grpc/utils/AbstractGrpcHandshaker.java
@@ -0,0 +1,178 @@
+/*
+ * Copyright 2019-present Open Networking Foundation
+ *
+ * Licensed 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.onosproject.grpc.utils;
+
+import com.google.common.util.concurrent.Striped;
+import io.grpc.ManagedChannel;
+import org.onosproject.grpc.api.GrpcChannelController;
+import org.onosproject.grpc.api.GrpcClient;
+import org.onosproject.grpc.api.GrpcClientController;
+import org.onosproject.net.DeviceId;
+import org.onosproject.net.device.DeviceAgentListener;
+import org.onosproject.net.device.DeviceHandshaker;
+import org.onosproject.net.provider.ProviderId;
+
+import java.net.URI;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.locks.Lock;
+
+import static java.util.concurrent.CompletableFuture.completedFuture;
+
+/**
+ * Abstract implementation of DeviceHandshaker that uses gRPC to establish a
+ * connection to the device.
+ *
+ * @param <CLIENT> gRPC client class
+ * @param <CTRL> gRPC controller class
+ */
+public abstract class AbstractGrpcHandshaker
+ <CLIENT extends GrpcClient, CTRL extends GrpcClientController<CLIENT>>
+ extends AbstractGrpcHandlerBehaviour<CLIENT, CTRL>
+ implements DeviceHandshaker {
+
+ /**
+ * Creates a new instance of this behaviour for the given gRPC controller
+ * class.
+ *
+ * @param controllerClass gRPC controller class
+ */
+ public AbstractGrpcHandshaker(Class<CTRL> controllerClass) {
+ super(controllerClass);
+ }
+
+ private static final Striped<Lock> DEVICE_LOCKS = Striped.lock(10);
+
+ @Override
+ public boolean connect() {
+ final GrpcChannelController channelController = handler().get(
+ GrpcChannelController.class);
+ final CTRL clientController = handler().get(controllerClass);
+ final DeviceId deviceId = data().deviceId();
+
+ final URI netcfgUri = mgmtUriFromNetcfg();
+ if (netcfgUri == null) {
+ return false;
+ }
+
+ DEVICE_LOCKS.get(deviceId).lock();
+ try {
+ if (clientController.get(deviceId) != null) {
+ throw new IllegalStateException(
+ "A client for this device already exists");
+ }
+
+ // Create or get an existing channel. We support sharing the same
+ // channel by different drivers for the same device.
+ final ManagedChannel channel;
+ final URI existingChannelUri = CHANNEL_URIS.get(deviceId);
+ if (existingChannelUri != null) {
+ if (!existingChannelUri.equals(netcfgUri)) {
+ throw new IllegalStateException(
+ "A gRPC channel with different URI already " +
+ "exists for this device");
+ }
+ channel = channelController.get(existingChannelUri)
+ .orElseThrow(() -> new IllegalStateException(
+ "Missing gRPC channel in controller"));
+ } else {
+ try {
+ channel = channelController.create(netcfgUri);
+ } catch (IllegalArgumentException ex) {
+ throw new IllegalStateException(
+ "A gRPC channel with same URI already exists", ex);
+ }
+ // Store channel URI for future use.
+ CHANNEL_URIS.put(deviceId, netcfgUri);
+ // Trigger connection.
+ channel.getState(true);
+ }
+
+ return clientController.create(deviceId, channel);
+ } finally {
+ DEVICE_LOCKS.get(deviceId).unlock();
+ }
+ }
+
+ @Override
+ public boolean hasConnection() {
+ final DeviceId deviceId = data().deviceId();
+ final URI netcfgUri = mgmtUriFromNetcfg();
+ // If a client already exists for this device, but the netcfg with the
+ // server endpoints has changed, this will return false.
+ DEVICE_LOCKS.get(deviceId).lock();
+ try {
+ final URI existingChannelUri = CHANNEL_URIS.get(deviceId);
+ return existingChannelUri != null &&
+ existingChannelUri.equals(netcfgUri) &&
+ handler().get(GrpcChannelController.class)
+ .get(existingChannelUri).isPresent() &&
+ handler().get(controllerClass)
+ .get(deviceId) != null;
+ } finally {
+ DEVICE_LOCKS.get(deviceId).unlock();
+ }
+ }
+
+ @Override
+ public void disconnect() {
+ final DeviceId deviceId = data().deviceId();
+ final URI netcfgUri = mgmtUriFromNetcfg();
+ // This removes any clients and channels associated with this device ID.
+ DEVICE_LOCKS.get(deviceId).lock();
+ try {
+ final URI existingChannelUri = CHANNEL_URIS.remove(deviceId);
+ handler().get(controllerClass).remove(deviceId);
+ if (existingChannelUri != null) {
+ handler().get(GrpcChannelController.class).destroy(existingChannelUri);
+ }
+ if (netcfgUri != null) {
+ // This should not be needed if we are sure there can never be
+ // two channels for the same device.
+ handler().get(GrpcChannelController.class).destroy(netcfgUri);
+ }
+ } finally {
+ DEVICE_LOCKS.get(deviceId).unlock();
+ }
+ }
+
+ @Override
+ public boolean isReachable() {
+ return setupBehaviour("isReachable()") && client.isServerReachable();
+ }
+
+ @Override
+ public CompletableFuture<Boolean> probeReachability() {
+ if (!setupBehaviour("probeReachability()")) {
+ return completedFuture(false);
+ }
+ return client.probeService();
+ }
+
+ @Override
+ public void addDeviceAgentListener(ProviderId providerId, DeviceAgentListener listener) {
+ // Don't use controller/deviceId class variables as they might be uninitialized.
+ handler().get(controllerClass)
+ .addDeviceAgentListener(data().deviceId(), providerId, listener);
+ }
+
+ @Override
+ public void removeDeviceAgentListener(ProviderId providerId) {
+ // Don't use controller/deviceId class variable as they might be uninitialized.
+ handler().get(controllerClass)
+ .removeDeviceAgentListener(data().deviceId(), providerId);
+ }
+}
diff --git a/protocols/grpc/utils/src/main/java/org/onosproject/grpc/utils/package-info.java b/protocols/grpc/utils/src/main/java/org/onosproject/grpc/utils/package-info.java
new file mode 100644
index 0000000..ab13d88
--- /dev/null
+++ b/protocols/grpc/utils/src/main/java/org/onosproject/grpc/utils/package-info.java
@@ -0,0 +1,20 @@
+/*
+ * Copyright 2019-present Open Networking Foundation
+ *
+ * Licensed 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.
+ */
+
+/**
+ * gRPC protocol utils.
+ */
+package org.onosproject.grpc.utils;