library/impl/src/main/java/org/opendaylight/ovsdb/lib/impl/StalePassiveConnectionService.java

   1 /*
   2  * Copyright © 2016, 2017 NEC Corporation 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.ovsdb.lib.impl;
   9
  10 import com.google.common.util.concurrent.FutureCallback;
  11 import com.google.common.util.concurrent.Futures;
  12 import com.google.common.util.concurrent.MoreExecutors;
  13
  14 import java.util.HashMap;
  15 import java.util.HashSet;
  16 import java.util.List;
  17 import java.util.Map;
  18 import java.util.Optional;
  19 import java.util.Set;
  20 import java.util.concurrent.ConcurrentHashMap;
  21 import java.util.function.Function;
  22
  23 import org.opendaylight.ovsdb.lib.OvsdbClient;
  24 import org.slf4j.Logger;
  25 import org.slf4j.LoggerFactory;
  26
  27 /**
  28  * StalePassiveConnectionService provides functionalities to clean up stale passive connections
  29  * from the same node before new connection request arrives, especially for connection flapping scenarios.
  30  *
  31  * <p>When new connection arrives all connections from the same node are pinged. The pings cause
  32  * the stale netty connections to close due to IOException. Those have not been closed after a timeout
  33  * will be closed programmatically. New connection request handling is then proceeded after all
  34  * stale connections are cleaned up in the OvsdbConnectionService
  35  *
  36  * @author Vinh Nguyen (vinh.nguyen@hcl.com) on 6/10/16.
  37  */
  38 public class StalePassiveConnectionService implements AutoCloseable {
  39
  40     private static final Logger LOG = LoggerFactory.getLogger(StalePassiveConnectionService.class);
  41
  42     private final Map<OvsdbClient, Set<OvsdbClient>> pendingClients = new ConcurrentHashMap<>();
  43     private final Function<OvsdbClient, Void> clientNotificationCallback;
  44
  45     public StalePassiveConnectionService(Function<OvsdbClient, Void> clientNotificationCallback) {
  46         this.clientNotificationCallback = clientNotificationCallback;
  47     }
  48
  49     public Map<OvsdbClient, Set<OvsdbClient>> getPendingClients() {
  50         return new HashMap<>(pendingClients);
  51     }
  52
  53     /**
  54      * This method makes sure that all stale connections from the same node are properly cleaned up before processing
  55      * new connection request.
  56      *
  57      * @param newOvsdbClient the connecting OvsdbClient
  58      * @param clientsFromSameNode list of existing OvsdbClients from the same node as the new OvsdbClient
  59      */
  60     public void handleNewPassiveConnection(final OvsdbClient newOvsdbClient,
  61                                            final List<OvsdbClient> clientsFromSameNode) {
  62         LOG.info("Adding client to pending list {}", newOvsdbClient.getConnectionInfo());
  63         pendingClients.put(newOvsdbClient, new HashSet<>());
  64         /*
  65             if old client echo succeeds
  66                do not notify new client as it has to wait
  67             else
  68                 if all old clients got disconnected/echo failed notify the new client
  69          */
  70         for (final OvsdbClient oldClient : clientsFromSameNode) {
  71             pendingClients.get(newOvsdbClient).add(oldClient);
  72             LOG.info("Echo testing client {}", oldClient.getConnectionInfo());
  73             Futures.addCallback(oldClient.echo(),
  74                 new FutureCallback<List<String>>() {
  75                     @Override
  76                     public void onSuccess(List<String> result) {
  77                         //old client still active
  78                         LOG.info("Echo testing of old client {} succeeded", oldClient.getConnectionInfo());
  79                     }
  80
  81                     @Override
  82                     public void onFailure(Throwable throwable) {
  83                         LOG.info("Echo testing of old client {} failed, disconnect and notify clients",
  84                                 oldClient.getConnectionInfo());
  85                         //disconnect the old client to cleanup, so that new connection can proceed
  86                         oldClient.disconnect();
  87                         onInactiveClient(oldClient);
  88                     }
  89                 }, MoreExecutors.directExecutor());
  90         }
  91     }
  92
  93     /**
  94      * Notify the service that the given client has disconnected.
  95      * @param disconnectedClient the client just disconnected
  96      */
  97     public synchronized void clientDisconnected(OvsdbClient disconnectedClient) {
  98         LOG.info("Client disconnected {}", disconnectedClient.getConnectionInfo());
  99         onInactiveClient(disconnectedClient);
 100     }
 101
 102     public synchronized void onInactiveClient(OvsdbClient disconnectedClient) {
 103         pendingClients.remove(disconnectedClient);
 104         pendingClients.entrySet().stream().forEach(entry -> entry.getValue().remove(disconnectedClient));
 105         Optional<OvsdbClient> clientOptional = pendingClients.entrySet().stream()
 106                 .filter(entry -> entry.getValue().isEmpty())
 107                 .map(entry -> entry.getKey())
 108                 .findFirst();
 109         if (clientOptional.isPresent()) {
 110             LOG.info("Sending notification for client {}", clientOptional.get().getConnectionInfo());
 111             pendingClients.remove(clientOptional.get());
 112             clientNotificationCallback.apply(clientOptional.get());
 113         }
 114     }
 115
 116     @Override
 117     public void close() {
 118     }
 119 }