BUG-3335 Add keepalive mechanism to netconf-connector 88/20988/3
authorMaros Marsalek <mmarsale@cisco.com>
Fri, 22 May 2015 14:40:58 +0000 (16:40 +0200)
committerGerrit Code Review <gerrit@opendaylight.org>
Wed, 27 May 2015 16:10:37 +0000 (16:10 +0000)
Invoke harmless RPC (get-config with empty filter) with a fixed delay to check
whether the netconf session is still active. The RPC is postponed if any other
RPC is invoked by a user/application.

This also prevents the remote device to close session due to being idle.

Change-Id: I013d2641c38d4c8adb5d3198795e337b91c4f95d
Signed-off-by: Maros Marsalek <mmarsale@cisco.com>
opendaylight/md-sal/sal-netconf-connector/src/main/java/org/opendaylight/controller/config/yang/md/sal/connector/netconf/NetconfConnectorModule.java
opendaylight/md-sal/sal-netconf-connector/src/main/java/org/opendaylight/controller/sal/connect/netconf/sal/KeepaliveSalFacade.java [new file with mode: 0644]
opendaylight/md-sal/sal-netconf-connector/src/main/java/org/opendaylight/controller/sal/connect/netconf/util/NetconfMessageTransformUtil.java
opendaylight/md-sal/sal-netconf-connector/src/main/yang/odl-sal-netconf-connector-cfg.yang
opendaylight/md-sal/sal-netconf-connector/src/test/java/org/opendaylight/controller/sal/connect/netconf/sal/KeepaliveSalFacadeTest.java [new file with mode: 0644]

index e115c36ac0fda36a079620ffbcfdb15d8072a08c..12e4855fbde9d8de73c049c2a10f2124c76f33bb 100644 (file)
@@ -16,6 +16,9 @@ import java.math.BigDecimal;
 import java.net.InetSocketAddress;
 import java.util.List;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadFactory;
 import org.opendaylight.controller.config.api.JmxAttributeValidationException;
 import org.opendaylight.controller.netconf.client.NetconfClientDispatcher;
 import org.opendaylight.controller.netconf.client.conf.NetconfClientConfiguration;
@@ -28,6 +31,7 @@ import org.opendaylight.controller.sal.connect.netconf.NetconfDevice;
 import org.opendaylight.controller.sal.connect.netconf.NetconfStateSchemas;
 import org.opendaylight.controller.sal.connect.netconf.listener.NetconfDeviceCommunicator;
 import org.opendaylight.controller.sal.connect.netconf.listener.NetconfSessionPreferences;
+import org.opendaylight.controller.sal.connect.netconf.sal.KeepaliveSalFacade;
 import org.opendaylight.controller.sal.connect.netconf.sal.NetconfDeviceSalFacade;
 import org.opendaylight.controller.sal.connect.util.RemoteDeviceId;
 import org.opendaylight.controller.sal.core.api.Broker;
@@ -90,6 +94,23 @@ public final class NetconfConnectorModule extends org.opendaylight.controller.co
         }
 
         userCapabilities = getUserCapabilities();
+
+        if(getKeepaliveExecutor() == null) {
+            logger.warn("Keepalive executor missing. Using default instance for now, the configuration needs to be updated");
+
+            // Instantiate the default executor, now we know its necessary
+            if(DEFAULT_KEEPALIVE_EXECUTOR == null) {
+                DEFAULT_KEEPALIVE_EXECUTOR = Executors.newScheduledThreadPool(2, new ThreadFactory() {
+                    @Override
+                    public Thread newThread(final Runnable r) {
+                        final Thread thread = new Thread(r);
+                        thread.setName("netconf-southound-keepalives-" + thread.getId());
+                        thread.setDaemon(true);
+                        return thread;
+                    }
+                });
+            }
+        }
     }
 
     private boolean isHostAddressPresent(final Host address) {
@@ -97,6 +118,9 @@ public final class NetconfConnectorModule extends org.opendaylight.controller.co
                address.getIpAddress() != null && (address.getIpAddress().getIpv4Address() != null || address.getIpAddress().getIpv6Address() != null);
     }
 
+    @Deprecated
+    private static ScheduledExecutorService DEFAULT_KEEPALIVE_EXECUTOR;
+
     @Override
     public java.lang.AutoCloseable createInstance() {
         final RemoteDeviceId id = new RemoteDeviceId(getIdentifier(), getSocketAddress());
@@ -106,9 +130,17 @@ public final class NetconfConnectorModule extends org.opendaylight.controller.co
         final Broker domBroker = getDomRegistryDependency();
         final BindingAwareBroker bindingBroker = getBindingRegistryDependency();
 
-        final RemoteDeviceHandler<NetconfSessionPreferences> salFacade
+        RemoteDeviceHandler<NetconfSessionPreferences> salFacade
                 = new NetconfDeviceSalFacade(id, domBroker, bindingBroker, bundleContext, getDefaultRequestTimeoutMillis());
 
+        final Long keepaliveDelay = getKeepaliveDelay();
+        if(shouldSendKeepalive()) {
+            // Keepalive executor is optional for now and a default instance is supported
+            final ScheduledExecutorService executor = getKeepaliveExecutor() == null ?
+                    DEFAULT_KEEPALIVE_EXECUTOR : getKeepaliveExecutorDependency().getExecutor();
+            salFacade = new KeepaliveSalFacade(id, salFacade, executor, keepaliveDelay);
+        }
+
         final NetconfDevice.SchemaResourcesDTO schemaResourcesDTO =
                 new NetconfDevice.SchemaResourcesDTO(schemaRegistry, schemaContextFactory, new NetconfStateSchemas.NetconfStateSchemasResolverImpl());
 
@@ -118,6 +150,10 @@ public final class NetconfConnectorModule extends org.opendaylight.controller.co
         final NetconfDeviceCommunicator listener = userCapabilities.isPresent() ?
                 new NetconfDeviceCommunicator(id, device, userCapabilities.get()) : new NetconfDeviceCommunicator(id, device);
 
+        if(shouldSendKeepalive()) {
+            ((KeepaliveSalFacade) salFacade).setListener(listener);
+        }
+
         final NetconfReconnectingClientConfiguration clientConfig = getClientConfig(listener);
         final NetconfClientDispatcher dispatcher = getClientDispatcherDependency();
 
@@ -126,6 +162,10 @@ public final class NetconfConnectorModule extends org.opendaylight.controller.co
         return new SalConnectorCloseable(listener, salFacade);
     }
 
+    private boolean shouldSendKeepalive() {
+        return getKeepaliveDelay() > 0;
+    }
+
     private Optional<NetconfSessionPreferences> getUserCapabilities() {
         if(getYangModuleCapabilities() == null) {
             return Optional.absent();
diff --git a/opendaylight/md-sal/sal-netconf-connector/src/main/java/org/opendaylight/controller/sal/connect/netconf/sal/KeepaliveSalFacade.java b/opendaylight/md-sal/sal-netconf-connector/src/main/java/org/opendaylight/controller/sal/connect/netconf/sal/KeepaliveSalFacade.java
new file mode 100644 (file)
index 0000000..d302cac
--- /dev/null
@@ -0,0 +1,241 @@
+/*
+ * Copyright (c) 2015 Cisco Systems, Inc. and others.  All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+package org.opendaylight.controller.sal.connect.netconf.sal;
+
+import static org.opendaylight.controller.sal.connect.netconf.util.NetconfBaseOps.getSourceNode;
+import static org.opendaylight.controller.sal.connect.netconf.util.NetconfMessageTransformUtil.NETCONF_GET_CONFIG_QNAME;
+import static org.opendaylight.controller.sal.connect.netconf.util.NetconfMessageTransformUtil.NETCONF_RUNNING_QNAME;
+import static org.opendaylight.controller.sal.connect.netconf.util.NetconfMessageTransformUtil.toPath;
+
+import com.google.common.base.Preconditions;
+import com.google.common.util.concurrent.CheckedFuture;
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+import org.opendaylight.controller.md.sal.dom.api.DOMNotification;
+import org.opendaylight.controller.md.sal.dom.api.DOMRpcAvailabilityListener;
+import org.opendaylight.controller.md.sal.dom.api.DOMRpcException;
+import org.opendaylight.controller.md.sal.dom.api.DOMRpcResult;
+import org.opendaylight.controller.md.sal.dom.api.DOMRpcService;
+import org.opendaylight.controller.sal.connect.api.RemoteDeviceHandler;
+import org.opendaylight.controller.sal.connect.netconf.listener.NetconfDeviceCommunicator;
+import org.opendaylight.controller.sal.connect.netconf.listener.NetconfSessionPreferences;
+import org.opendaylight.controller.sal.connect.netconf.util.NetconfMessageTransformUtil;
+import org.opendaylight.controller.sal.connect.util.RemoteDeviceId;
+import org.opendaylight.yangtools.concepts.ListenerRegistration;
+import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode;
+import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
+import org.opendaylight.yangtools.yang.model.api.SchemaContext;
+import org.opendaylight.yangtools.yang.model.api.SchemaPath;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * SalFacade proxy that invokes keepalive RPCs to prevent session shutdown from remote device
+ * and to detect incorrect session drops (netconf session is inactive, but TCP/SSH connection is still present).
+ * The keepalive RPC is a get-config with empty filter.
+ */
+public final class KeepaliveSalFacade implements RemoteDeviceHandler<NetconfSessionPreferences> {
+
+    private static final Logger LOG = LoggerFactory.getLogger(KeepaliveSalFacade.class);
+
+    // 2 minutes keepalive delay by default
+    private static final long DEFAULT_DELAY = TimeUnit.MINUTES.toSeconds(2);
+
+    private final RemoteDeviceId id;
+    private final RemoteDeviceHandler<NetconfSessionPreferences> salFacade;
+    private final ScheduledExecutorService executor;
+    private final long keepaliveDelaySeconds;
+    private final ResetKeepalive resetKeepaliveTask;
+
+    private volatile NetconfDeviceCommunicator listener;
+    private volatile ScheduledFuture<?> currentKeepalive;
+    private volatile DOMRpcService currentDeviceRpc;
+
+    public KeepaliveSalFacade(final RemoteDeviceId id, final RemoteDeviceHandler<NetconfSessionPreferences> salFacade,
+                              final ScheduledExecutorService executor, final long keepaliveDelaySeconds) {
+        this.id = id;
+        this.salFacade = salFacade;
+        this.executor = executor;
+        this.keepaliveDelaySeconds = keepaliveDelaySeconds;
+        this.resetKeepaliveTask = new ResetKeepalive();
+    }
+
+    public KeepaliveSalFacade(final RemoteDeviceId id, final RemoteDeviceHandler<NetconfSessionPreferences> salFacade,
+                              final ScheduledExecutorService executor) {
+        this(id, salFacade, executor, DEFAULT_DELAY);
+    }
+
+    /**
+     * Set the netconf session listener whenever ready
+     *
+     * @param listener netconf session listener
+     */
+    public void setListener(final NetconfDeviceCommunicator listener) {
+        this.listener = listener;
+    }
+
+    /**
+     * Just cancel current keepalive task.
+     * If its already started, let it finish ... not such a big deal.
+     *
+     * Then schedule next keepalive.
+     */
+    private void resetKeepalive() {
+        LOG.trace("{}: Resetting netconf keepalive timer", id);
+        currentKeepalive.cancel(false);
+        scheduleKeepalive();
+    }
+
+    /**
+     * Cancel current keepalive and also reset current deviceRpc
+     */
+    private void stopKeepalives() {
+        currentKeepalive.cancel(false);
+        currentDeviceRpc = null;
+    }
+
+    private void reconnect() {
+        Preconditions.checkState(listener != null, "%s: Unable to reconnect, session listener is missing", id);
+        stopKeepalives();
+        LOG.info("{}: Reconnecting inactive netconf session", id);
+        listener.disconnect();
+    }
+
+    @Override
+    public void onDeviceConnected(final SchemaContext remoteSchemaContext, final NetconfSessionPreferences netconfSessionPreferences, final DOMRpcService deviceRpc) {
+        this.currentDeviceRpc = deviceRpc;
+        final DOMRpcService deviceRpc1 = new KeepaliveDOMRpcService(deviceRpc, resetKeepaliveTask);
+        salFacade.onDeviceConnected(remoteSchemaContext, netconfSessionPreferences, deviceRpc1);
+
+        LOG.debug("{}: Netconf session initiated, starting keepalives", id);
+        scheduleKeepalive();
+    }
+
+    private void scheduleKeepalive() {
+        Preconditions.checkState(currentDeviceRpc != null);
+        LOG.trace("{}: Scheduling next keepalive in {} {}", id, keepaliveDelaySeconds, TimeUnit.SECONDS);
+        currentKeepalive = executor.schedule(new Keepalive(), keepaliveDelaySeconds, TimeUnit.SECONDS);
+    }
+
+    @Override
+    public void onDeviceDisconnected() {
+        stopKeepalives();
+        salFacade.onDeviceDisconnected();
+    }
+
+    @Override
+    public void onDeviceFailed(final Throwable throwable) {
+        stopKeepalives();
+        salFacade.onDeviceFailed(throwable);
+    }
+
+    @Override
+    public void onNotification(final DOMNotification domNotification) {
+        resetKeepalive();
+        salFacade.onNotification(domNotification);
+    }
+
+    @Override
+    public void close() {
+        stopKeepalives();
+        salFacade.close();
+    }
+
+    // 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);
+
+    /**
+     * Invoke keepalive RPC and check the response. In case of any received response the keepalive
+     * is considered successful and schedules next keepalive with a fixed delay. If the response is unsuccessful (no
+     * response received, or the rcp could not even be sent) immediate reconnect is triggered as netconf session
+     * is considered inactive/failed.
+     */
+    private class Keepalive implements Runnable, FutureCallback<DOMRpcResult> {
+
+        @Override
+        public void run() {
+            LOG.trace("{}: Invoking keepalive RPC", id);
+
+            try {
+                Futures.addCallback(currentDeviceRpc.invokeRpc(PATH, KEEPALIVE_PAYLOAD), this);
+            } 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) {
+            LOG.debug("{}: Keepalive RPC successful with response: {}", id, result.getResult());
+            scheduleKeepalive();
+        }
+
+        @Override
+        public void onFailure(@Nonnull final Throwable t) {
+            LOG.warn("{}: Keepalive RPC failed. Reconnecting netconf session.", id, t);
+            reconnect();
+        }
+    }
+
+    /**
+     * Reset keepalive after each RPC response received
+     */
+    private class ResetKeepalive implements com.google.common.util.concurrent.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 t) {
+            // 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, t);
+            reconnect();
+        }
+    }
+
+    /**
+     * DOMRpcService proxy that attaches reset-keepalive-task to each RPC invocation.
+     */
+    private static final class KeepaliveDOMRpcService implements DOMRpcService {
+
+        private final DOMRpcService deviceRpc;
+        private ResetKeepalive resetKeepaliveTask;
+
+        public KeepaliveDOMRpcService(final DOMRpcService deviceRpc, final ResetKeepalive resetKeepaliveTask) {
+            this.deviceRpc = deviceRpc;
+            this.resetKeepaliveTask = resetKeepaliveTask;
+        }
+
+        @Nonnull
+        @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);
+            return domRpcResultDOMRpcExceptionCheckedFuture;
+        }
+
+        @Override
+        public <T extends DOMRpcAvailabilityListener> ListenerRegistration<T> registerRpcListener(@Nonnull final T listener) {
+            // There is no real communication with the device (yet), no reset here
+            return deviceRpc.registerRpcListener(listener);
+        }
+    }
+}
index d9121fb42697c7bb0e28359ffdf4718530c6e5f8..9dc455c0516c67fc012f1b3c7b2198368df9030a 100644 (file)
@@ -143,6 +143,20 @@ public class NetconfMessageTransformUtil {
     public static final ContainerNode CREATE_SUBSCRIPTION_RPC_CONTENT =
             Builders.containerBuilder().withNodeIdentifier(new YangInstanceIdentifier.NodeIdentifier(CREATE_SUBSCRIPTION_RPC_QNAME)).build();
 
+    public static final DataContainerChild<?, ?> EMPTY_FILTER;
+
+    static {
+        final NormalizedNodeAttrBuilder<YangInstanceIdentifier.NodeIdentifier, DOMSource, AnyXmlNode> anyXmlBuilder = Builders.anyXmlBuilder().withNodeIdentifier(toId(NETCONF_FILTER_QNAME));
+        anyXmlBuilder.withAttributes(Collections.singletonMap(NETCONF_TYPE_QNAME, SUBTREE));
+
+        final Element element = XmlUtil.createElement(BLANK_DOCUMENT, NETCONF_FILTER_QNAME.getLocalName(), Optional.of(NETCONF_FILTER_QNAME.getNamespace().toString()));
+        element.setAttributeNS(NETCONF_FILTER_QNAME.getNamespace().toString(), NETCONF_TYPE_QNAME.getLocalName(), "subtree");
+
+        anyXmlBuilder.withValue(new DOMSource(element));
+
+        EMPTY_FILTER = anyXmlBuilder.build();
+    }
+
     public static DataContainerChild<?, ?> toFilterStructure(final YangInstanceIdentifier identifier, final SchemaContext ctx) {
         final NormalizedNodeAttrBuilder<YangInstanceIdentifier.NodeIdentifier, DOMSource, AnyXmlNode> anyXmlBuilder = Builders.anyXmlBuilder().withNodeIdentifier(toId(NETCONF_FILTER_QNAME));
         anyXmlBuilder.withAttributes(Collections.singletonMap(NETCONF_TYPE_QNAME, SUBTREE));
index 08c4de64f2d475926e74ea87cfb0e343bdd45371..799c9c89995d414b410f1aa1ad02f4a4b4756e80 100644 (file)
@@ -150,6 +150,25 @@ module odl-sal-netconf-connector-cfg {
                 }
                 default 1.5;
             }
+
+            // Keepalive configuration
+            leaf keepalive-delay {
+                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";
+            }
+
+            container keepalive-executor {
+                uses config:service-ref {
+                    refine type {
+                        mandatory false;
+                        config:required-identity th:scheduled-threadpool;
+                    }
+                }
+
+                description "Dedicated solely to keepalive execution";
+            }
         }
     }
 }
\ No newline at end of file
diff --git a/opendaylight/md-sal/sal-netconf-connector/src/test/java/org/opendaylight/controller/sal/connect/netconf/sal/KeepaliveSalFacadeTest.java b/opendaylight/md-sal/sal-netconf-connector/src/test/java/org/opendaylight/controller/sal/connect/netconf/sal/KeepaliveSalFacadeTest.java
new file mode 100644 (file)
index 0000000..71d1b83
--- /dev/null
@@ -0,0 +1,159 @@
+/*
+ * Copyright (c) 2015 Cisco Systems, Inc. and others.  All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+package org.opendaylight.controller.sal.connect.netconf.sal;
+
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.timeout;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+import com.google.common.util.concurrent.Futures;
+import java.net.InetSocketAddress;
+import java.util.concurrent.Executors;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+import org.opendaylight.controller.md.sal.dom.api.DOMRpcResult;
+import org.opendaylight.controller.md.sal.dom.api.DOMRpcService;
+import org.opendaylight.controller.md.sal.dom.spi.DefaultDOMRpcResult;
+import org.opendaylight.controller.sal.connect.api.RemoteDeviceHandler;
+import org.opendaylight.controller.sal.connect.netconf.listener.NetconfDeviceCommunicator;
+import org.opendaylight.controller.sal.connect.netconf.listener.NetconfSessionPreferences;
+import org.opendaylight.controller.sal.connect.netconf.util.NetconfMessageTransformUtil;
+import org.opendaylight.controller.sal.connect.util.RemoteDeviceId;
+import org.opendaylight.yangtools.yang.common.RpcError;
+import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
+import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
+import org.opendaylight.yangtools.yang.data.impl.schema.Builders;
+import org.opendaylight.yangtools.yang.model.api.SchemaContext;
+import org.opendaylight.yangtools.yang.model.api.SchemaPath;
+
+public class KeepaliveSalFacadeTest {
+
+    private static final RemoteDeviceId REMOTE_DEVICE_ID = new RemoteDeviceId("test", new InetSocketAddress("localhost", 22));
+
+    @Mock
+    private RemoteDeviceHandler<NetconfSessionPreferences> underlyingSalFacade;
+
+    private static java.util.concurrent.ScheduledExecutorService executorService;
+
+    @Mock
+    private NetconfDeviceCommunicator listener;
+    @Mock
+    private DOMRpcService deviceRpc;
+
+    private DOMRpcService proxyRpc;
+
+    @Before
+    public void setUp() throws Exception {
+        executorService = Executors.newScheduledThreadPool(1);
+
+        MockitoAnnotations.initMocks(this);
+
+        doNothing().when(listener).disconnect();
+        doReturn("mockedRpc").when(deviceRpc).toString();
+        doNothing().when(underlyingSalFacade).onDeviceConnected(
+                any(SchemaContext.class), any(NetconfSessionPreferences.class), any(DOMRpcService.class));
+    }
+
+    @After
+    public void tearDown() throws Exception {
+        executorService.shutdownNow();
+    }
+
+    @Test
+    public void testKeepaliveSuccess() throws Exception {
+        final DOMRpcResult result = new DefaultDOMRpcResult(Builders.containerBuilder().withNodeIdentifier(
+                new YangInstanceIdentifier.NodeIdentifier(NetconfMessageTransformUtil.NETCONF_RUNNING_QNAME)).build());
+
+        doReturn(Futures.immediateCheckedFuture(result)).when(deviceRpc).invokeRpc(any(SchemaPath.class), any(NormalizedNode.class));
+
+        final KeepaliveSalFacade keepaliveSalFacade =
+                new KeepaliveSalFacade(REMOTE_DEVICE_ID, underlyingSalFacade, executorService, 1L);
+        keepaliveSalFacade.setListener(listener);
+
+        keepaliveSalFacade.onDeviceConnected(null, null, deviceRpc);
+
+        verify(underlyingSalFacade).onDeviceConnected(
+                any(SchemaContext.class), any(NetconfSessionPreferences.class), any(DOMRpcService.class));
+
+        verify(deviceRpc, timeout(15000).times(5)).invokeRpc(any(SchemaPath.class), any(NormalizedNode.class));
+    }
+
+    @Test
+    public void testKeepaliveFail() throws Exception {
+        final DOMRpcResult result = new DefaultDOMRpcResult(Builders.containerBuilder().withNodeIdentifier(
+                new YangInstanceIdentifier.NodeIdentifier(NetconfMessageTransformUtil.NETCONF_RUNNING_QNAME)).build());
+
+        final DOMRpcResult resultFail = new DefaultDOMRpcResult(mock(RpcError.class));
+
+        doReturn(Futures.immediateCheckedFuture(result))
+                .doReturn(Futures.immediateCheckedFuture(resultFail))
+                .doReturn(Futures.immediateFailedCheckedFuture(new IllegalStateException("illegal-state")))
+                .when(deviceRpc).invokeRpc(any(SchemaPath.class), any(NormalizedNode.class));
+
+        final KeepaliveSalFacade keepaliveSalFacade =
+                new KeepaliveSalFacade(REMOTE_DEVICE_ID, underlyingSalFacade, executorService, 1L);
+        keepaliveSalFacade.setListener(listener);
+
+        keepaliveSalFacade.onDeviceConnected(null, null, deviceRpc);
+
+        verify(underlyingSalFacade).onDeviceConnected(
+                any(SchemaContext.class), any(NetconfSessionPreferences.class), any(DOMRpcService.class));
+
+        // 1 failed that results in disconnect
+        verify(listener, timeout(15000).times(1)).disconnect();
+        // 3 attempts total
+        verify(deviceRpc, times(3)).invokeRpc(any(SchemaPath.class), any(NormalizedNode.class));
+
+        // Reconnect with same keepalive responses
+        doReturn(Futures.immediateCheckedFuture(result))
+                .doReturn(Futures.immediateCheckedFuture(resultFail))
+                .doReturn(Futures.immediateFailedCheckedFuture(new IllegalStateException("illegal-state")))
+                .when(deviceRpc).invokeRpc(any(SchemaPath.class), any(NormalizedNode.class));
+
+        keepaliveSalFacade.onDeviceConnected(null, null, deviceRpc);
+
+        // 1 failed that results in disconnect, 2 total with previous fail
+        verify(listener, timeout(15000).times(2)).disconnect();
+        // 6 attempts now total
+        verify(deviceRpc, times(3 * 2)).invokeRpc(any(SchemaPath.class), any(NormalizedNode.class));
+    }
+
+    @Test
+    public void testNonKeepaliveRpcFailure() throws Exception {
+        doAnswer(new Answer() {
+            @Override
+            public Object answer(final InvocationOnMock invocationOnMock) throws Throwable {
+                proxyRpc = (DOMRpcService) invocationOnMock.getArguments()[2];
+                return null;
+            }
+        }).when(underlyingSalFacade).onDeviceConnected(any(SchemaContext.class), any(NetconfSessionPreferences.class), any(DOMRpcService.class));
+
+        doReturn(Futures.immediateFailedCheckedFuture(new IllegalStateException("illegal-state")))
+                .when(deviceRpc).invokeRpc(any(SchemaPath.class), any(NormalizedNode.class));
+
+        final KeepaliveSalFacade keepaliveSalFacade =
+                new KeepaliveSalFacade(REMOTE_DEVICE_ID, underlyingSalFacade, executorService, 100L);
+        keepaliveSalFacade.setListener(listener);
+
+        keepaliveSalFacade.onDeviceConnected(null, null, deviceRpc);
+
+        proxyRpc.invokeRpc(mock(SchemaPath.class), mock(NormalizedNode.class));
+
+        verify(listener, times(1)).disconnect();
+    }
+}
\ No newline at end of file