ODL Netconf应用层keepalive机制实现

void onDeviceConnected(SchemaContext remoteSchemaContext, PREF netconfSessionPreferences, DOMRpcService deviceRpc); void onDeviceDisconnected(); void onDeviceFailed(Throwable throwable); void onNotification(DOMNotification domNotification); void close();

@Override public void onDeviceConnected(final SchemaContext remoteSchemaContext, final NetconfSessionPreferences netconfSessionPreferences, final DOMRpcService deviceRpc) { this.currentDeviceRpc = deviceRpc; final DOMRpcService deviceRpc1 = new KeepaliveDOMRpcService(deviceRpc, resetKeepaliveTask, defaultRequestTimeoutMillis, executor); salFacade.onDeviceConnected(remoteSchemaContext, netconfSessionPreferences, deviceRpc1);

LOG.debug("{}: Netconf session initiated, starting keepalives", id); scheduleKeepalive(); }

@Override public CheckedFuture<DOMRpcResult, DOMRpcException> invokeRpc(@Nonnull final SchemaPath type, final NormalizedNode<?, ?> input) { final CheckedFuture<DOMRpcResult, DOMRpcException> domRpcResultDOMRpcExceptionCheckedFuture = deviceRpc.invokeRpc(type, input); Futures.addCallback(domRpcResultDOMRpcExceptionCheckedFuture, resetKeepaliveTask, MoreExecutors.directExecutor());

final RequestTimeoutTask timeoutTask = new RequestTimeoutTask(domRpcResultDOMRpcExceptionCheckedFuture); executor.schedule(timeoutTask, defaultRequestTimeoutMillis, TimeUnit.MILLISECONDS);

return domRpcResultDOMRpcExceptionCheckedFuture; }

/** * Reset keepalive after each RPC response received. */ private class ResetKeepalive implements FutureCallback<DOMRpcResult> { @Override public void onSuccess(@Nullable final DOMRpcResult result) { // No matter what response we got, // rpc-reply or rpc-error, we got it from device so the netconf session is OK. resetKeepalive(); }

@Override public void onFailure(@Nonnull final Throwable throwable) { // User/Application RPC failed (The RPC did not reach the remote device or .. // TODO what other reasons could cause this ?) // There is no point in keeping this session. Reconnect. LOG.warn("{}: Rpc failure detected. Reconnecting netconf session", id, throwable); reconnect(); } }

private void scheduleKeepalive() { Preconditions.checkState(currentDeviceRpc != null); LOG.trace("{}: Scheduling next keepalive in {} {}", id, keepaliveDelaySeconds, TimeUnit.SECONDS); currentKeepalive = executor.schedule(new Keepalive(currentKeepalive), keepaliveDelaySeconds, TimeUnit.SECONDS); }

// Keepalive configuration leaf keepalive-delay { config true; type uint32; default 120; description "Netconf connector sends keepalive RPCs while the session is idle, this delay specifies the delay between keepalive RPC in seconds If a value <1 is provided, no keepalives will be sent"; }

private class Keepalive implements Runnable, FutureCallback<DOMRpcResult> { ... }

// Keepalive RPC static resources private static final SchemaPath PATH = toPath(NETCONF_GET_CONFIG_QNAME); private static final ContainerNode KEEPALIVE_PAYLOAD = NetconfMessageTransformUtil.wrap(NETCONF_GET_CONFIG_QNAME, getSourceNode(NETCONF_RUNNING_QNAME), NetconfMessageTransformUtil.EMPTY_FILTER);

... @Override public void run() { LOG.trace("{}: Invoking keepalive RPC", id);

try { if (previousKeepalive != null && !previousKeepalive.isDone()) { onFailure(new IllegalStateException("Previous keepalive timed out")); } else { // get-config 请求底层设备 Futures.addCallback(currentDeviceRpc.invokeRpc(PATH, KEEPALIVE_PAYLOAD), this, MoreExecutors.directExecutor()); } } catch (NullPointerException e) { LOG.debug("{}: Skipping keepalive while reconnecting", id); // Empty catch block intentional // Do nothing. The currentDeviceRpc was null and it means we hit the reconnect window and // attempted to send keepalive while we were reconnecting. Next keepalive will be scheduled // after reconnect so no action necessary here. } }

@Override public void onSuccess(final DOMRpcResult result) { // No matter what response we got, rpc-reply or rpc-error, // we got it from device so the netconf session is OK if (result != null && result.getResult() != null) { LOG.debug("{}: Keepalive RPC successful with response: {}", id, result.getResult()); scheduleKeepalive(); } else if (result != null && result.getErrors() != null) { LOG.warn("{}: Keepalive RPC failed with error: {}", id, result.getErrors()); scheduleKeepalive(); } else { LOG.warn("{} Keepalive RPC returned null with response: {}. Reconnecting netconf session", id, result); reconnect(); } }

private void scheduleKeepalives() { lastKeepAliveSucceeded.set(true); Preconditions.checkState(currentDeviceRpc != null); LOG.trace("{}: Scheduling keepalives every {} {}", id, keepaliveDelaySeconds, TimeUnit.SECONDS); // 定时调用 currentKeepalive = executor.scheduleWithFixedDelay(new Keepalive(), keepaliveDelaySeconds, keepaliveDelaySeconds, TimeUnit.SECONDS); }

private class Keepalive implements Runnable, FutureCallback<DOMRpcResult> {

@Override public void run() { LOG.trace("{}: Invoking keepalive RPC", id);

try { // 判断标识位 boolean lastJobSucceeded = lastKeepAliveSucceeded.getAndSet(false); if (!lastJobSucceeded) { onFailure(new IllegalStateException("Previous keepalive timed out")); } else { Futures.addCallback(currentDeviceRpc.invokeRpc(PATH, KEEPALIVE_PAYLOAD), this, MoreExecutors.directExecutor()); } } catch (NullPointerException e) { LOG.debug("{}: Skipping keepalive while reconnecting", id); // Empty catch block intentional // Do nothing. The currentDeviceRpc was null and it means we hit the reconnect window and // attempted to send keepalive while we were reconnecting. Next keepalive will be scheduled // after reconnect so no action necessary here. } }

@Override public void onSuccess(final DOMRpcResult result) { // No matter what response we got, rpc-reply or rpc-error, // we got it from device so the netconf session is OK if (result != null && result.getResult() != null) { LOG.debug("{}: Keepalive RPC successful with response: {}", id, result.getResult()); // 设置标识位 lastKeepAliveSucceeded.set(true); } else if (result != null && result.getErrors() != null) { LOG.warn("{}: Keepalive RPC failed with error: {}", id, result.getErrors()); // 设置标识位 lastKeepAliveSucceeded.set(true); } else { LOG.warn("{} Keepalive RPC returned null with response: {}. Reconnecting netconf session", id, result); reconnect(); } }

@Override public void onFailure(@Nonnull final Throwable throwable) { LOG.warn("{}: Keepalive RPC failed. Reconnecting netconf session.", id, throwable); reconnect(); } }