2 * Copyright (c) 2015 Cisco Systems, Inc. and others. All rights reserved.
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
8 package org.opendaylight.openflowplugin.impl.registry.flow;
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;
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;
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);
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;
60 public DeviceFlowRegistryImpl(final short version, final DataBroker dataBroker, final KeyedInstanceIdentifier<Node, NodeKey> instanceIdentifier) {
61 this.dataBroker = dataBroker;
62 this.instanceIdentifier = instanceIdentifier;
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);
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);
80 public synchronized ListenableFuture<List<Optional<FlowCapableNode>>> fill() {
81 LOG.debug("Filling flow registry with flows for node: {}", instanceIdentifier.getKey().getId().getValue());
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);
87 // First, try to fill registry with flows from DS/Configuration
88 final CheckedFuture<Optional<FlowCapableNode>, ReadFailedException> configFuture = fillFromDatastore(LogicalDatastoreType.CONFIGURATION, path);
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);
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;
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();
108 // Bail out early if transaction is null
109 if (transaction == null) {
110 return Futures.immediateFailedCheckedFuture(
111 new ReadFailedException("Read transaction is null"));
114 // Prepare read operation from datastore for path
115 final CheckedFuture<Optional<FlowCapableNode>, ReadFailedException> future =
116 transaction.read(logicalDatastoreType, path);
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"));
124 Futures.addCallback(future, new FutureCallback<Optional<FlowCapableNode>>() {
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);
138 // After we are done with reading from datastore, close the transaction
143 public void onFailure(Throwable t) {
144 // Even when read operation failed, close the transaction
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));
161 public synchronized void storeDescriptor(final FlowRegistryKey flowRegistryKey, final FlowDescriptor flowDescriptor) {
162 final FlowRegistryKey correctFlowRegistryKey = correctFlowRegistryKey(flowRegistry.keySet(), flowRegistryKey);
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);
173 LOG.trace("Storing flowDescriptor with table ID : {} and flow ID : {} for flow hash : {}",
174 flowDescriptor.getTableKey().getId(), flowDescriptor.getFlowId().getValue(), correctFlowRegistryKey.hashCode());
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);
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());
191 correctFlowRegistryKey,
192 FlowDescriptorFactory.create(
193 flowDescriptor.getTableKey().getId(),
194 createAlienFlowId(flowDescriptor.getTableKey().getId())));
200 public synchronized void forEachEntry(final BiConsumer<FlowRegistryKey, FlowDescriptor> consumer) {
201 flowRegistry.forEach(consumer);
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);
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);
223 public synchronized void addMark(final FlowRegistryKey flowRegistryKey) {
224 LOG.trace("Removing flow descriptor for flow hash : {}", flowRegistryKey.hashCode());
225 marks.add(flowRegistryKey);
230 public synchronized boolean hasMark(final FlowRegistryKey flowRegistryKey) {
231 return Objects.nonNull(flowRegistryKey) && marks.contains(correctFlowRegistryKey(marks, flowRegistryKey));
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);
245 public synchronized void forEach(final Consumer<FlowRegistryKey> consumer) {
246 flowRegistry.keySet().forEach(consumer);
251 public synchronized int size() {
252 return flowRegistry.size();
257 public synchronized void close() {
258 final Iterator<ListenableFuture<List<Optional<FlowCapableNode>>>> iterator = lastFillFutures.iterator();
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);
268 flowRegistry.clear();
273 private FlowRegistryKey correctFlowRegistryKey(final Collection<FlowRegistryKey> flowRegistryKeys, final FlowRegistryKey key) {
274 if (Objects.isNull(key)) {
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
283 if (LOG.isTraceEnabled()) {
284 LOG.trace("Failed to retrieve flow descriptor for flow hash : {}, trying with custom equals method", key.hashCode());
287 for (final FlowRegistryKey flowRegistryKey : flowRegistryKeys) {
288 if (key.equals(flowRegistryKey)) {
289 return flowRegistryKey;
294 // If we failed to find key at all or key is already present in set of keys, just return original key
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);
306 Map<FlowRegistryKey, FlowDescriptor> getAllFlowDescriptors() {