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.collect.BiMap;
  12 import com.google.common.collect.HashBiMap;
  13 import com.google.common.collect.Maps;
  14 import com.google.common.util.concurrent.FluentFuture;
  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 com.google.common.util.concurrent.MoreExecutors;
  19 import java.util.ArrayList;
  20 import java.util.Arrays;
  21 import java.util.Collections;
  22 import java.util.Iterator;
  23 import java.util.List;
  24 import java.util.Map;
  25 import java.util.Objects;
  26 import java.util.Optional;
  27 import java.util.concurrent.atomic.AtomicInteger;
  28 import java.util.function.Consumer;
  29 import javax.annotation.Nonnull;
  30 import javax.annotation.concurrent.ThreadSafe;
  31 import org.opendaylight.mdsal.binding.api.DataBroker;
  32 import org.opendaylight.mdsal.binding.api.ReadTransaction;
  33 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
  34 import org.opendaylight.mdsal.common.api.ReadFailedException;
  35 import org.opendaylight.openflowplugin.api.openflow.registry.flow.DeviceFlowRegistry;
  36 import org.opendaylight.openflowplugin.api.openflow.registry.flow.FlowDescriptor;
  37 import org.opendaylight.openflowplugin.api.openflow.registry.flow.FlowRegistryKey;
  38 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
  39 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowId;
  40 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow;
  41 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
  42 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.NodeKey;
  43 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.extension.general.rev140714.GeneralAugMatchNodesNodeTableFlow;
  44 import org.opendaylight.yangtools.util.concurrent.FluentFutures;
  45 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  46 import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier;
  47 import org.slf4j.Logger;
  48 import org.slf4j.LoggerFactory;
  49
  50 @ThreadSafe
  51 public class DeviceFlowRegistryImpl implements DeviceFlowRegistry {
  52     private static final Logger LOG = LoggerFactory.getLogger(DeviceFlowRegistryImpl.class);
  53     private static final String ALIEN_SYSTEM_FLOW_ID = "#UF$TABLE*";
  54     private static final AtomicInteger UNACCOUNTED_FLOWS_COUNTER = new AtomicInteger(0);
  55
  56     private final BiMap<FlowRegistryKey, FlowDescriptor> flowRegistry = Maps.synchronizedBiMap(HashBiMap.create());
  57     private final DataBroker dataBroker;
  58     private final KeyedInstanceIdentifier<Node, NodeKey> instanceIdentifier;
  59     private final List<ListenableFuture<List<Optional<FlowCapableNode>>>> lastFillFutures = new ArrayList<>();
  60     private final Consumer<Flow> flowConsumer;
  61
  62     public DeviceFlowRegistryImpl(final short version,
  63                                   final DataBroker dataBroker,
  64                                   final KeyedInstanceIdentifier<Node, NodeKey> instanceIdentifier) {
  65         this.dataBroker = dataBroker;
  66         this.instanceIdentifier = instanceIdentifier;
  67
  68         // Specifies what to do with flow read from data store
  69         flowConsumer = flow -> {
  70             final FlowRegistryKey flowRegistryKey = FlowRegistryKeyFactory.create(version, flow);
  71
  72             if (getExistingKey(flowRegistryKey) == null) {
  73                 // Now, we will update the registry
  74                 storeDescriptor(flowRegistryKey, FlowDescriptorFactory.create(flow.getTableId(), flow.getId()));
  75             }
  76         };
  77     }
  78
  79     @Override
  80     public ListenableFuture<List<Optional<FlowCapableNode>>> fill() {
  81         if (LOG.isDebugEnabled()) {
  82             LOG.debug("Filling flow registry with flows for node: {}", instanceIdentifier.getKey().getId().getValue());
  83         }
  84
  85         // Prepare path for read transaction
  86         // TODO: Read only Tables, and not entire FlowCapableNode (fix Yang model)
  87         final InstanceIdentifier<FlowCapableNode> path = instanceIdentifier.augmentation(FlowCapableNode.class);
  88
  89         // First, try to fill registry with flows from DS/Configuration
  90         final FluentFuture<Optional<FlowCapableNode>> configFuture =
  91                 fillFromDatastore(LogicalDatastoreType.CONFIGURATION, path);
  92
  93         // Now, try to fill registry with flows from DS/Operational
  94         // in case of cluster fail over, when clients are not using DS/Configuration
  95         // for adding flows, but only RPCs
  96         final FluentFuture<Optional<FlowCapableNode>> operationalFuture =
  97                 fillFromDatastore(LogicalDatastoreType.OPERATIONAL, path);
  98
  99         // And at last, chain and return futures created above.
 100         // Also, cache this future, so call to DeviceFlowRegistry.close() will be able
 101         // to cancel this future immediately if it will be still in progress
 102         final ListenableFuture<List<Optional<FlowCapableNode>>> lastFillFuture =
 103                 Futures.allAsList(Arrays.asList(configFuture, operationalFuture));
 104         lastFillFutures.add(lastFillFuture);
 105         return lastFillFuture;
 106     }
 107
 108     private FluentFuture<Optional<FlowCapableNode>> fillFromDatastore(final LogicalDatastoreType logicalDatastoreType,
 109                               final InstanceIdentifier<FlowCapableNode> path) {
 110         // Create new read-only transaction
 111         final ReadTransaction transaction = dataBroker.newReadOnlyTransaction();
 112
 113         // Bail out early if transaction is null
 114         if (transaction == null) {
 115             return FluentFutures.immediateFailedFluentFuture(
 116                     new ReadFailedException("Read transaction is null"));
 117         }
 118
 119         // Prepare read operation from datastore for path
 120         final FluentFuture<Optional<FlowCapableNode>> future = transaction.read(logicalDatastoreType, path);
 121
 122         // Bail out early if future is null
 123         if (future == null) {
 124             return FluentFutures.immediateFailedFluentFuture(
 125                     new ReadFailedException("Future from read transaction is null"));
 126         }
 127
 128         future.addCallback(new FutureCallback<Optional<FlowCapableNode>>() {
 129             @Override
 130             public void onSuccess(Optional<FlowCapableNode> result) {
 131                 result.map(Collections::singleton).orElse(Collections.emptySet()).stream()
 132                         .filter(Objects::nonNull)
 133                         .filter(flowCapableNode -> Objects.nonNull(flowCapableNode.getTable()))
 134                         .flatMap(flowCapableNode -> flowCapableNode.getTable().stream())
 135                         .filter(Objects::nonNull)
 136                         .filter(table -> Objects.nonNull(table.getFlow()))
 137                         .flatMap(table -> table.getFlow().stream())
 138                         .filter(Objects::nonNull)
 139                         .filter(flow -> Objects.nonNull(flow.getId()))
 140                         .forEach(flowConsumer);
 141
 142                 // After we are done with reading from datastore, close the transaction
 143                 transaction.close();
 144             }
 145
 146             @Override
 147             public void onFailure(Throwable throwable) {
 148                 // Even when read operation failed, close the transaction
 149                 transaction.close();
 150             }
 151         }, MoreExecutors.directExecutor());
 152
 153         return future;
 154     }
 155
 156     @Override
 157     public FlowDescriptor retrieveDescriptor(@Nonnull final FlowRegistryKey flowRegistryKey) {
 158         if (LOG.isTraceEnabled()) {
 159             LOG.trace("Retrieving flow descriptor for flow registry : {}", flowRegistryKey.toString());
 160         }
 161
 162         FlowRegistryKey existingFlowRegistryKey = getExistingKey(flowRegistryKey);
 163         if (existingFlowRegistryKey != null) {
 164             return flowRegistry.get(existingFlowRegistryKey);
 165         }
 166         return null;
 167     }
 168
 169     @Override
 170     public void storeDescriptor(@Nonnull final FlowRegistryKey flowRegistryKey,
 171                                 @Nonnull final FlowDescriptor flowDescriptor) {
 172         try {
 173             if (LOG.isTraceEnabled()) {
 174                 LOG.trace("Storing flowDescriptor with table ID : {} and flow ID : {} for flow hash : {}",
 175                         flowDescriptor.getTableKey().getId(),
 176                         flowDescriptor.getFlowId().getValue(),
 177                         flowRegistryKey.toString());
 178             }
 179
 180             addToFlowRegistry(flowRegistryKey, flowDescriptor);
 181         } catch (IllegalArgumentException ex) {
 182             if (LOG.isWarnEnabled()) {
 183                 LOG.warn("Flow with flow ID {} already exists in table {}, generating alien flow ID",
 184                         flowDescriptor.getFlowId().getValue(),
 185                         flowDescriptor.getTableKey().getId());
 186             }
 187
 188             // We are trying to store new flow to flow registry, but we already have different flow with same flow ID
 189             // stored in registry, so we need to create alien ID for this new flow here.
 190             addToFlowRegistry(
 191                     flowRegistryKey,
 192                     FlowDescriptorFactory.create(
 193                             flowDescriptor.getTableKey().getId(),
 194                             createAlienFlowId(flowDescriptor.getTableKey().getId())));
 195         }
 196     }
 197
 198     @Override
 199     public void store(final FlowRegistryKey flowRegistryKey) {
 200         if (retrieveDescriptor(flowRegistryKey) == null) {
 201             LOG.debug("Flow descriptor for flow hash : {} not found, generating alien flow ID", flowRegistryKey);
 202
 203             // We do not found flow in flow registry, that means it do not have any ID already assigned, so we need
 204             // to generate new alien flow ID here.
 205             storeDescriptor(
 206                     flowRegistryKey,
 207                     FlowDescriptorFactory.create(
 208                             flowRegistryKey.getTableId(),
 209                             createAlienFlowId(flowRegistryKey.getTableId())));
 210         }
 211     }
 212
 213     @Override
 214     public void addMark(final FlowRegistryKey flowRegistryKey) {
 215         if (LOG.isTraceEnabled()) {
 216             LOG.trace("Removing flow descriptor for flow hash : {}", flowRegistryKey.toString());
 217         }
 218
 219         removeFromFlowRegistry(flowRegistryKey);
 220     }
 221
 222     @Override
 223     public void processMarks() {
 224         // Do nothing
 225     }
 226
 227     @Override
 228     public void forEach(final Consumer<FlowRegistryKey> consumer) {
 229         synchronized (flowRegistry) {
 230             flowRegistry.keySet().forEach(consumer);
 231         }
 232     }
 233
 234     @Override
 235     public int size() {
 236         return flowRegistry.size();
 237     }
 238
 239     @Override
 240     public void close() {
 241         final Iterator<ListenableFuture<List<Optional<FlowCapableNode>>>> iterator = lastFillFutures.iterator();
 242
 243         // We need to force interrupt and clear all running futures that are trying to read flow IDs from data store
 244         while (iterator.hasNext()) {
 245             final ListenableFuture<List<Optional<FlowCapableNode>>> next = iterator.next();
 246             boolean success = next.cancel(true);
 247             LOG.trace("Cancelling filling flow registry with flows job {} with result: {}", next, success);
 248             iterator.remove();
 249         }
 250
 251         flowRegistry.clear();
 252     }
 253
 254     @VisibleForTesting
 255     static FlowId createAlienFlowId(final short tableId) {
 256         final String alienId = ALIEN_SYSTEM_FLOW_ID + tableId + '-' + UNACCOUNTED_FLOWS_COUNTER.incrementAndGet();
 257         LOG.debug("Created alien flow id {} for table id {}", alienId, tableId);
 258         return new FlowId(alienId);
 259     }
 260
 261     //Hashcode generation of the extension augmentation can differ for the same object received from the datastore and
 262     // the one received after deserialization of switch message. OpenFlowplugin extensions are list, and the order in
 263     // which it can receive the extensions back from switch can differ and that lead to a different hashcode. In that
 264     // scenario, hashcode won't match and flowRegistry return the  related key. To overcome this issue, these methods
 265     // make sure that key is stored only if it doesn't equals to any existing key.
 266     private void addToFlowRegistry(final FlowRegistryKey flowRegistryKey, final FlowDescriptor flowDescriptor) {
 267         FlowRegistryKey existingFlowRegistryKey = getExistingKey(flowRegistryKey);
 268         if (existingFlowRegistryKey == null) {
 269             flowRegistry.put(flowRegistryKey, flowDescriptor);
 270         } else {
 271             flowRegistry.put(existingFlowRegistryKey, flowDescriptor);
 272         }
 273     }
 274
 275     private void removeFromFlowRegistry(final FlowRegistryKey flowRegistryKey) {
 276         FlowRegistryKey existingFlowRegistryKey = getExistingKey(flowRegistryKey);
 277         if (existingFlowRegistryKey != null) {
 278             flowRegistry.remove(existingFlowRegistryKey);
 279         } else {
 280             flowRegistry.remove(flowRegistryKey);
 281         }
 282     }
 283
 284     private FlowRegistryKey getExistingKey(final FlowRegistryKey flowRegistryKey) {
 285         if (flowRegistryKey.getMatch().augmentation(GeneralAugMatchNodesNodeTableFlow.class) == null) {
 286             if (flowRegistry.containsKey(flowRegistryKey)) {
 287                 return flowRegistryKey;
 288             }
 289         } else {
 290             synchronized (flowRegistry) {
 291                 for (Map.Entry<FlowRegistryKey, FlowDescriptor> keyValueSet : flowRegistry.entrySet()) {
 292                     if (keyValueSet.getKey().equals(flowRegistryKey)) {
 293                         return keyValueSet.getKey();
 294                     }
 295                 }
 296             }
 297         }
 298         return null;
 299     }
 300
 301     @VisibleForTesting
 302     Map<FlowRegistryKey, FlowDescriptor> getAllFlowDescriptors() {
 303         return flowRegistry;
 304     }
 305 }