netconf/sal-netconf-connector/src/main/java/org/opendaylight/netconf/sal/connect/netconf/sal/KeepaliveSalFacade.java

   1 /*
   2  * Copyright (c) 2015 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8 package org.opendaylight.netconf.sal.connect.netconf.sal;
   9
  10 import static com.google.common.base.Preconditions.checkState;
  11 import static org.opendaylight.netconf.sal.connect.netconf.util.NetconfBaseOps.getSourceNode;
  12 import static org.opendaylight.netconf.sal.connect.netconf.util.NetconfMessageTransformUtil.NETCONF_GET_CONFIG_NODEID;
  13 import static org.opendaylight.netconf.sal.connect.netconf.util.NetconfMessageTransformUtil.NETCONF_GET_CONFIG_PATH;
  14 import static org.opendaylight.netconf.sal.connect.netconf.util.NetconfMessageTransformUtil.NETCONF_RUNNING_QNAME;
  15
  16 import com.google.common.util.concurrent.FluentFuture;
  17 import com.google.common.util.concurrent.FutureCallback;
  18 import com.google.common.util.concurrent.MoreExecutors;
  19 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
  20 import java.util.concurrent.ScheduledExecutorService;
  21 import java.util.concurrent.ScheduledFuture;
  22 import java.util.concurrent.TimeUnit;
  23 import java.util.concurrent.atomic.AtomicBoolean;
  24 import org.opendaylight.mdsal.dom.api.DOMActionService;
  25 import org.opendaylight.mdsal.dom.api.DOMNotification;
  26 import org.opendaylight.mdsal.dom.api.DOMRpcAvailabilityListener;
  27 import org.opendaylight.mdsal.dom.api.DOMRpcResult;
  28 import org.opendaylight.mdsal.dom.api.DOMRpcService;
  29 import org.opendaylight.netconf.sal.connect.api.RemoteDeviceHandler;
  30 import org.opendaylight.netconf.sal.connect.netconf.listener.NetconfDeviceCommunicator;
  31 import org.opendaylight.netconf.sal.connect.netconf.listener.NetconfSessionPreferences;
  32 import org.opendaylight.netconf.sal.connect.netconf.util.NetconfMessageTransformUtil;
  33 import org.opendaylight.netconf.sal.connect.util.RemoteDeviceId;
  34 import org.opendaylight.yangtools.concepts.ListenerRegistration;
  35 import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode;
  36 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  37 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
  38 import org.opendaylight.yangtools.yang.model.api.SchemaPath;
  39 import org.slf4j.Logger;
  40 import org.slf4j.LoggerFactory;
  41
  42 /**
  43  * SalFacade proxy that invokes keepalive RPCs to prevent session shutdown from remote device
  44  * and to detect incorrect session drops (netconf session is inactive, but TCP/SSH connection is still present).
  45  * The keepalive RPC is a get-config with empty filter.
  46  */
  47 public final class KeepaliveSalFacade implements RemoteDeviceHandler<NetconfSessionPreferences> {
  48
  49     private static final Logger LOG = LoggerFactory.getLogger(KeepaliveSalFacade.class);
  50
  51     // 2 minutes keepalive delay by default
  52     private static final long DEFAULT_DELAY = TimeUnit.MINUTES.toSeconds(2);
  53
  54     // 1 minute transaction timeout by default
  55     private static final long DEFAULT_TRANSACTION_TIMEOUT_MILLI = TimeUnit.MILLISECONDS.toMillis(60000);
  56
  57     private final RemoteDeviceId id;
  58     private final RemoteDeviceHandler<NetconfSessionPreferences> salFacade;
  59     private final ScheduledExecutorService executor;
  60     private final long keepaliveDelaySeconds;
  61     private final ResetKeepalive resetKeepaliveTask;
  62     private final long defaultRequestTimeoutMillis;
  63
  64     private volatile NetconfDeviceCommunicator listener;
  65     private volatile ScheduledFuture<?> currentKeepalive;
  66     private volatile DOMRpcService currentDeviceRpc;
  67     private final AtomicBoolean lastKeepAliveSucceeded = new AtomicBoolean(false);
  68
  69     public KeepaliveSalFacade(final RemoteDeviceId id, final RemoteDeviceHandler<NetconfSessionPreferences> salFacade,
  70                               final ScheduledExecutorService executor, final long keepaliveDelaySeconds,
  71                               final long defaultRequestTimeoutMillis) {
  72         this.id = id;
  73         this.salFacade = salFacade;
  74         this.executor = executor;
  75         this.keepaliveDelaySeconds = keepaliveDelaySeconds;
  76         this.defaultRequestTimeoutMillis = defaultRequestTimeoutMillis;
  77         this.resetKeepaliveTask = new ResetKeepalive();
  78     }
  79
  80     public KeepaliveSalFacade(final RemoteDeviceId id, final RemoteDeviceHandler<NetconfSessionPreferences> salFacade,
  81                               final ScheduledExecutorService executor) {
  82         this(id, salFacade, executor, DEFAULT_DELAY, DEFAULT_TRANSACTION_TIMEOUT_MILLI);
  83     }
  84
  85     /**
  86      * Set the netconf session listener whenever ready.
  87      *
  88      * @param listener netconf session listener
  89      */
  90     public void setListener(final NetconfDeviceCommunicator listener) {
  91         this.listener = listener;
  92     }
  93
  94     /**
  95      * Just cancel current keepalive task.
  96      * If its already started, let it finish ... not such a big deal.
  97      *
  98      * <p>
  99      * Then schedule next keepalive.
 100      */
 101     void resetKeepalive() {
 102         LOG.trace("{}: Resetting netconf keepalive timer", id);
 103         if (currentKeepalive != null) {
 104             currentKeepalive.cancel(false);
 105         }
 106         scheduleKeepalives();
 107     }
 108
 109     /**
 110      * Cancel current keepalive and also reset current deviceRpc.
 111      */
 112     private void stopKeepalives() {
 113         if (currentKeepalive != null) {
 114             currentKeepalive.cancel(false);
 115         }
 116         currentDeviceRpc = null;
 117     }
 118
 119     void reconnect() {
 120         checkState(listener != null, "%s: Unable to reconnect, session listener is missing", id);
 121         stopKeepalives();
 122         LOG.info("{}: Reconnecting inactive netconf session", id);
 123         listener.disconnect();
 124     }
 125
 126     @Override
 127     public void onDeviceConnected(final SchemaContext remoteSchemaContext,
 128                           final NetconfSessionPreferences netconfSessionPreferences, final DOMRpcService deviceRpc) {
 129         onDeviceConnected(remoteSchemaContext, netconfSessionPreferences, deviceRpc, null);
 130     }
 131
 132     @Override
 133     public void onDeviceConnected(final SchemaContext remoteSchemaContext,
 134             final NetconfSessionPreferences netconfSessionPreferences, final DOMRpcService deviceRpc,
 135             final DOMActionService deviceAction) {
 136         this.currentDeviceRpc = deviceRpc;
 137         final DOMRpcService deviceRpc1 =
 138                 new KeepaliveDOMRpcService(deviceRpc, resetKeepaliveTask, defaultRequestTimeoutMillis, executor,
 139                         new ResponseWaitingScheduler());
 140
 141         salFacade.onDeviceConnected(remoteSchemaContext, netconfSessionPreferences, deviceRpc1, deviceAction);
 142
 143         LOG.debug("{}: Netconf session initiated, starting keepalives", id);
 144         scheduleKeepalives();
 145     }
 146
 147     private void scheduleKeepalives() {
 148         lastKeepAliveSucceeded.set(true);
 149         checkState(currentDeviceRpc != null);
 150         LOG.trace("{}: Scheduling keepalives every  {} {}", id, keepaliveDelaySeconds, TimeUnit.SECONDS);
 151         currentKeepalive = executor.scheduleWithFixedDelay(new Keepalive(),
 152           keepaliveDelaySeconds, keepaliveDelaySeconds, TimeUnit.SECONDS);
 153     }
 154
 155     @Override
 156     public void onDeviceDisconnected() {
 157         stopKeepalives();
 158         salFacade.onDeviceDisconnected();
 159     }
 160
 161     @Override
 162     public void onDeviceFailed(final Throwable throwable) {
 163         stopKeepalives();
 164         salFacade.onDeviceFailed(throwable);
 165     }
 166
 167     @Override
 168     public void onNotification(final DOMNotification domNotification) {
 169         resetKeepalive();
 170         salFacade.onNotification(domNotification);
 171     }
 172
 173     @Override
 174     public void close() {
 175         stopKeepalives();
 176         salFacade.close();
 177     }
 178
 179     // Keepalive RPC static resources
 180     private static final ContainerNode KEEPALIVE_PAYLOAD = NetconfMessageTransformUtil.wrap(NETCONF_GET_CONFIG_NODEID,
 181             getSourceNode(NETCONF_RUNNING_QNAME), NetconfMessageTransformUtil.EMPTY_FILTER);
 182
 183     /**
 184      * Invoke keepalive RPC and check the response. In case of any received response the keepalive
 185      * is considered successful and schedules next keepalive with a fixed delay. If the response is unsuccessful (no
 186      * response received, or the rcp could not even be sent) immediate reconnect is triggered as netconf session
 187      * is considered inactive/failed.
 188      */
 189     private class Keepalive implements Runnable, FutureCallback<DOMRpcResult> {
 190
 191         @Override
 192         public void run() {
 193             LOG.trace("{}: Invoking keepalive RPC", id);
 194
 195             try {
 196                 final boolean lastJobSucceeded = lastKeepAliveSucceeded.getAndSet(false);
 197                 if (!lastJobSucceeded) {
 198                     onFailure(new IllegalStateException("Previous keepalive timed out"));
 199                 } else {
 200                     currentDeviceRpc.invokeRpc(NETCONF_GET_CONFIG_PATH, KEEPALIVE_PAYLOAD).addCallback(this,
 201                                         MoreExecutors.directExecutor());
 202                 }
 203             } catch (final NullPointerException e) {
 204                 LOG.debug("{}: Skipping keepalive while reconnecting", id);
 205                 // Empty catch block intentional
 206                 // Do nothing. The currentDeviceRpc was null and it means we hit the reconnect window and
 207                 // attempted to send keepalive while we were reconnecting. Next keepalive will be scheduled
 208                 // after reconnect so no action necessary here.
 209             }
 210         }
 211
 212         @SuppressFBWarnings(value = "RCN_REDUNDANT_NULLCHECK_OF_NONNULL_VALUE",
 213                 justification = "Unrecognised NullableDecl")
 214         @Override
 215         public void onSuccess(final DOMRpcResult result) {
 216             // No matter what response we got, rpc-reply or rpc-error,
 217             // we got it from device so the netconf session is OK
 218             if (result == null) {
 219                 LOG.warn("{} Keepalive RPC returned null with response. Reconnecting netconf session", id);
 220                 reconnect();
 221                 return;
 222             }
 223
 224             if (result.getResult() != null) {
 225                 lastKeepAliveSucceeded.set(true);
 226             }  else if (result.getErrors() != null) {
 227                 LOG.warn("{}: Keepalive RPC failed with error: {}", id, result.getErrors());
 228                 lastKeepAliveSucceeded.set(true);
 229             } else {
 230                 LOG.warn("{} Keepalive RPC returned null with response. Reconnecting netconf session", id);
 231                 reconnect();
 232             }
 233         }
 234
 235         @Override
 236         public void onFailure(final Throwable throwable) {
 237             LOG.warn("{}: Keepalive RPC failed. Reconnecting netconf session.", id, throwable);
 238             reconnect();
 239         }
 240     }
 241
 242     /**
 243      * Reset keepalive after each RPC response received.
 244      */
 245     private class ResetKeepalive implements FutureCallback<DOMRpcResult> {
 246         @Override
 247         public void onSuccess(final DOMRpcResult result) {
 248             // No matter what response we got,
 249             // rpc-reply or rpc-error, we got it from device so the netconf session is OK.
 250             resetKeepalive();
 251         }
 252
 253         @Override
 254         public void onFailure(final Throwable throwable) {
 255             // User/Application RPC failed (The RPC did not reach the remote device or ..
 256             // TODO what other reasons could cause this ?)
 257             // There is no point in keeping this session. Reconnect.
 258             LOG.warn("{}: Rpc failure detected. Reconnecting netconf session", id, throwable);
 259             reconnect();
 260         }
 261     }
 262
 263     private final class ResponseWaitingScheduler {
 264
 265         private ScheduledFuture<?> schedule;
 266
 267         public void initScheduler(final Runnable runnable) {
 268             if (currentKeepalive != null) {
 269                 currentKeepalive.cancel(true);
 270             } else {
 271                 LOG.trace("Keepalive does not exist.");
 272             }
 273             scheduleKeepalives();
 274             //Listening on the result should be done before the keepalive rpc will be send
 275             final long delay = (keepaliveDelaySeconds * 1000) - 500;
 276             schedule = executor.schedule(runnable, delay, TimeUnit.MILLISECONDS);
 277         }
 278
 279         public void stopScheduler() {
 280             if (schedule != null) {
 281                 schedule.cancel(true);
 282             } else {
 283                 LOG.trace("Scheduler does not exist.");
 284             }
 285         }
 286     }
 287
 288     private static final class ResponseWaiting implements Runnable {
 289
 290         private final FluentFuture<DOMRpcResult> rpcResultFuture;
 291         private final ResponseWaitingScheduler responseWaitingScheduler;
 292
 293         ResponseWaiting(final ResponseWaitingScheduler responseWaitingScheduler,
 294                 final FluentFuture<DOMRpcResult> rpcResultFuture) {
 295             this.responseWaitingScheduler = responseWaitingScheduler;
 296             this.rpcResultFuture = rpcResultFuture;
 297         }
 298
 299         public void start() {
 300             LOG.trace("Start to waiting for result.");
 301             responseWaitingScheduler.initScheduler(this);
 302         }
 303
 304         public void stop() {
 305             LOG.info("Stop to waiting for result.");
 306             responseWaitingScheduler.stopScheduler();
 307         }
 308
 309         @Override
 310         public void run() {
 311             if (!rpcResultFuture.isCancelled() && !rpcResultFuture.isDone()) {
 312                 LOG.trace("Waiting for result");
 313                 responseWaitingScheduler.initScheduler(this);
 314             } else {
 315                 LOG.trace("Result has been cancelled or done.");
 316             }
 317         }
 318     }
 319
 320     /*
 321      * Request timeout task is called once the defaultRequestTimeoutMillis is
 322      * reached. At this moment, if the request is not yet finished, we cancel
 323      * it.
 324      */
 325     private static final class RequestTimeoutTask implements Runnable {
 326
 327         private final FluentFuture<DOMRpcResult> rpcResultFuture;
 328         private final ResponseWaiting responseWaiting;
 329
 330         RequestTimeoutTask(final FluentFuture<DOMRpcResult> rpcResultFuture, final ResponseWaiting responseWaiting) {
 331             this.rpcResultFuture = rpcResultFuture;
 332             this.responseWaiting = responseWaiting;
 333         }
 334
 335         @Override
 336         public void run() {
 337             if (!rpcResultFuture.isDone()) {
 338                 rpcResultFuture.cancel(true);
 339             }
 340             if (responseWaiting != null) {
 341                 responseWaiting.stop();
 342             }
 343         }
 344     }
 345
 346     /**
 347      * DOMRpcService proxy that attaches reset-keepalive-task and schedule
 348      * request-timeout-task to each RPC invocation.
 349      */
 350     public static final class KeepaliveDOMRpcService implements DOMRpcService {
 351
 352         private final DOMRpcService deviceRpc;
 353         private final ResetKeepalive resetKeepaliveTask;
 354         private final long defaultRequestTimeoutMillis;
 355         private final ScheduledExecutorService executor;
 356         private final ResponseWaitingScheduler responseWaitingScheduler;
 357
 358         KeepaliveDOMRpcService(final DOMRpcService deviceRpc, final ResetKeepalive resetKeepaliveTask,
 359                 final long defaultRequestTimeoutMillis, final ScheduledExecutorService executor,
 360                 final ResponseWaitingScheduler responseWaitingScheduler) {
 361             this.deviceRpc = deviceRpc;
 362             this.resetKeepaliveTask = resetKeepaliveTask;
 363             this.defaultRequestTimeoutMillis = defaultRequestTimeoutMillis;
 364             this.executor = executor;
 365             this.responseWaitingScheduler = responseWaitingScheduler;
 366         }
 367
 368         public DOMRpcService getDeviceRpc() {
 369             return deviceRpc;
 370         }
 371
 372         @Override
 373         public FluentFuture<DOMRpcResult> invokeRpc(final SchemaPath type, final NormalizedNode<?, ?> input) {
 374             final FluentFuture<DOMRpcResult> rpcResultFuture = deviceRpc.invokeRpc(type, input);
 375             final ResponseWaiting responseWaiting = new ResponseWaiting(responseWaitingScheduler, rpcResultFuture);
 376             responseWaiting.start();
 377             rpcResultFuture.addCallback(resetKeepaliveTask, MoreExecutors.directExecutor());
 378
 379             final RequestTimeoutTask timeoutTask = new RequestTimeoutTask(rpcResultFuture, responseWaiting);
 380             executor.schedule(timeoutTask, defaultRequestTimeoutMillis, TimeUnit.MILLISECONDS);
 381
 382             return rpcResultFuture;
 383         }
 384
 385         @Override
 386         public <T extends DOMRpcAvailabilityListener> ListenerRegistration<T> registerRpcListener(final T listener) {
 387             // There is no real communication with the device (yet), no reset here
 388             return deviceRpc.registerRpcListener(listener);
 389         }
 390     }
 391 }