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