plugin/src/main/java/org/opendaylight/ovsdb/plugin/impl/ConnectionServiceImpl.java

   1 /*
   2  * Copyright (c) 2013, 2015 Red Hat, 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
   9 package org.opendaylight.ovsdb.plugin.impl;
  10
  11 import io.netty.channel.ChannelHandler;
  12
  13 import java.io.IOException;
  14 import java.net.InetAddress;
  15 import java.util.ArrayList;
  16 import java.util.Collection;
  17 import java.util.List;
  18 import java.util.Map;
  19 import java.util.Set;
  20 import java.util.concurrent.ConcurrentHashMap;
  21 import java.util.concurrent.ConcurrentMap;
  22 import java.util.concurrent.ExecutionException;
  23
  24 import org.opendaylight.ovsdb.lib.MonitorCallBack;
  25 import org.opendaylight.ovsdb.lib.OvsdbClient;
  26 import org.opendaylight.ovsdb.lib.OvsdbConnection;
  27 import org.opendaylight.ovsdb.lib.OvsdbConnectionInfo;
  28 import org.opendaylight.ovsdb.lib.OvsdbConnectionListener;
  29 import org.opendaylight.ovsdb.lib.message.MonitorRequest;
  30 import org.opendaylight.ovsdb.lib.message.MonitorRequestBuilder;
  31 import org.opendaylight.ovsdb.lib.message.MonitorSelect;
  32 import org.opendaylight.ovsdb.lib.message.TableUpdates;
  33 import org.opendaylight.ovsdb.lib.schema.DatabaseSchema;
  34 import org.opendaylight.ovsdb.lib.schema.GenericTableSchema;
  35 import org.opendaylight.ovsdb.lib.schema.TableSchema;
  36 import org.opendaylight.ovsdb.plugin.api.Connection;
  37 import org.opendaylight.ovsdb.plugin.api.ConnectionConstants;
  38 import org.opendaylight.ovsdb.plugin.api.Status;
  39 import org.opendaylight.ovsdb.plugin.api.StatusCode;
  40 import org.opendaylight.ovsdb.plugin.api.OvsdbConnectionService;
  41 import org.opendaylight.ovsdb.plugin.api.OvsdbInventoryService;
  42 import org.opendaylight.ovsdb.utils.config.ConfigProperties;
  43 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
  44 import org.slf4j.Logger;
  45 import org.slf4j.LoggerFactory;
  46
  47 import com.google.common.collect.Lists;
  48
  49
  50 /**
  51  * Represents the openflow plugin component in charge of programming the flows
  52  * the flow programming and relay them to functional modules above SAL.
  53  */
  54 public class ConnectionServiceImpl implements OvsdbConnectionService,
  55                                               OvsdbConnectionListener {
  56     private static final Logger LOG = LoggerFactory.getLogger(ConnectionServiceImpl.class);
  57
  58     // Properties that can be set in config.ini
  59     private static final Integer DEFAULT_OVSDB_PORT = 6640;
  60     private static final String OVSDB_LISTENPORT = "ovsdb.listenPort";
  61
  62
  63     public void putOvsdbConnection (String identifier, Connection connection) {
  64         ovsdbConnections.put(identifier, connection);
  65     }
  66
  67     private ConcurrentMap<String, Connection> ovsdbConnections = new ConcurrentHashMap<String, Connection>();
  68     private List<ChannelHandler> handlers = null;
  69
  70     private volatile OvsdbInventoryService ovsdbInventoryService;
  71     private volatile OvsdbConnection connectionLib;
  72
  73     public void setOvsdbInventoryService(OvsdbInventoryService inventoryService) {
  74         this.ovsdbInventoryService = inventoryService;
  75     }
  76
  77     public void setOvsdbConnection(OvsdbConnection ovsdbConnection) {
  78         this.connectionLib = ovsdbConnection;
  79     }
  80
  81     public void init() {
  82     }
  83
  84     /**
  85      * Function called by the dependency manager when at least one dependency
  86      * become unsatisfied or when the component is shutting down because for
  87      * example bundle is being stopped.
  88      */
  89     void destroy() {
  90     }
  91
  92     /**
  93      * Function called by dependency manager after "init ()" is called and after
  94      * the services provided by the class are registered in the service registry
  95      */
  96     void start() {
  97         /* Start ovsdb server before getting connection clients */
  98         String portString = ConfigProperties.getProperty(OvsdbConnectionService.class, OVSDB_LISTENPORT);
  99         int ovsdbListenPort = DEFAULT_OVSDB_PORT;
 100         if (portString != null) {
 101             ovsdbListenPort = Integer.parseInt(portString);
 102         }
 103
 104         if (!connectionLib.startOvsdbManager(ovsdbListenPort)) {
 105             LOG.warn("Start OVSDB manager call from ConnectionService was not necessary");
 106         }
 107
 108         /* Then get connection clients */
 109         Collection<OvsdbClient> connections = connectionLib.getConnections();
 110         for (OvsdbClient client : connections) {
 111             LOG.info("CONNECT start connected clients client = {}", client);
 112             this.connected(client);
 113         }
 114     }
 115
 116     /**
 117      * Function called by the dependency manager before the services exported by
 118      * the component are unregistered, this will be followed by a "destroy ()"
 119      * calls
 120      */
 121     void stopping() {
 122         for (Connection connection : ovsdbConnections.values()) {
 123             connection.disconnect();
 124         }
 125     }
 126
 127     public Status disconnect(Node node) {
 128         Connection connection = getConnection(node);
 129         if (connection != null) {
 130             ovsdbConnections.remove(normalizeId(node.getId().getValue()));
 131             connection.disconnect();
 132             ovsdbInventoryService.removeNode(node);
 133             return new Status(StatusCode.SUCCESS);
 134         } else {
 135             return new Status(StatusCode.NOTFOUND);
 136         }
 137     }
 138
 139     public Node connect(String identifier, Map<ConnectionConstants, String> params) {
 140         InetAddress address;
 141         Integer port;
 142
 143         try {
 144             address = InetAddress.getByName(params.get(ConnectionConstants.ADDRESS));
 145         } catch (Exception e) {
 146             LOG.error("Unable to resolve {}", params.get(ConnectionConstants.ADDRESS), e);
 147             return null;
 148         }
 149
 150         try {
 151             port = Integer.parseInt(params.get(ConnectionConstants.PORT));
 152             if (port == 0) {
 153                 port = DEFAULT_OVSDB_PORT;
 154             }
 155         } catch (Exception e) {
 156             port = DEFAULT_OVSDB_PORT;
 157         }
 158
 159         try {
 160             OvsdbClient client = connectionLib.connect(address, port);
 161             return handleNewConnection(identifier, client);
 162         } catch (InterruptedException e) {
 163             LOG.error("Thread was interrupted during connect", e);
 164         } catch (ExecutionException e) {
 165             LOG.error("ExecutionException in handleNewConnection for identifier " + identifier, e);
 166         }
 167         return null;
 168     }
 169
 170     public List<ChannelHandler> getHandlers() {
 171         return handlers;
 172     }
 173
 174     public void setHandlers(List<ChannelHandler> handlers) {
 175         this.handlers = handlers;
 176     }
 177
 178     private String normalizeId (String identifier) {
 179         String id = identifier;
 180
 181         String[] pair = identifier.split("\\|");
 182         if (pair[0].equals("OVS")) {
 183             id = pair[1];
 184         }
 185
 186         return id;
 187     }
 188
 189     @Override
 190     public Connection getConnection(Node node) {
 191         return ovsdbConnections.get(normalizeId(node.getId().getValue()));
 192     }
 193
 194     @Override
 195     public Node getNode (String identifier) {
 196         Connection connection = ovsdbConnections.get(normalizeId(identifier));
 197         if (connection != null) {
 198             return connection.getNode();
 199         } else {
 200             return null;
 201         }
 202     }
 203
 204     @Override
 205     public List<Node> getNodes() {
 206         List<Node> nodes = new ArrayList<>();
 207         for (Connection connection : ovsdbConnections.values()) {
 208             nodes.add(connection.getNode());
 209         }
 210         return nodes;
 211     }
 212
 213     private Node handleNewConnection(String identifier, OvsdbClient client) throws InterruptedException, ExecutionException {
 214         Connection connection = new Connection(identifier, client);
 215         Node node = connection.getNode();
 216         ovsdbConnections.put(identifier, connection);
 217         List<String> dbs = client.getDatabases().get();
 218         for (String db : dbs) {
 219             client.getSchema(db).get();
 220         }
 221         // Keeping the Initial inventory update(s) on its own thread.
 222         new Thread() {
 223             Connection connection;
 224             String identifier;
 225
 226             @Override
 227             public void run() {
 228                 try {
 229                     LOG.info("Initialize inventory for {}", connection.toString());
 230                     initializeInventoryForNewNode(connection);
 231                 } catch (InterruptedException | ExecutionException | IOException e) {
 232                     LOG.error("Failed to initialize inventory for node with identifier {}", identifier, e);
 233                     ovsdbConnections.remove(identifier);
 234                 }
 235             }
 236             public Thread initializeConnectionParams(String identifier, Connection connection) {
 237                 this.identifier = identifier;
 238                 this.connection = connection;
 239                 return this;
 240             }
 241         }.initializeConnectionParams(identifier, connection).start();
 242         return node;
 243     }
 244
 245     public void channelClosed(Node node) throws Exception {
 246         LOG.info("Connection to Node : {} closed", node);
 247         disconnect(node);
 248         ovsdbInventoryService.removeNode(node);
 249     }
 250
 251     private void initializeInventoryForNewNode (Connection connection) throws InterruptedException, ExecutionException, IOException {
 252         OvsdbClient client = connection.getClient();
 253         InetAddress address = client.getConnectionInfo().getRemoteAddress();
 254         int port = client.getConnectionInfo().getRemotePort();
 255
 256         List<String> databases = client.getDatabases().get();
 257         if (databases == null) {
 258             LOG.error("Unable to get Databases for the ovsdb connection : {}", client.getConnectionInfo());
 259             return;
 260         }
 261         for (String database : databases) {
 262             DatabaseSchema dbSchema = client.getSchema(database).get();
 263             TableUpdates updates = this.monitorTables(connection.getNode(), dbSchema);
 264             ovsdbInventoryService.processTableUpdates(connection.getNode(), dbSchema.getName(), updates);
 265         }
 266         LOG.info("Notifying Inventory Listeners for Node Added: {}", connection.getNode().toString());
 267         ovsdbInventoryService.notifyNodeAdded(connection.getNode(), address, port);
 268     }
 269
 270     public TableUpdates monitorTables(Node node, DatabaseSchema dbSchema) throws ExecutionException, InterruptedException, IOException {
 271         Connection connection = getConnection(node);
 272         OvsdbClient client = connection.getClient();
 273         if (dbSchema == null) {
 274             LOG.error("Unable to get Database Schema for the ovsdb connection : {}", client.getConnectionInfo());
 275             return null;
 276         }
 277         Set<String> tables = dbSchema.getTables();
 278         if (tables == null) {
 279             LOG.warn("Database {} without any tables. Strange !", dbSchema.getName());
 280             return null;
 281         }
 282         List<MonitorRequest<GenericTableSchema>> monitorRequests = Lists.newArrayList();
 283         for (String tableName : tables) {
 284             GenericTableSchema tableSchema = dbSchema.table(tableName, GenericTableSchema.class);
 285             monitorRequests.add(this.getAllColumnsMonitorRequest(tableSchema));
 286         }
 287         return client.monitor(dbSchema, monitorRequests, new UpdateMonitor(node));
 288     }
 289
 290     /**
 291      * As per RFC 7047, section 4.1.5, if a Monitor request is sent without any columns, the update response will not include
 292      * the _uuid column.
 293      * ----------------------------------------------------------------------------------------------------------------------------------
 294      * Each &lt;monitor-request&gt; specifies one or more columns and the manner in which the columns (or the entire table) are to be monitored.
 295      * The "columns" member specifies the columns whose values are monitored. It MUST NOT contain duplicates.
 296      * If "columns" is omitted, all columns in the table, except for "_uuid", are monitored.
 297      * ----------------------------------------------------------------------------------------------------------------------------------
 298      * In order to overcome this limitation, this method
 299      *
 300      * @return MonitorRequest that includes all the Bridge Columns including _uuid
 301      */
 302     public <T extends TableSchema<T>> MonitorRequest<T> getAllColumnsMonitorRequest (T tableSchema) {
 303         Set<String> columns = tableSchema.getColumns();
 304         MonitorRequestBuilder<T> monitorBuilder = MonitorRequestBuilder.builder(tableSchema);
 305         for (String column : columns) {
 306             monitorBuilder.addColumn(column);
 307         }
 308         return monitorBuilder.with(new MonitorSelect(true, true, true, true)).build();
 309     }
 310
 311     private class UpdateMonitor implements MonitorCallBack {
 312         Node node = null;
 313         public UpdateMonitor(Node node) {
 314             this.node = node;
 315         }
 316
 317         @Override
 318         public void update(TableUpdates result, DatabaseSchema dbSchema) {
 319             ovsdbInventoryService.processTableUpdates(node, dbSchema.getName(), result);
 320         }
 321
 322         @Override
 323         public void exception(Throwable t) {
 324             System.out.println("Exception t = " + t);
 325         }
 326     }
 327
 328     private String getConnectionIdentifier(OvsdbClient client) {
 329         OvsdbConnectionInfo info = client.getConnectionInfo();
 330         return info.getRemoteAddress().getHostAddress()+":"+info.getRemotePort();
 331     }
 332
 333
 334     @Override
 335     public void connected(OvsdbClient client) {
 336         String identifier = getConnectionIdentifier(client);
 337         try {
 338             this.handleNewConnection(identifier, client);
 339         } catch (InterruptedException | ExecutionException e) {
 340             e.printStackTrace();
 341         }
 342     }
 343
 344     @Override
 345     public void disconnected(OvsdbClient client) {
 346         Connection connection = ovsdbConnections.get(this.getConnectionIdentifier(client));
 347         if (connection == null) {
 348             return;
 349         }
 350         this.disconnect(connection.getNode());
 351     }
 352 }