Switch to MD-SAL APIs
[openflowplugin.git] / 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(@Nonnull 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 (Objects.isNull(retrieveDescriptor(flowRegistryKey))) {
201             if (LOG.isDebugEnabled()) {
202                 LOG.debug("Flow descriptor for flow hash : {} not found, generating alien flow ID",
203                         flowRegistryKey.toString());
204             }
205
206             // We do not found flow in flow registry, that means it do not have any ID already assigned, so we need
207             // to generate new alien flow ID here.
208             storeDescriptor(
209                     flowRegistryKey,
210                     FlowDescriptorFactory.create(
211                             flowRegistryKey.getTableId(),
212                             createAlienFlowId(flowRegistryKey.getTableId())));
213         }
214     }
215
216     @Override
217     public void addMark(final FlowRegistryKey flowRegistryKey) {
218         if (LOG.isTraceEnabled()) {
219             LOG.trace("Removing flow descriptor for flow hash : {}", flowRegistryKey.toString());
220         }
221
222         removeFromFlowRegistry(flowRegistryKey);
223     }
224
225     @Override
226     public void processMarks() {
227         // Do nothing
228     }
229
230     @Override
231     public void forEach(final Consumer<FlowRegistryKey> consumer) {
232         synchronized (flowRegistry) {
233             flowRegistry.keySet().forEach(consumer);
234         }
235     }
236
237     @Override
238     public int size() {
239         return flowRegistry.size();
240     }
241
242     @Override
243     public void close() {
244         final Iterator<ListenableFuture<List<Optional<FlowCapableNode>>>> iterator = lastFillFutures.iterator();
245
246         // We need to force interrupt and clear all running futures that are trying to read flow IDs from data store
247         while (iterator.hasNext()) {
248             final ListenableFuture<List<Optional<FlowCapableNode>>> next = iterator.next();
249             boolean success = next.cancel(true);
250             LOG.trace("Cancelling filling flow registry with flows job {} with result: {}", next, success);
251             iterator.remove();
252         }
253
254         flowRegistry.clear();
255     }
256
257     @VisibleForTesting
258     static FlowId createAlienFlowId(final short tableId) {
259         final String alienId = ALIEN_SYSTEM_FLOW_ID + tableId + '-' + UNACCOUNTED_FLOWS_COUNTER.incrementAndGet();
260         LOG.debug("Created alien flow id {} for table id {}", alienId, tableId);
261         return new FlowId(alienId);
262     }
263
264     //Hashcode generation of the extension augmentation can differ for the same object received from the datastore and
265     // the one received after deserialization of switch message. OpenFlowplugin extensions are list, and the order in
266     // which it can receive the extensions back from switch can differ and that lead to a different hashcode. In that
267     // scenario, hashcode won't match and flowRegistry return the  related key. To overcome this issue, these methods
268     // make sure that key is stored only if it doesn't equals to any existing key.
269     private void addToFlowRegistry(final FlowRegistryKey flowRegistryKey, final FlowDescriptor flowDescriptor) {
270         FlowRegistryKey existingFlowRegistryKey = getExistingKey(flowRegistryKey);
271         if (existingFlowRegistryKey == null) {
272             flowRegistry.put(flowRegistryKey, flowDescriptor);
273         } else {
274             flowRegistry.put(existingFlowRegistryKey, flowDescriptor);
275         }
276     }
277
278     private void removeFromFlowRegistry(final FlowRegistryKey flowRegistryKey) {
279         FlowRegistryKey existingFlowRegistryKey = getExistingKey(flowRegistryKey);
280         if (existingFlowRegistryKey != null) {
281             flowRegistry.remove(existingFlowRegistryKey);
282         } else {
283             flowRegistry.remove(flowRegistryKey);
284         }
285     }
286
287     private FlowRegistryKey getExistingKey(final FlowRegistryKey flowRegistryKey) {
288         if (flowRegistryKey.getMatch().augmentation(GeneralAugMatchNodesNodeTableFlow.class) == null) {
289             if (flowRegistry.containsKey(flowRegistryKey)) {
290                 return flowRegistryKey;
291             }
292         } else {
293             synchronized (flowRegistry) {
294                 for (Map.Entry<FlowRegistryKey, FlowDescriptor> keyValueSet : flowRegistry.entrySet()) {
295                     if (keyValueSet.getKey().equals(flowRegistryKey)) {
296                         return keyValueSet.getKey();
297                     }
298                 }
299             }
300         }
301         return null;
302     }
303
304     @VisibleForTesting
305     Map<FlowRegistryKey, FlowDescriptor> getAllFlowDescriptors() {
306         return flowRegistry;
307     }
308 }