openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/registry/flow/DeviceFlowRegistryImpl.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.openflowplugin.impl.registry.flow;
   9
  10 import com.google.common.annotations.VisibleForTesting;
  11 import com.google.common.base.Optional;
  12 import com.google.common.collect.BiMap;
  13 import com.google.common.collect.HashBiMap;
  14 import com.google.common.util.concurrent.CheckedFuture;
  15 import com.google.common.util.concurrent.FutureCallback;
  16 import com.google.common.util.concurrent.Futures;
  17 import com.google.common.util.concurrent.ListenableFuture;
  18 import java.util.ArrayList;
  19 import java.util.Arrays;
  20 import java.util.Collection;
  21 import java.util.Iterator;
  22 import java.util.List;
  23 import java.util.Map;
  24 import java.util.Objects;
  25 import java.util.concurrent.atomic.AtomicInteger;
  26 import java.util.function.BiConsumer;
  27 import java.util.function.Consumer;
  28 import javax.annotation.concurrent.GuardedBy;
  29 import javax.annotation.concurrent.ThreadSafe;
  30 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  31 import org.opendaylight.controller.md.sal.binding.api.ReadOnlyTransaction;
  32 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  33 import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
  34 import org.opendaylight.openflowplugin.api.openflow.registry.flow.DeviceFlowRegistry;
  35 import org.opendaylight.openflowplugin.api.openflow.registry.flow.FlowDescriptor;
  36 import org.opendaylight.openflowplugin.api.openflow.registry.flow.FlowRegistryKey;
  37 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
  38 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowId;
  39 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow;
  40 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
  41 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.NodeKey;
  42 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  43 import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier;
  44 import org.slf4j.Logger;
  45 import org.slf4j.LoggerFactory;
  46
  47 @ThreadSafe
  48 public class DeviceFlowRegistryImpl implements DeviceFlowRegistry {
  49     private static final Logger LOG = LoggerFactory.getLogger(DeviceFlowRegistryImpl.class);
  50     private static final String ALIEN_SYSTEM_FLOW_ID = "#UF$TABLE*";
  51     private static final AtomicInteger UNACCOUNTED_FLOWS_COUNTER = new AtomicInteger(0);
  52
  53     private final BiMap<FlowRegistryKey, FlowDescriptor> flowRegistry = HashBiMap.create();
  54     private final List<FlowRegistryKey> marks = new ArrayList<>();
  55     private final DataBroker dataBroker;
  56     private final KeyedInstanceIdentifier<Node, NodeKey> instanceIdentifier;
  57     private final List<ListenableFuture<List<Optional<FlowCapableNode>>>> lastFillFutures = new ArrayList<>();
  58     private final Consumer<Flow> flowConsumer;
  59
  60     public DeviceFlowRegistryImpl(final short version, final DataBroker dataBroker, final KeyedInstanceIdentifier<Node, NodeKey> instanceIdentifier) {
  61         this.dataBroker = dataBroker;
  62         this.instanceIdentifier = instanceIdentifier;
  63
  64         // Specifies what to do with flow read from datastore
  65         flowConsumer = flow -> {
  66             // Create flow registry key from flow
  67             final FlowRegistryKey key = FlowRegistryKeyFactory.create(version, flow);
  68
  69             // Now, we will update the registry, but we will also try to prevent duplicate entries
  70             if (!flowRegistry.containsKey(key)) {
  71                 LOG.trace("Found flow with table ID : {} and flow ID : {}", flow.getTableId(), flow.getId().getValue());
  72                 final FlowDescriptor descriptor = FlowDescriptorFactory.create(flow.getTableId(), flow.getId());
  73                 storeDescriptor(key, descriptor);
  74             }
  75         };
  76     }
  77
  78     @Override
  79     @GuardedBy("this")
  80     public synchronized ListenableFuture<List<Optional<FlowCapableNode>>> fill() {
  81         LOG.debug("Filling flow registry with flows for node: {}", instanceIdentifier.getKey().getId().getValue());
  82
  83         // Prepare path for read transaction
  84         // TODO: Read only Tables, and not entire FlowCapableNode (fix Yang model)
  85         final InstanceIdentifier<FlowCapableNode> path = instanceIdentifier.augmentation(FlowCapableNode.class);
  86
  87         // First, try to fill registry with flows from DS/Configuration
  88         final CheckedFuture<Optional<FlowCapableNode>, ReadFailedException> configFuture = fillFromDatastore(LogicalDatastoreType.CONFIGURATION, path);
  89
  90         // Now, try to fill registry with flows from DS/Operational
  91         // in case of cluster fail over, when clients are not using DS/Configuration
  92         // for adding flows, but only RPCs
  93         final CheckedFuture<Optional<FlowCapableNode>, ReadFailedException> operationalFuture = fillFromDatastore(LogicalDatastoreType.OPERATIONAL, path);
  94
  95         // And at last, chain and return futures created above.
  96         // Also, cache this future, so call to DeviceFlowRegistry.close() will be able
  97         // to cancel this future immediately if it will be still in progress
  98         final ListenableFuture<List<Optional<FlowCapableNode>>> lastFillFuture = Futures.allAsList(Arrays.asList(configFuture, operationalFuture));
  99         lastFillFutures.add(lastFillFuture);
 100         return lastFillFuture;
 101     }
 102
 103     @GuardedBy("this")
 104     private CheckedFuture<Optional<FlowCapableNode>, ReadFailedException> fillFromDatastore(final LogicalDatastoreType logicalDatastoreType, final InstanceIdentifier<FlowCapableNode> path) {
 105         // Create new read-only transaction
 106         final ReadOnlyTransaction transaction = dataBroker.newReadOnlyTransaction();
 107
 108         // Bail out early if transaction is null
 109         if (transaction == null) {
 110             return Futures.immediateFailedCheckedFuture(
 111                 new ReadFailedException("Read transaction is null"));
 112         }
 113
 114         // Prepare read operation from datastore for path
 115         final CheckedFuture<Optional<FlowCapableNode>, ReadFailedException> future =
 116             transaction.read(logicalDatastoreType, path);
 117
 118         // Bail out early if future is null
 119         if (future == null) {
 120             return Futures.immediateFailedCheckedFuture(
 121                 new ReadFailedException("Future from read transaction is null"));
 122         }
 123
 124         Futures.addCallback(future, new FutureCallback<Optional<FlowCapableNode>>() {
 125             @Override
 126             public void onSuccess(Optional<FlowCapableNode> result) {
 127                 result.asSet().stream()
 128                     .filter(Objects::nonNull)
 129                     .filter(flowCapableNode -> Objects.nonNull(flowCapableNode.getTable()))
 130                     .flatMap(flowCapableNode -> flowCapableNode.getTable().stream())
 131                     .filter(Objects::nonNull)
 132                     .filter(table -> Objects.nonNull(table.getFlow()))
 133                     .flatMap(table -> table.getFlow().stream())
 134                     .filter(Objects::nonNull)
 135                     .filter(flow -> Objects.nonNull(flow.getId()))
 136                     .forEach(flowConsumer);
 137
 138                 // After we are done with reading from datastore, close the transaction
 139                 transaction.close();
 140             }
 141
 142             @Override
 143             public void onFailure(Throwable t) {
 144                 // Even when read operation failed, close the transaction
 145                 transaction.close();
 146             }
 147         });
 148
 149         return future;
 150     }
 151
 152     @Override
 153     @GuardedBy("this")
 154     public synchronized FlowDescriptor retrieveDescriptor(final FlowRegistryKey flowRegistryKey) {
 155         LOG.trace("Retrieving flow descriptor for flow hash : {}", flowRegistryKey.hashCode());
 156         return flowRegistry.get(correctFlowRegistryKey(flowRegistry.keySet(), flowRegistryKey));
 157     }
 158
 159     @Override
 160     @GuardedBy("this")
 161     public synchronized void storeDescriptor(final FlowRegistryKey flowRegistryKey, final FlowDescriptor flowDescriptor) {
 162         final FlowRegistryKey correctFlowRegistryKey = correctFlowRegistryKey(flowRegistry.keySet(), flowRegistryKey);
 163
 164         try {
 165             if (hasMark(correctFlowRegistryKey)) {
 166                 // We are probably doing update of flow ID or table ID, so remove mark for removal of this flow
 167                 // and replace it with new value
 168                 marks.remove(correctFlowRegistryKey(marks, correctFlowRegistryKey));
 169                 flowRegistry.forcePut(correctFlowRegistryKey, flowDescriptor);
 170                 return;
 171             }
 172
 173             LOG.trace("Storing flowDescriptor with table ID : {} and flow ID : {} for flow hash : {}",
 174                 flowDescriptor.getTableKey().getId(), flowDescriptor.getFlowId().getValue(), correctFlowRegistryKey.hashCode());
 175
 176             flowRegistry.put(correctFlowRegistryKey, flowDescriptor);
 177         } catch (IllegalArgumentException ex) {
 178             if (hasMark(flowRegistry.inverse().get(flowDescriptor))) {
 179                 // We are probably doing update of flow, but without changing flow ID or table ID, so we need to replace
 180                 // old value with new value, but keep the old value marked for removal
 181                 flowRegistry.forcePut(correctFlowRegistryKey, flowDescriptor);
 182                 return;
 183             }
 184
 185             // We are trying to store new flow to flow registry, but we already have different flow with same flow ID
 186             // stored in registry, so we need to create alien ID for this new flow here.
 187             LOG.warn("Flow with flow ID {} already exists in table {}, generating alien flow ID", flowDescriptor.getFlowId().getValue(),
 188                 flowDescriptor.getTableKey().getId());
 189
 190             flowRegistry.put(
 191                 correctFlowRegistryKey,
 192                 FlowDescriptorFactory.create(
 193                     flowDescriptor.getTableKey().getId(),
 194                     createAlienFlowId(flowDescriptor.getTableKey().getId())));
 195         }
 196     }
 197
 198     @Override
 199     @GuardedBy("this")
 200     public synchronized void forEachEntry(final BiConsumer<FlowRegistryKey, FlowDescriptor> consumer) {
 201         flowRegistry.forEach(consumer);
 202     }
 203
 204     @Override
 205     @GuardedBy("this")
 206     public synchronized void store(final FlowRegistryKey flowRegistryKey) {
 207         if (Objects.isNull(retrieveDescriptor(flowRegistryKey))) {
 208             // We do not found flow in flow registry, that means it do not have any ID already assigned, so we need
 209             // to generate new alien flow ID here.
 210             LOG.debug("Flow descriptor for flow hash : {} not found, generating alien flow ID", flowRegistryKey.hashCode());
 211             final short tableId = flowRegistryKey.getTableId();
 212             final FlowId alienFlowId = createAlienFlowId(tableId);
 213             final FlowDescriptor flowDescriptor = FlowDescriptorFactory.create(tableId, alienFlowId);
 214
 215             // Finally we got flowDescriptor, so now we will store it to registry,
 216             // so next time we won't need to generate it again
 217             storeDescriptor(flowRegistryKey, flowDescriptor);
 218         }
 219     }
 220
 221     @Override
 222     @GuardedBy("this")
 223     public synchronized void addMark(final FlowRegistryKey flowRegistryKey) {
 224         LOG.trace("Removing flow descriptor for flow hash : {}", flowRegistryKey.hashCode());
 225         marks.add(flowRegistryKey);
 226     }
 227
 228     @Override
 229     @GuardedBy("this")
 230     public synchronized boolean hasMark(final FlowRegistryKey flowRegistryKey) {
 231         return Objects.nonNull(flowRegistryKey) && marks.contains(correctFlowRegistryKey(marks, flowRegistryKey));
 232
 233     }
 234
 235     @Override
 236     @GuardedBy("this")
 237     public synchronized void processMarks() {
 238         // Remove all flows that was marked for removal from flow registry and clear all marks.
 239         marks.forEach(flowRegistry::remove);
 240         marks.clear();
 241     }
 242
 243     @Override
 244     @GuardedBy("this")
 245     public synchronized void forEach(final Consumer<FlowRegistryKey> consumer) {
 246         flowRegistry.keySet().forEach(consumer);
 247     }
 248
 249     @Override
 250     @GuardedBy("this")
 251     public synchronized int size() {
 252         return flowRegistry.size();
 253     }
 254
 255     @Override
 256     @GuardedBy("this")
 257     public synchronized void close() {
 258         final Iterator<ListenableFuture<List<Optional<FlowCapableNode>>>> iterator = lastFillFutures.iterator();
 259
 260         // We need to force interrupt and clear all running futures that are trying to read flow IDs from datastore
 261         while (iterator.hasNext()) {
 262             final ListenableFuture<List<Optional<FlowCapableNode>>> next = iterator.next();
 263             boolean success = next.cancel(true);
 264             LOG.trace("Cancelling filling flow registry with flows job {} with result: {}", next, success);
 265             iterator.remove();
 266         }
 267
 268         flowRegistry.clear();
 269         marks.clear();
 270     }
 271
 272     @GuardedBy("this")
 273     private FlowRegistryKey correctFlowRegistryKey(final Collection<FlowRegistryKey> flowRegistryKeys, final FlowRegistryKey key) {
 274         if (Objects.isNull(key)) {
 275             return null;
 276         }
 277
 278         if (!flowRegistryKeys.contains(key)) {
 279             // If we failed to compare FlowRegistryKey by hashCode, try to retrieve correct FlowRegistryKey
 280             // from set of keys using custom comparator method for Match. This case can occur when we have different
 281             // augmentations on extensions, or switch returned things like IP address or port in different format that
 282             // we sent.
 283             if (LOG.isTraceEnabled()) {
 284                 LOG.trace("Failed to retrieve flow descriptor for flow hash : {}, trying with custom equals method", key.hashCode());
 285             }
 286
 287             for (final FlowRegistryKey flowRegistryKey : flowRegistryKeys) {
 288                 if (key.equals(flowRegistryKey)) {
 289                     return flowRegistryKey;
 290                 }
 291             }
 292         }
 293
 294         // If we failed to find key at all or key is already present in set of keys, just return original key
 295         return key;
 296     }
 297
 298     @VisibleForTesting
 299     static FlowId createAlienFlowId(final short tableId) {
 300         final String alienId = ALIEN_SYSTEM_FLOW_ID + tableId + '-' + UNACCOUNTED_FLOWS_COUNTER.incrementAndGet();
 301         LOG.debug("Created alien flow id {} for table id {}", alienId, tableId);
 302         return new FlowId(alienId);
 303     }
 304
 305     @VisibleForTesting
 306     Map<FlowRegistryKey, FlowDescriptor> getAllFlowDescriptors() {
 307         return flowRegistry;
 308     }
 309 }