applications/forwardingrules-sync/src/main/java/org/opendaylight/openflowplugin/applications/frsync/util/ReconcileUtil.java

   1 /*
   2  * Copyright (c) 2016 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.applications.frsync.util;
   9
  10 import com.google.common.base.Function;
  11 import com.google.common.collect.Iterables;
  12 import com.google.common.util.concurrent.AsyncFunction;
  13 import com.google.common.util.concurrent.Futures;
  14 import com.google.common.util.concurrent.ListenableFuture;
  15 import com.google.common.util.concurrent.MoreExecutors;
  16 import java.util.ArrayList;
  17 import java.util.Collection;
  18 import java.util.Collections;
  19 import java.util.HashMap;
  20 import java.util.Iterator;
  21 import java.util.List;
  22 import java.util.Map;
  23 import java.util.Objects;
  24 import java.util.Set;
  25 import org.opendaylight.yang.gen.v1.urn.opendaylight.action.types.rev131112.action.action.GroupActionCase;
  26 import org.opendaylight.yang.gen.v1.urn.opendaylight.action.types.rev131112.action.list.Action;
  27 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
  28 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.meters.Meter;
  29 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.Table;
  30 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.TableKey;
  31 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow;
  32 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.transaction.rev150304.FlowCapableTransactionService;
  33 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.transaction.rev150304.SendBarrierInput;
  34 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.transaction.rev150304.SendBarrierInputBuilder;
  35 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.transaction.rev150304.SendBarrierOutput;
  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.opendaylight.yangtools.yang.common.Uint32;
  47 import org.opendaylight.yangtools.yang.common.Uint8;
  48 import org.slf4j.Logger;
  49 import org.slf4j.LoggerFactory;
  50
  51 /**
  52  * Util methods for group reconcil task (future chaining, transforms).
  53  */
  54 public final class ReconcileUtil {
  55     private static final Logger LOG = LoggerFactory.getLogger(ReconcileUtil.class);
  56
  57     private ReconcileUtil() {
  58         // Hidden on purpose
  59     }
  60
  61     /**
  62      * Creates a single rpc result of type Void honoring all partial rpc results.
  63      *
  64      * @param previousItemAction description for case when the triggering future contains failure
  65      * @param <D>                type of rpc output (gathered in list)
  66      * @return single rpc result of type Void honoring all partial rpc results
  67      */
  68     public static <D> Function<List<RpcResult<D>>, RpcResult<Void>> createRpcResultCondenser(
  69             final String previousItemAction) {
  70         return input -> {
  71             final RpcResultBuilder<Void> resultSink;
  72             if (input != null) {
  73                 List<RpcError> errors = new ArrayList<>();
  74                 for (RpcResult<D> rpcResult : input) {
  75                     if (!rpcResult.isSuccessful()) {
  76                         errors.addAll(rpcResult.getErrors());
  77                     }
  78                 }
  79                 if (errors.isEmpty()) {
  80                     resultSink = RpcResultBuilder.success();
  81                 } else {
  82                     resultSink = RpcResultBuilder.<Void>failed().withRpcErrors(errors);
  83                 }
  84             } else {
  85                 resultSink = RpcResultBuilder.<Void>failed()
  86                         .withError(RpcError.ErrorType.APPLICATION, "previous " + previousItemAction + " failed");
  87             }
  88             return resultSink.build();
  89         };
  90     }
  91
  92     /**
  93      * Creates a single rpc result of type Void honoring all partial rpc results.
  94      *
  95      * @param actionDescription description for case when the triggering future contains failure
  96      * @param <D>               type of rpc output (gathered in list)
  97      * @return single rpc result of type Void honoring all partial rpc results
  98      */
  99     public static <D> Function<RpcResult<D>, RpcResult<Void>> createRpcResultToVoidFunction(
 100             final String actionDescription) {
 101         return input -> {
 102             final RpcResultBuilder<Void> resultSink;
 103             if (input != null) {
 104                 List<RpcError> errors = new ArrayList<>();
 105                 if (!input.isSuccessful()) {
 106                     errors.addAll(input.getErrors());
 107                     resultSink = RpcResultBuilder.<Void>failed().withRpcErrors(errors);
 108                 } else {
 109                     resultSink = RpcResultBuilder.success();
 110                 }
 111             } else {
 112                 resultSink = RpcResultBuilder.<Void>failed()
 113                         .withError(RpcError.ErrorType.APPLICATION, "action of " + actionDescription + " failed");
 114             }
 115             return resultSink.build();
 116         };
 117     }
 118
 119     /**
 120      * Flushes a chain barrier.
 121      *
 122      * @param nodeIdent flow capable node path - target device for routed rpc
 123      * @param flowCapableTransactionService barrier rpc service
 124      * @return async barrier result
 125      */
 126     public static AsyncFunction<RpcResult<Void>, RpcResult<Void>> chainBarrierFlush(
 127             final InstanceIdentifier<Node> nodeIdent,
 128             final FlowCapableTransactionService flowCapableTransactionService) {
 129         return input -> {
 130             final SendBarrierInput barrierInput = new SendBarrierInputBuilder()
 131                     .setNode(new NodeRef(nodeIdent))
 132                     .build();
 133             ListenableFuture<RpcResult<SendBarrierOutput>> result
 134                     = flowCapableTransactionService.sendBarrier(barrierInput);
 135
 136             return Futures.transformAsync(result, input1 -> {
 137                 if (input1.isSuccessful()) {
 138                     return Futures.immediateFuture(RpcResultBuilder.<Void>success().build());
 139                 } else {
 140                     return Futures.immediateFailedFuture(null);
 141                 }
 142             }, MoreExecutors.directExecutor());
 143         };
 144     }
 145
 146     /**
 147      * Returns a list of safe synchronization steps with updates.
 148      *
 149      * @param nodeId target node
 150      * @param installedGroupsArg groups resent on device
 151      * @param pendingGroups      groups configured for device
 152      * @return list of safe synchronization steps with updates
 153      */
 154     public static List<ItemSyncBox<Group>> resolveAndDivideGroupDiffs(final NodeId nodeId,
 155                                                                       final Map<Uint32, Group> installedGroupsArg,
 156                                                                       final Collection<Group> pendingGroups) {
 157         return resolveAndDivideGroupDiffs(nodeId, installedGroupsArg, pendingGroups, true);
 158     }
 159
 160     /**
 161      * Returns a list of safe synchronization steps.
 162      *
 163      * @param nodeId             target node
 164      * @param installedGroupsArg groups resent on device
 165      * @param pendingGroups      groups configured for device
 166      * @param gatherUpdates      check content of pending item if present on device (and create update task eventually)
 167      * @return list of safe synchronization steps
 168      */
 169     public static List<ItemSyncBox<Group>> resolveAndDivideGroupDiffs(final NodeId nodeId,
 170                                                                       final Map<Uint32, Group> installedGroupsArg,
 171                                                                       final Collection<Group> pendingGroups,
 172                                                                       final boolean gatherUpdates) {
 173         final Map<Uint32, Group> installedGroups = new HashMap<>(installedGroupsArg);
 174         final List<ItemSyncBox<Group>> plan = new ArrayList<>();
 175
 176         while (!Iterables.isEmpty(pendingGroups)) {
 177             final ItemSyncBox<Group> stepPlan = new ItemSyncBox<>();
 178             final Iterator<Group> iterator = pendingGroups.iterator();
 179             final Map<Uint32, Group> installIncrement = new HashMap<>();
 180
 181             while (iterator.hasNext()) {
 182                 final Group group = iterator.next();
 183
 184                 final Group existingGroup = installedGroups.get(group.getGroupId().getValue());
 185                 if (existingGroup != null) {
 186                     if (!gatherUpdates) {
 187                         iterator.remove();
 188                     } else {
 189                         // check buckets and eventually update
 190                         if (group.equals(existingGroup)) {
 191                             iterator.remove();
 192                         } else {
 193                             if (checkGroupPrecondition(installedGroups.keySet(), group)) {
 194                                 iterator.remove();
 195                                 LOG.trace("Group {} on device {} differs - planned for update", group.getGroupId(),
 196                                         nodeId);
 197                                 stepPlan.getItemsToUpdate().add(new ItemSyncBox.ItemUpdateTuple<>(existingGroup,
 198                                         group));
 199                             }
 200                         }
 201                     }
 202                 } else if (checkGroupPrecondition(installedGroups.keySet(), group)) {
 203                     iterator.remove();
 204                     installIncrement.put(group.getGroupId().getValue(), group);
 205                     stepPlan.getItemsToPush().add(group);
 206                 }
 207             }
 208
 209             if (!stepPlan.isEmpty()) {
 210                 // atomic update of installed flows in order to keep plan portions clean of local group dependencies
 211                 installedGroups.putAll(installIncrement);
 212                 plan.add(stepPlan);
 213             } else if (!pendingGroups.isEmpty()) {
 214                 LOG.warn("Failed to resolve and divide groups into preconditions-match based ordered plan: {}, "
 215                         + "resolving stuck at level {}", nodeId.getValue(), plan.size());
 216                 throw new IllegalStateException("Failed to resolve and divide groups when matching preconditions");
 217             }
 218         }
 219
 220         return plan;
 221     }
 222
 223     public static boolean checkGroupPrecondition(final Set<Uint32> installedGroupIds, final Group pendingGroup) {
 224         boolean okToInstall = true;
 225         // check each bucket in the pending group
 226         for (Bucket bucket : pendingGroup.getBuckets().nonnullBucket().values()) {
 227             for (Action action : bucket.nonnullAction().values()) {
 228                 // if the output action is a group
 229                 if (GroupActionCase.class.equals(action.getAction().implementedInterface())) {
 230                     Uint32 groupId = ((GroupActionCase) action.getAction()).getGroupAction().getGroupId();
 231                     // see if that output group is installed
 232                     if (!installedGroupIds.contains(groupId)) {
 233                         // if not installed, we have missing dependencies and cannot install this pending group
 234                         okToInstall = false;
 235                         break;
 236                     }
 237                 }
 238             }
 239             if (!okToInstall) {
 240                 break;
 241             }
 242         }
 243         return okToInstall;
 244     }
 245
 246     public static <E> int countTotalPushed(final Iterable<ItemSyncBox<E>> groupsAddPlan) {
 247         int count = 0;
 248         for (ItemSyncBox<E> groupItemSyncBox : groupsAddPlan) {
 249             count += groupItemSyncBox.getItemsToPush().size();
 250         }
 251         return count;
 252     }
 253
 254     public static <E> int countTotalUpdated(final Iterable<ItemSyncBox<E>> groupsAddPlan) {
 255         int count = 0;
 256         for (ItemSyncBox<E> groupItemSyncBox : groupsAddPlan) {
 257             count += groupItemSyncBox.getItemsToUpdate().size();
 258         }
 259         return count;
 260     }
 261
 262     /**
 263      * Resolves meter differences.
 264      *
 265      * @param nodeId              target node
 266      * @param meterOperationalMap meters present on device
 267      * @param metersConfigured    meters configured for device
 268      * @param gatherUpdates       check content of pending item if present on device (and create update task eventually)
 269      * @return synchronization box
 270      */
 271     public static ItemSyncBox<Meter> resolveMeterDiffs(final NodeId nodeId,
 272                                                        final Map<MeterId, Meter> meterOperationalMap,
 273                                                        final Collection<Meter> metersConfigured,
 274                                                        final boolean gatherUpdates) {
 275         LOG.trace("resolving meters for {}", nodeId.getValue());
 276         final ItemSyncBox<Meter> syncBox = new ItemSyncBox<>();
 277         for (Meter meter : metersConfigured) {
 278             final Meter existingMeter = meterOperationalMap.get(meter.getMeterId());
 279             if (existingMeter == null) {
 280                 syncBox.getItemsToPush().add(meter);
 281             } else {
 282                 // compare content and eventually update
 283                 if (gatherUpdates && !meter.equals(existingMeter)) {
 284                     syncBox.getItemsToUpdate().add(new ItemSyncBox.ItemUpdateTuple<>(existingMeter, meter));
 285                 }
 286             }
 287         }
 288         return syncBox;
 289     }
 290
 291     /**
 292      * Resolves flow differences in a table.
 293      *
 294      * @param flowsConfigured    flows resent on device
 295      * @param flowOperationalMap flows configured for device
 296      * @param gatherUpdates      check content of pending item if present on device (and create update task eventually)
 297      * @return list of safe synchronization steps
 298      */
 299     private static ItemSyncBox<Flow> resolveFlowDiffsInTable(final Collection<Flow> flowsConfigured,
 300                                                             final Map<FlowDescriptor, Flow> flowOperationalMap,
 301                                                             final boolean gatherUpdates) {
 302         final ItemSyncBox<Flow> flowsSyncBox = new ItemSyncBox<>();
 303         // loop configured flows and check if already present on device
 304         for (final Flow flow : flowsConfigured) {
 305             final Flow existingFlow = FlowCapableNodeLookups.flowMapLookupExisting(flow, flowOperationalMap);
 306
 307             if (existingFlow == null) {
 308                 flowsSyncBox.getItemsToPush().add(flow);
 309             } else {
 310                 // check instructions and eventually update
 311                 if (gatherUpdates && !Objects.equals(flow.getInstructions(), existingFlow.getInstructions())) {
 312                     flowsSyncBox.getItemsToUpdate().add(new ItemSyncBox.ItemUpdateTuple<>(existingFlow, flow));
 313                 }
 314             }
 315         }
 316         return flowsSyncBox;
 317     }
 318
 319     /**
 320      * Resolves flow differences in all tables.
 321      *
 322      * @param nodeId              target node
 323      * @param tableOperationalMap flow-tables resent on device
 324      * @param tablesConfigured    flow-tables configured for device
 325      * @param gatherUpdates       check content of pending item if present on device (and create update task eventually)
 326      * @return map : key={@link TableKey}, value={@link ItemSyncBox} of safe synchronization steps
 327      */
 328     public static Map<TableKey, ItemSyncBox<Flow>> resolveFlowDiffsInAllTables(final NodeId nodeId,
 329             final Map<Uint8, Table> tableOperationalMap, final Collection<Table> tablesConfigured,
 330             final boolean gatherUpdates) {
 331         LOG.trace("resolving flows in tables for {}", nodeId.getValue());
 332         final Map<TableKey, ItemSyncBox<Flow>> tableFlowSyncBoxes = new HashMap<>();
 333         for (final Table tableConfigured : tablesConfigured) {
 334             final Collection<Flow> flowsConfigured = tableConfigured.nonnullFlow().values();
 335             if (flowsConfigured.isEmpty()) {
 336                 continue;
 337             }
 338
 339             // lookup table (on device)
 340             final Table tableOperational = tableOperationalMap.get(tableConfigured.getId());
 341             // wrap existing (on device) flows in current table into map
 342             final Map<FlowDescriptor, Flow> flowOperationalMap = FlowCapableNodeLookups.wrapFlowsToMap(
 343                     tableOperational != null
 344                             ? tableOperational.nonnullFlow().values()
 345                             : null);
 346
 347
 348             final ItemSyncBox<Flow> flowsSyncBox = resolveFlowDiffsInTable(
 349                     flowsConfigured, flowOperationalMap, gatherUpdates);
 350             if (!flowsSyncBox.isEmpty()) {
 351                 tableFlowSyncBoxes.put(tableConfigured.key(), flowsSyncBox);
 352             }
 353         }
 354         return tableFlowSyncBoxes;
 355     }
 356
 357     public static Collection<Group> safeGroups(FlowCapableNode node) {
 358         return node == null ? Collections.emptyList() : node.nonnullGroup().values();
 359     }
 360
 361     public static Collection<Table> safeTables(FlowCapableNode node) {
 362         return node == null ? Collections.emptyList() : node.nonnullTable().values();
 363     }
 364
 365     public static Collection<Meter> safeMeters(FlowCapableNode node) {
 366         return node == null ? Collections.emptyList() : node.nonnullMeter().values();
 367     }
 368 }