openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/util/FlowUtil.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
   9 package org.opendaylight.openflowplugin.impl.util;
  10
  11 import com.google.common.annotations.VisibleForTesting;
  12 import com.google.common.base.Function;
  13 import com.google.common.base.Preconditions;
  14 import java.util.ArrayList;
  15 import java.util.Collection;
  16 import java.util.Collections;
  17 import java.util.Iterator;
  18 import java.util.List;
  19 import java.util.concurrent.atomic.AtomicInteger;
  20 import javax.annotation.Nullable;
  21 import org.apache.commons.lang3.tuple.Pair;
  22 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
  23 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowId;
  24 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.Table;
  25 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.TableKey;
  26 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow;
  27 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.FlowKey;
  28 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.rev131026.FlowRef;
  29 import org.opendaylight.yang.gen.v1.urn.opendaylight.flows.service.rev160314.AddFlowsBatchOutput;
  30 import org.opendaylight.yang.gen.v1.urn.opendaylight.flows.service.rev160314.AddFlowsBatchOutputBuilder;
  31 import org.opendaylight.yang.gen.v1.urn.opendaylight.flows.service.rev160314.BatchFlowIdGrouping;
  32 import org.opendaylight.yang.gen.v1.urn.opendaylight.flows.service.rev160314.BatchFlowOutputListGrouping;
  33 import org.opendaylight.yang.gen.v1.urn.opendaylight.flows.service.rev160314.RemoveFlowsBatchOutput;
  34 import org.opendaylight.yang.gen.v1.urn.opendaylight.flows.service.rev160314.RemoveFlowsBatchOutputBuilder;
  35 import org.opendaylight.yang.gen.v1.urn.opendaylight.flows.service.rev160314.UpdateFlowsBatchOutput;
  36 import org.opendaylight.yang.gen.v1.urn.opendaylight.flows.service.rev160314.UpdateFlowsBatchOutputBuilder;
  37 import org.opendaylight.yang.gen.v1.urn.opendaylight.flows.service.rev160314.batch.flow.output.list.grouping.BatchFailedFlowsOutput;
  38 import org.opendaylight.yang.gen.v1.urn.opendaylight.flows.service.rev160314.batch.flow.output.list.grouping.BatchFailedFlowsOutputBuilder;
  39 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
  40 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  41 import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier;
  42 import org.opendaylight.yangtools.yang.common.RpcError;
  43 import org.opendaylight.yangtools.yang.common.RpcResult;
  44 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
  45 import org.slf4j.Logger;
  46 import org.slf4j.LoggerFactory;
  47
  48 /**
  49  * Created by Martin Bobak &lt;mbobak@cisco.com&gt; on 8.4.2015.
  50  */
  51 public final class FlowUtil {
  52
  53     private static final String ALIEN_SYSTEM_FLOW_ID = "#UF$TABLE*";
  54     private static final AtomicInteger unaccountedFlowsCounter = new AtomicInteger(0);
  55     private static final Logger LOG = LoggerFactory.getLogger(FlowUtil.class);
  56     private static final RpcResultBuilder<List<BatchFailedFlowsOutput>> SUCCESSFUL_FLOW_OUTPUT_RPC_RESULT =
  57             RpcResultBuilder.success(Collections.<BatchFailedFlowsOutput>emptyList());
  58
  59     /** Attach barrier response to given {@link RpcResult}&lt;RemoveFlowsBatchOutput&gt; */
  60     public static final Function<Pair<RpcResult<RemoveFlowsBatchOutput>, RpcResult<Void>>, RpcResult<RemoveFlowsBatchOutput>>
  61             FLOW_REMOVE_COMPOSING_TRANSFORM = createComposingFunction();
  62
  63     /** Attach barrier response to given {@link RpcResult}&lt;AddFlowsBatchOutput&gt; */
  64     public static final Function<Pair<RpcResult<AddFlowsBatchOutput>, RpcResult<Void>>, RpcResult<AddFlowsBatchOutput>>
  65             FLOW_ADD_COMPOSING_TRANSFORM = createComposingFunction();
  66
  67     /** Attach barrier response to given {@link RpcResult}&lt;UpdateFlowsBatchOutput&gt; */
  68     public static final Function<Pair<RpcResult<UpdateFlowsBatchOutput>, RpcResult<Void>>, RpcResult<UpdateFlowsBatchOutput>>
  69             FLOW_UPDATE_COMPOSING_TRANSFORM = createComposingFunction();
  70
  71     /**
  72      * Gather errors into collection and wrap it into {@link RpcResult} and propagate all {@link RpcError}
  73      */
  74     public static final Function<RpcResult<List<BatchFailedFlowsOutput>>, RpcResult<RemoveFlowsBatchOutput>> FLOW_REMOVE_TRANSFORM =
  75             new Function<RpcResult<List<BatchFailedFlowsOutput>>, RpcResult<RemoveFlowsBatchOutput>>() {
  76                 @Nullable
  77                 @Override
  78                 public RpcResult<RemoveFlowsBatchOutput> apply(@Nullable final RpcResult<List<BatchFailedFlowsOutput>> batchFlowsCumulativeResult) {
  79                     final RemoveFlowsBatchOutput batchOutput = new RemoveFlowsBatchOutputBuilder()
  80                             .setBatchFailedFlowsOutput(batchFlowsCumulativeResult.getResult()).build();
  81
  82                     final RpcResultBuilder<RemoveFlowsBatchOutput> resultBld =
  83                             createCumulativeRpcResult(batchFlowsCumulativeResult, batchOutput);
  84                     return resultBld.build();
  85                 }
  86             };
  87
  88     /**
  89      * Gather errors into collection and wrap it into {@link RpcResult} and propagate all {@link RpcError}
  90      */
  91     public static final Function<RpcResult<List<BatchFailedFlowsOutput>>, RpcResult<AddFlowsBatchOutput>> FLOW_ADD_TRANSFORM =
  92             new Function<RpcResult<List<BatchFailedFlowsOutput>>, RpcResult<AddFlowsBatchOutput>>() {
  93                 @Nullable
  94                 @Override
  95                 public RpcResult<AddFlowsBatchOutput> apply(@Nullable final RpcResult<List<BatchFailedFlowsOutput>> batchFlowsCumulativeResult) {
  96                     final AddFlowsBatchOutput batchOutput = new AddFlowsBatchOutputBuilder()
  97                             .setBatchFailedFlowsOutput(batchFlowsCumulativeResult.getResult()).build();
  98
  99                     final RpcResultBuilder<AddFlowsBatchOutput> resultBld =
 100                             createCumulativeRpcResult(batchFlowsCumulativeResult, batchOutput);
 101                     return resultBld.build();
 102                 }
 103             };
 104
 105     /**
 106      * Gather errors into collection and wrap it into {@link RpcResult} and propagate all {@link RpcError}
 107      */
 108     public static final Function<RpcResult<List<BatchFailedFlowsOutput>>, RpcResult<UpdateFlowsBatchOutput>> FLOW_UPDATE_TRANSFORM =
 109             new Function<RpcResult<List<BatchFailedFlowsOutput>>, RpcResult<UpdateFlowsBatchOutput>>() {
 110                 @Nullable
 111                 @Override
 112                 public RpcResult<UpdateFlowsBatchOutput> apply(@Nullable final RpcResult<List<BatchFailedFlowsOutput>> batchFlowsCumulativeResult) {
 113                     final UpdateFlowsBatchOutput batchOutput = new UpdateFlowsBatchOutputBuilder()
 114                             .setBatchFailedFlowsOutput(batchFlowsCumulativeResult.getResult()).build();
 115
 116                     final RpcResultBuilder<UpdateFlowsBatchOutput> resultBld =
 117                             createCumulativeRpcResult(batchFlowsCumulativeResult, batchOutput);
 118                     return resultBld.build();
 119                 }
 120             };
 121
 122     private FlowUtil() {
 123         throw new IllegalStateException("This class should not be instantiated.");
 124     }
 125
 126     /**
 127      * Wrap given list of problematic flow-ids into {@link RpcResult} of given type.
 128      *
 129      * @param batchFlowsCumulativeResult list of ids failed flows
 130      * @param batchOutput
 131      * @param <T>                        flow operation type
 132      * @return batch flow operation output of given type containing list of flow-ids and corresponding success flag
 133      */
 134     private static <T extends BatchFlowOutputListGrouping>
 135     RpcResultBuilder<T> createCumulativeRpcResult(final @Nullable RpcResult<List<BatchFailedFlowsOutput>> batchFlowsCumulativeResult,
 136                                                   final T batchOutput) {
 137         final RpcResultBuilder<T> resultBld;
 138         if (batchFlowsCumulativeResult.isSuccessful()) {
 139             resultBld = RpcResultBuilder.success(batchOutput);
 140         } else {
 141             resultBld = RpcResultBuilder.failed();
 142             resultBld.withResult(batchOutput)
 143                     .withRpcErrors(batchFlowsCumulativeResult.getErrors());
 144         }
 145         return resultBld;
 146     }
 147
 148     public static FlowId createAlienFlowId(final short tableId) {
 149         final StringBuilder sBuilder = new StringBuilder(ALIEN_SYSTEM_FLOW_ID)
 150                 .append(tableId).append('-').append(unaccountedFlowsCounter.incrementAndGet());
 151         String alienId = sBuilder.toString();
 152         return new FlowId(alienId);
 153
 154     }
 155
 156     /**
 157      * Factory method: create {@link Function} which attaches barrier response to given {@link RpcResult}&lt;T&gt;
 158      * and changes success flag if needed.
 159      * <br>
 160      * Original rpcResult is the {@link Pair#getLeft()} and barrier result is the {@link Pair#getRight()}.
 161      *
 162      * @param <T> type of rpcResult value
 163      * @return reusable static function
 164      */
 165     @VisibleForTesting
 166     static <T extends BatchFlowOutputListGrouping>
 167     Function<Pair<RpcResult<T>, RpcResult<Void>>, RpcResult<T>> createComposingFunction() {
 168         return new Function<Pair<RpcResult<T>, RpcResult<Void>>, RpcResult<T>>() {
 169             @Nullable
 170             @Override
 171             public RpcResult<T> apply(@Nullable final Pair<RpcResult<T>, RpcResult<Void>> input) {
 172                 final RpcResultBuilder<T> resultBld;
 173                 if (input.getLeft().isSuccessful() && input.getRight().isSuccessful()) {
 174                     resultBld = RpcResultBuilder.success();
 175                 } else {
 176                     resultBld = RpcResultBuilder.failed();
 177                 }
 178
 179                 final ArrayList<RpcError> rpcErrors = new ArrayList<>(input.getLeft().getErrors());
 180                 rpcErrors.addAll(input.getRight().getErrors());
 181                 resultBld.withRpcErrors(rpcErrors);
 182
 183                 resultBld.withResult(input.getLeft().getResult());
 184
 185                 return resultBld.build();
 186             }
 187         };
 188     }
 189
 190     /**
 191      * @param nodePath path to {@link Node}
 192      * @param tableId  path to {@link Table} under {@link Node}
 193      * @param flowId   path to {@link Flow} under {@link Table}
 194      * @return instance identifier assembled for given node, table and flow
 195      */
 196     public static FlowRef buildFlowPath(final InstanceIdentifier<Node> nodePath,
 197                                         final short tableId, final FlowId flowId) {
 198         final KeyedInstanceIdentifier<Flow, FlowKey> flowPath = nodePath
 199                 .augmentation(FlowCapableNode.class)
 200                 .child(Table.class, new TableKey(tableId))
 201                 .child(Flow.class, new FlowKey(new FlowId(flowId)));
 202
 203         return new FlowRef(flowPath);
 204     }
 205
 206     /**
 207      * Factory method: creates {@link Function} which keeps info of original inputs (passed to flow-rpc) and processes
 208      * list of all flow-rpc results.
 209      *
 210      * @param inputBatchFlows collection of problematic flow-ids wrapped in container of given type &lt;O&gt;
 211      * @param <O>             result container type
 212      * @return static reusable function
 213      */
 214     public static <O> Function<List<RpcResult<O>>, RpcResult<List<BatchFailedFlowsOutput>>> createCumulatingFunction(
 215             final List<? extends BatchFlowIdGrouping> inputBatchFlows) {
 216         return new Function<List<RpcResult<O>>, RpcResult<List<BatchFailedFlowsOutput>>>() {
 217             @Nullable
 218             @Override
 219             public RpcResult<List<BatchFailedFlowsOutput>> apply(@Nullable final List<RpcResult<O>> innerInput) {
 220                 final int sizeOfFutures = innerInput.size();
 221                 final int sizeOfInputBatch = inputBatchFlows.size();
 222                 Preconditions.checkArgument(sizeOfFutures == sizeOfInputBatch,
 223                         "wrong amount of returned futures: {} <> {}", sizeOfFutures, sizeOfInputBatch);
 224
 225                 final ArrayList<BatchFailedFlowsOutput> batchFlows = new ArrayList<>(sizeOfFutures);
 226                 final Iterator<? extends BatchFlowIdGrouping> batchFlowIterator = inputBatchFlows.iterator();
 227
 228                 Collection<RpcError> flowErrors = new ArrayList<>(sizeOfFutures);
 229
 230                 int batchOrder = 0;
 231                 for (RpcResult<O> flowModOutput : innerInput) {
 232                     final FlowId flowId = batchFlowIterator.next().getFlowId();
 233
 234                     if (!flowModOutput.isSuccessful()) {
 235                         batchFlows.add(new BatchFailedFlowsOutputBuilder()
 236                                 .setFlowId(flowId)
 237                                 .setBatchOrder(batchOrder)
 238                                 .build());
 239                         flowErrors.addAll(flowModOutput.getErrors());
 240                     }
 241                     batchOrder++;
 242                 }
 243
 244                 final RpcResultBuilder<List<BatchFailedFlowsOutput>> resultBuilder;
 245                 if (!flowErrors.isEmpty()) {
 246                     resultBuilder = RpcResultBuilder.<List<BatchFailedFlowsOutput>>failed()
 247                             .withRpcErrors(flowErrors).withResult(batchFlows);
 248                 } else {
 249                     resultBuilder = SUCCESSFUL_FLOW_OUTPUT_RPC_RESULT;
 250                 }
 251                 return resultBuilder.build();
 252             }
 253         };
 254     }
 255 }