Bump odlparent to 5.0.0
[netconf.git] / 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 }