2 * Copyright (c) 2016 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
9 package org.opendaylight.openflowplugin.applications.frsync.util;
11 import com.google.common.base.Function;
12 import com.google.common.base.MoreObjects;
13 import com.google.common.collect.ImmutableList;
14 import com.google.common.collect.Iterables;
15 import com.google.common.util.concurrent.AsyncFunction;
16 import com.google.common.util.concurrent.JdkFutureAdapters;
17 import java.util.ArrayList;
18 import java.util.Collection;
19 import java.util.Collections;
20 import java.util.HashMap;
21 import java.util.Iterator;
22 import java.util.List;
24 import java.util.Objects;
26 import org.opendaylight.yang.gen.v1.urn.opendaylight.action.types.rev131112.action.action.GroupActionCase;
27 import org.opendaylight.yang.gen.v1.urn.opendaylight.action.types.rev131112.action.list.Action;
28 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
29 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.meters.Meter;
30 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.Table;
31 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.TableKey;
32 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow;
33 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.transaction.rev150304.FlowCapableTransactionService;
34 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.transaction.rev150304.SendBarrierInput;
35 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.transaction.rev150304.SendBarrierInputBuilder;
36 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.group.buckets.Bucket;
37 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.groups.Group;
38 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
39 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRef;
40 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
41 import org.opendaylight.yang.gen.v1.urn.opendaylight.meter.types.rev130918.MeterId;
42 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
43 import org.opendaylight.yangtools.yang.common.RpcError;
44 import org.opendaylight.yangtools.yang.common.RpcResult;
45 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
46 import org.slf4j.Logger;
47 import org.slf4j.LoggerFactory;
50 * Util methods for group reconcil task (future chaining, transforms).
52 public final class ReconcileUtil {
54 private static final Logger LOG = LoggerFactory.getLogger(ReconcileUtil.class);
56 private ReconcileUtil() {
57 throw new IllegalStateException("This class should not be instantiated.");
61 * @param previousItemAction description for case when the triggering future contains failure
62 * @param <D> type of rpc output (gathered in list)
63 * @return single rpc result of type Void honoring all partial rpc results
65 public static <D> Function<List<RpcResult<D>>, RpcResult<Void>> createRpcResultCondenser(final String previousItemAction) {
67 final RpcResultBuilder<Void> resultSink;
69 List<RpcError> errors = new ArrayList<>();
70 for (RpcResult<D> rpcResult : input) {
71 if (!rpcResult.isSuccessful()) {
72 errors.addAll(rpcResult.getErrors());
75 if (errors.isEmpty()) {
76 resultSink = RpcResultBuilder.success();
78 resultSink = RpcResultBuilder.<Void>failed().withRpcErrors(errors);
81 resultSink = RpcResultBuilder.<Void>failed()
82 .withError(RpcError.ErrorType.APPLICATION, "previous " + previousItemAction + " failed");
84 return resultSink.build();
89 * @param actionDescription description for case when the triggering future contains failure
90 * @param <D> type of rpc output (gathered in list)
91 * @return single rpc result of type Void honoring all partial rpc results
93 public static <D> Function<RpcResult<D>, RpcResult<Void>> createRpcResultToVoidFunction(final String actionDescription) {
95 final RpcResultBuilder<Void> resultSink;
97 List<RpcError> errors = new ArrayList<>();
98 if (!input.isSuccessful()) {
99 errors.addAll(input.getErrors());
100 resultSink = RpcResultBuilder.<Void>failed().withRpcErrors(errors);
102 resultSink = RpcResultBuilder.success();
105 resultSink = RpcResultBuilder.<Void>failed()
106 .withError(RpcError.ErrorType.APPLICATION, "action of " + actionDescription + " failed");
108 return resultSink.build();
113 * @param nodeIdent flow capable node path - target device for routed rpc
114 * @param flowCapableTransactionService barrier rpc service
115 * @return async barrier result
117 public static AsyncFunction<RpcResult<Void>, RpcResult<Void>> chainBarrierFlush(
118 final InstanceIdentifier<Node> nodeIdent,
119 final FlowCapableTransactionService flowCapableTransactionService) {
121 final SendBarrierInput barrierInput = new SendBarrierInputBuilder()
122 .setNode(new NodeRef(nodeIdent))
124 return JdkFutureAdapters.listenInPoolThread(flowCapableTransactionService.sendBarrier(barrierInput));
129 * @param nodeId target node
130 * @param installedGroupsArg groups resent on device
131 * @param pendingGroups groups configured for device
132 * @return list of safe synchronization steps with updates
134 public static List<ItemSyncBox<Group>> resolveAndDivideGroupDiffs(final NodeId nodeId,
135 final Map<Long, Group> installedGroupsArg,
136 final Collection<Group> pendingGroups) {
137 return resolveAndDivideGroupDiffs(nodeId, installedGroupsArg, pendingGroups, true);
141 * @param nodeId target node
142 * @param installedGroupsArg groups resent on device
143 * @param pendingGroups groups configured for device
144 * @param gatherUpdates check content of pending item if present on device (and create update task eventually)
145 * @return list of safe synchronization steps
147 public static List<ItemSyncBox<Group>> resolveAndDivideGroupDiffs(final NodeId nodeId,
148 final Map<Long, Group> installedGroupsArg,
149 final Collection<Group> pendingGroups,
150 final boolean gatherUpdates) {
151 final Map<Long, Group> installedGroups = new HashMap<>(installedGroupsArg);
152 final List<ItemSyncBox<Group>> plan = new ArrayList<>();
154 while (!Iterables.isEmpty(pendingGroups)) {
155 final ItemSyncBox<Group> stepPlan = new ItemSyncBox<>();
156 final Iterator<Group> iterator = pendingGroups.iterator();
157 final Map<Long, Group> installIncrement = new HashMap<>();
159 while (iterator.hasNext()) {
160 final Group group = iterator.next();
162 final Group existingGroup = installedGroups.get(group.getGroupId().getValue());
163 if (existingGroup != null) {
164 if (!gatherUpdates) {
167 // check buckets and eventually update
168 if (group.equals(existingGroup)) {
171 if (checkGroupPrecondition(installedGroups.keySet(), group)) {
173 LOG.trace("Group {} on device {} differs - planned for update", group.getGroupId(), nodeId);
174 stepPlan.getItemsToUpdate().add(new ItemSyncBox.ItemUpdateTuple<>(existingGroup, group));
178 } else if (checkGroupPrecondition(installedGroups.keySet(), group)) {
180 installIncrement.put(group.getGroupId().getValue(), group);
181 stepPlan.getItemsToPush().add(group);
185 if (!stepPlan.isEmpty()) {
186 // atomic update of installed flows in order to keep plan portions clean of local group dependencies
187 installedGroups.putAll(installIncrement);
189 } else if (!pendingGroups.isEmpty()) {
190 LOG.warn("Failed to resolve and divide groups into preconditions-match based ordered plan: {}, " +
191 "resolving stuck at level {}", nodeId.getValue(), plan.size());
192 throw new IllegalStateException("Failed to resolve and divide groups when matching preconditions");
199 public static boolean checkGroupPrecondition(final Set<Long> installedGroupIds, final Group pendingGroup) {
200 boolean okToInstall = true;
201 // check each bucket in the pending group
202 for (Bucket bucket : pendingGroup.getBuckets().getBucket()) {
203 for (Action action : bucket.getAction()) {
204 // if the output action is a group
205 if (GroupActionCase.class.equals(action.getAction().getImplementedInterface())) {
206 Long groupId = ((GroupActionCase) (action.getAction())).getGroupAction().getGroupId();
207 // see if that output group is installed
208 if (!installedGroupIds.contains(groupId)) {
209 // if not installed, we have missing dependencies and cannot install this pending group
222 public static <E> int countTotalPushed(final Iterable<ItemSyncBox<E>> groupsAddPlan) {
224 for (ItemSyncBox<E> groupItemSyncBox : groupsAddPlan) {
225 count += groupItemSyncBox.getItemsToPush().size();
230 public static <E> int countTotalUpdated(final Iterable<ItemSyncBox<E>> groupsAddPlan) {
232 for (ItemSyncBox<E> groupItemSyncBox : groupsAddPlan) {
233 count += groupItemSyncBox.getItemsToUpdate().size();
239 * @param nodeId target node
240 * @param meterOperationalMap meters present on device
241 * @param metersConfigured meters configured for device
242 * @param gatherUpdates check content of pending item if present on device (and create update task eventually)
243 * @return synchronization box
245 public static ItemSyncBox<Meter> resolveMeterDiffs(final NodeId nodeId,
246 final Map<MeterId, Meter> meterOperationalMap,
247 final List<Meter> metersConfigured,
248 final boolean gatherUpdates) {
249 LOG.trace("resolving meters for {}", nodeId.getValue());
250 final ItemSyncBox<Meter> syncBox = new ItemSyncBox<>();
251 for (Meter meter : metersConfigured) {
252 final Meter existingMeter = meterOperationalMap.get(meter.getMeterId());
253 if (existingMeter == null) {
254 syncBox.getItemsToPush().add(meter);
256 // compare content and eventually update
257 if (gatherUpdates && !meter.equals(existingMeter)) {
258 syncBox.getItemsToUpdate().add(new ItemSyncBox.ItemUpdateTuple<>(existingMeter, meter));
266 * @param flowsConfigured flows resent on device
267 * @param flowOperationalMap flows configured for device
268 * @param gatherUpdates check content of pending item if present on device (and create update task eventually)
269 * @return list of safe synchronization steps
271 public static ItemSyncBox<Flow> resolveFlowDiffsInTable(final List<Flow> flowsConfigured,
272 final Map<SwitchFlowId, Flow> flowOperationalMap,
273 final boolean gatherUpdates) {
274 final ItemSyncBox<Flow> flowsSyncBox = new ItemSyncBox<>();
275 // loop configured flows and check if already present on device
276 for (final Flow flow : flowsConfigured) {
277 final Flow existingFlow = FlowCapableNodeLookups.flowMapLookupExisting(flow, flowOperationalMap);
279 if (existingFlow == null) {
280 flowsSyncBox.getItemsToPush().add(flow);
282 // check instructions and eventually update
283 if (gatherUpdates && !Objects.equals(flow.getInstructions(), existingFlow.getInstructions())) {
284 flowsSyncBox.getItemsToUpdate().add(new ItemSyncBox.ItemUpdateTuple<>(existingFlow, flow));
292 * @param nodeId target node
293 * @param tableOperationalMap flow-tables resent on device
294 * @param tablesConfigured flow-tables configured for device
295 * @param gatherUpdates check content of pending item if present on device (and create update task eventually)
296 * @return map : key={@link TableKey}, value={@link ItemSyncBox} of safe synchronization steps
298 public static Map<TableKey, ItemSyncBox<Flow>> resolveFlowDiffsInAllTables(final NodeId nodeId,
299 final Map<Short, Table> tableOperationalMap,
300 final List<Table> tablesConfigured,
301 final boolean gatherUpdates) {
302 LOG.trace("resolving flows in tables for {}", nodeId.getValue());
303 final Map<TableKey, ItemSyncBox<Flow>> tableFlowSyncBoxes = new HashMap<>();
304 for (final Table tableConfigured : tablesConfigured) {
305 final List<Flow> flowsConfigured = tableConfigured.getFlow();
306 if (flowsConfigured == null || flowsConfigured.isEmpty()) {
310 // lookup table (on device)
311 final Table tableOperational = tableOperationalMap.get(tableConfigured.getId());
312 // wrap existing (on device) flows in current table into map
313 final Map<SwitchFlowId, Flow> flowOperationalMap = FlowCapableNodeLookups.wrapFlowsToMap(
314 tableOperational != null
315 ? tableOperational.getFlow()
319 final ItemSyncBox<Flow> flowsSyncBox = resolveFlowDiffsInTable(
320 flowsConfigured, flowOperationalMap, gatherUpdates);
321 if (!flowsSyncBox.isEmpty()) {
322 tableFlowSyncBoxes.put(tableConfigured.getKey(), flowsSyncBox);
325 return tableFlowSyncBoxes;
328 public static List<Group> safeGroups(FlowCapableNode node) {
330 return Collections.emptyList();
333 return MoreObjects.firstNonNull(node.getGroup(), ImmutableList.<Group>of());
336 public static List<Table> safeTables(FlowCapableNode node) {
338 return Collections.emptyList();
341 return MoreObjects.firstNonNull(node.getTable(), ImmutableList.<Table>of());
344 public static List<Meter> safeMeters(FlowCapableNode node) {
346 return Collections.emptyList();
349 return MoreObjects.firstNonNull(node.getMeter(), ImmutableList.<Meter>of());