renderers/ofoverlay/src/main/java/org/opendaylight/groupbasedpolicy/renderer/ofoverlay/OfWriter.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.groupbasedpolicy.renderer.ofoverlay;
  10
  11 import static org.opendaylight.groupbasedpolicy.renderer.ofoverlay.flow.FlowUtils.createGroupPath;
  12 import static org.opendaylight.groupbasedpolicy.renderer.ofoverlay.flow.FlowUtils.createNodePath;
  13
  14 import javax.annotation.Nullable;
  15 import java.util.ArrayList;
  16 import java.util.HashMap;
  17 import java.util.HashSet;
  18 import java.util.List;
  19 import java.util.Map;
  20 import java.util.Set;
  21 import java.util.concurrent.ConcurrentHashMap;
  22 import java.util.concurrent.ConcurrentMap;
  23 import java.util.concurrent.ExecutionException;
  24
  25 import com.google.common.base.Equivalence;
  26 import com.google.common.base.Optional;
  27 import com.google.common.base.Preconditions;
  28 import com.google.common.collect.Collections2;
  29 import com.google.common.collect.Sets;
  30 import com.google.common.util.concurrent.CheckedFuture;
  31 import com.google.common.util.concurrent.FutureCallback;
  32 import com.google.common.util.concurrent.Futures;
  33 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  34 import org.opendaylight.controller.md.sal.binding.api.ReadWriteTransaction;
  35 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  36 import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
  37 import org.opendaylight.groupbasedpolicy.renderer.ofoverlay.equivalence.EquivalenceFabric;
  38 import org.opendaylight.groupbasedpolicy.renderer.ofoverlay.flow.FlowUtils;
  39 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
  40 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.Table;
  41 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.TableBuilder;
  42 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow;
  43 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.FlowBuilder;
  44 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.rev131026.flow.MatchBuilder;
  45 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.GroupId;
  46 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.GroupTypes;
  47 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.group.BucketsBuilder;
  48 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.group.buckets.Bucket;
  49 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.groups.Group;
  50 import org.opendaylight.yang.gen.v1.urn.opendaylight.group.types.rev131018.groups.GroupBuilder;
  51 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
  52 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  53 import org.slf4j.Logger;
  54 import org.slf4j.LoggerFactory;
  55
  56 public class OfWriter {
  57
  58     private final ConcurrentMap<InstanceIdentifier<Table>, TableBuilder> flowMap =
  59             new ConcurrentHashMap<>();
  60     private final ConcurrentMap<InstanceIdentifier<Group>, GroupBuilder> groupByIid =
  61             new ConcurrentHashMap<>();
  62     private final ConcurrentMap<NodeId, Set<GroupId>> groupIdsByNode = new ConcurrentHashMap<>();
  63
  64     private static final Logger LOG = LoggerFactory.getLogger(OfWriter.class);
  65
  66     public Table getTableForNode(NodeId nodeId, short tableId) {
  67         return getTableBuilderForNode(nodeId, tableId).build();
  68     }
  69
  70     private TableBuilder getTableBuilderForNode(NodeId nodeId, short tableId) {
  71         InstanceIdentifier<Table> tableIid = FlowUtils.createTablePath(nodeId, tableId);
  72         if (this.flowMap.get(tableIid) == null) {
  73             this.flowMap.put(tableIid,
  74                     new TableBuilder().setId(tableId).setFlow(new ArrayList<Flow>()));
  75         }
  76         return this.flowMap.get(tableIid);
  77     }
  78
  79     public boolean groupExists(NodeId nodeId, long groupId) {
  80         return (getGroupForNode(nodeId, groupId) != null);
  81     }
  82
  83     /**
  84      * Gets group (or null if group does not exist) for node
  85      *
  86      * @param nodeId  NodeId
  87      * @param groupId long
  88      * @return Group or null
  89      */
  90     private Group getGroupForNode(NodeId nodeId, long groupId) {
  91         InstanceIdentifier<Group> giid = FlowUtils.createGroupPath(nodeId, groupId);
  92         if (this.groupByIid.get(giid) == null) {
  93             return null;
  94         }
  95         return this.groupByIid.get(giid).build();
  96     }
  97
  98     /**
  99      * Short form of {@link #writeGroup(NodeId, GroupId, GroupTypes, String, String, Boolean)} with default parameters:<br>
 100      * groupTypes = {@code GroupTypes.GroupAll}<br>
 101      * containerName = null<br>
 102      * groupName = null<br>
 103      * barrier = null
 104      *
 105      * @param nodeId     NodeId
 106      * @param groupId    GroupId
 107      * @see OfWriter#writeGroup(NodeId, GroupId, GroupTypes, String, String, Boolean)
 108      */
 109     public void writeGroup(NodeId nodeId, GroupId groupId) {
 110         writeGroup(nodeId, groupId, GroupTypes.GroupAll, null, null, null);
 111     }
 112
 113     /**
 114      * Writes a new group for OVS
 115      *
 116      * @param nodeId        NodeId
 117      * @param groupId       GroupId
 118      * @param groupTypes    GroupTypes
 119      * @param containerName String
 120      * @param groupName     String
 121      * @param barrier       Boolean
 122      */
 123     public void writeGroup(NodeId nodeId, GroupId groupId, @Nullable GroupTypes groupTypes,
 124             @Nullable String containerName, @Nullable String groupName,
 125             @Nullable Boolean barrier) {
 126         Preconditions.checkNotNull(nodeId);
 127         Preconditions.checkNotNull(groupId);
 128
 129         GroupBuilder gb = new GroupBuilder().setGroupId(groupId)
 130                 .setBarrier(barrier)
 131                 .setContainerName(containerName)
 132                 .setGroupName(groupName)
 133                 .setGroupType(groupTypes)
 134                 .setBuckets(new BucketsBuilder().setBucket(new ArrayList<Bucket>()).build());
 135
 136         groupByIid.put(FlowUtils.createGroupPath(nodeId, groupId), gb);
 137         if (this.groupIdsByNode.get(nodeId) == null) {
 138             this.groupIdsByNode.put(nodeId, new HashSet<GroupId>());
 139         }
 140         this.groupIdsByNode.get(nodeId).add(groupId);
 141     }
 142
 143     /**
 144      * Writes a Bucket to Group.<br>
 145      * Group has to be created previously,<br>
 146      * or an IllegalStateException will be thrown.
 147      *
 148      * @param nodeId  NodeId
 149      * @param groupId GroupId
 150      * @param bucket  Bucket to be added to group
 151      * @throws IllegalStateException if the Group is absent
 152      * @see #writeGroup(NodeId, GroupId, GroupTypes, String, String, Boolean)
 153      * @see #writeGroup(NodeId, GroupId)
 154      */
 155     public void writeBucket(NodeId nodeId, GroupId groupId, Bucket bucket) {
 156         Preconditions.checkNotNull(nodeId);
 157         Preconditions.checkNotNull(groupId);
 158         Preconditions.checkNotNull(bucket);
 159
 160         GroupBuilder gb = groupByIid.get(FlowUtils.createGroupPath(nodeId, groupId));
 161         if (gb != null) {
 162             gb.getBuckets().getBucket().add(bucket);
 163         } else {
 164             LOG.error("Group {} on node {} does not exist", groupId, nodeId);
 165             throw new IllegalStateException();
 166         }
 167     }
 168
 169     public void writeFlow(NodeId nodeId, short tableId, Flow flow) {
 170         Preconditions.checkNotNull(flow);
 171         Preconditions.checkNotNull(nodeId);
 172
 173         if (flow.getMatch() == null) {
 174             flow = new FlowBuilder(flow).setMatch(new MatchBuilder().build()).build();
 175         }
 176         TableBuilder tableBuilder = this.getTableBuilderForNode(nodeId, tableId);
 177         // transforming List<Flow> to Set (with customized equals/hashCode) to eliminate duplicate entries
 178         List<Flow> flows = tableBuilder.getFlow();
 179         Set<Equivalence.Wrapper<Flow>> wrappedFlows = new HashSet<>(
 180                 Collections2.transform(flows, EquivalenceFabric.FLOW_WRAPPER_FUNCTION));
 181
 182         Equivalence.Wrapper<Flow> wFlow = EquivalenceFabric.FLOW_EQUIVALENCE.wrap(flow);
 183
 184         if (!wrappedFlows.contains(wFlow)) {
 185             tableBuilder.getFlow().add(flow);
 186         } else {
 187             LOG.debug("Flow already exists in OfData - {}", flow);
 188         }
 189     }
 190
 191     /**
 192      * Update groups and flows on every node
 193      * Only flows created by gbp - which are present in actualFlowMap - can be removed. It ensures no other flows
 194      * are deleted
 195      * Newly created flows are returned and will be used as actual in next update
 196      *
 197      * @param actualFlowMap map of flows which are currently present on all nodes
 198      * @return map of newly created flows. These flows will be "actual" in next update
 199      */
 200     public Map<InstanceIdentifier<Table>, TableBuilder> commitToDataStore(DataBroker dataBroker,
 201                                                                           Map<InstanceIdentifier<Table>, TableBuilder> actualFlowMap) {
 202         Map<InstanceIdentifier<Table>, TableBuilder> actualFlows = new HashMap<>();
 203         if (dataBroker != null) {
 204
 205             for (NodeId nodeId : groupIdsByNode.keySet()) {
 206                 try {
 207                     updateGroups(dataBroker, nodeId);
 208                 } catch (ExecutionException | InterruptedException e) {
 209                     LOG.error("Could not update Group table on node {}", nodeId);
 210                 }
 211             }
 212
 213             for (Map.Entry<InstanceIdentifier<Table>, TableBuilder> newEntry : flowMap.entrySet()) {
 214                 try {
 215                     // Get actual flows on the same node/table
 216                     Map.Entry<InstanceIdentifier<Table>, TableBuilder> actualEntry = null;
 217                     for (Map.Entry<InstanceIdentifier<Table>, TableBuilder> a : actualFlowMap.entrySet()) {
 218                         if (a.getKey().equals(newEntry.getKey())) {
 219                             actualEntry = a;
 220                         }
 221                     }
 222                     // Get the currently configured flows for this table
 223                     updateFlowTable(dataBroker, newEntry, actualEntry);
 224                     actualFlows.put(newEntry.getKey(), newEntry.getValue());
 225                 } catch (Exception e) {
 226                     LOG.warn("Couldn't read flow table {}", newEntry.getKey());
 227                 }
 228             }
 229         }
 230         return actualFlows;
 231     }
 232
 233     private void updateFlowTable(DataBroker dataBroker, Map.Entry<InstanceIdentifier<Table>, TableBuilder> desiredFlowMap,
 234                                  Map.Entry<InstanceIdentifier<Table>, TableBuilder> actualFlowMap)
 235             throws ExecutionException, InterruptedException {
 236
 237         // Actual state
 238         List<Flow> actualFlows = new ArrayList<>();
 239         if (actualFlowMap != null && actualFlowMap.getValue() != null) {
 240             actualFlows = actualFlowMap.getValue().getFlow();
 241         }
 242         // New state
 243         List<Flow> desiredFlows = new ArrayList<>(desiredFlowMap.getValue().getFlow());
 244
 245         // Sets with custom equivalence rules
 246         Set<Equivalence.Wrapper<Flow>> wrappedActualFlows = new HashSet<>(
 247                 Collections2.transform(actualFlows, EquivalenceFabric.FLOW_WRAPPER_FUNCTION));
 248         Set<Equivalence.Wrapper<Flow>> wrappedDesiredFlows = new HashSet<>(
 249                 Collections2.transform(desiredFlows, EquivalenceFabric.FLOW_WRAPPER_FUNCTION));
 250
 251         // All gbp flows which are not updated will be removed
 252         Sets.SetView<Equivalence.Wrapper<Flow>> deletions = Sets.difference(wrappedActualFlows, wrappedDesiredFlows);
 253         // New flows (they were not there before)
 254         Sets.SetView<Equivalence.Wrapper<Flow>> additions = Sets.difference(wrappedDesiredFlows, wrappedActualFlows);
 255
 256         final InstanceIdentifier<Table> tableIid = desiredFlowMap.getKey();
 257         ReadWriteTransaction t = dataBroker.newReadWriteTransaction();
 258
 259         if (!deletions.isEmpty()) {
 260             for (Equivalence.Wrapper<Flow> wf : deletions) {
 261                 Flow f = wf.get();
 262                 if (f != null) {
 263                     t.delete(LogicalDatastoreType.CONFIGURATION,
 264                             FlowUtils.createFlowPath(tableIid, f.getId()));
 265                 }
 266             }
 267         }
 268         if (!additions.isEmpty()) {
 269             for (Equivalence.Wrapper<Flow> wf : additions) {
 270                 Flow f = wf.get();
 271                 if (f != null) {
 272                     if (f.getMatch() == null) {
 273                         f = new FlowBuilder(f).setMatch(new MatchBuilder().build()).build();
 274                     }
 275                     t.put(LogicalDatastoreType.CONFIGURATION,
 276                             FlowUtils.createFlowPath(tableIid, f.getId()), f, true);
 277                 }
 278             }
 279         }
 280         CheckedFuture<Void, TransactionCommitFailedException> f = t.submit();
 281         Futures.addCallback(f, new FutureCallback<Void>() {
 282
 283             @Override
 284             public void onFailure(Throwable t) {
 285                 LOG.error("Could not write flow table {}: {}", tableIid, t);
 286             }
 287
 288             @Override
 289             public void onSuccess(Void result) {
 290                 LOG.debug("Flow table {} updated.", tableIid);
 291             }
 292         });
 293     }
 294
 295     private void updateGroups(DataBroker dataBroker, final NodeId nodeId)
 296             throws ExecutionException, InterruptedException {
 297
 298         if (this.groupIdsByNode.get(nodeId) == null) {
 299             this.groupIdsByNode.put(nodeId, new HashSet<GroupId>());
 300         }
 301         Set<GroupId> createdGroupIds = new HashSet<>(this.groupIdsByNode.get(nodeId));
 302         // groups from inner structure
 303         Set<Group> createdGroups = new HashSet<>();
 304         for (GroupId gid : createdGroupIds) {
 305             Group g = getGroupForNode(nodeId, gid.getValue());
 306             if (g != null) {
 307                 createdGroups.add(g);
 308             }
 309         }
 310         // groups from datastore
 311         Set<Group> existingGroups = new HashSet<>();
 312         ReadWriteTransaction t = dataBroker.newReadWriteTransaction();
 313         FlowCapableNode fcn = null;
 314         InstanceIdentifier<FlowCapableNode> fcniid =
 315                 createNodePath(nodeId).builder().augmentation(FlowCapableNode.class).build();
 316         Optional<FlowCapableNode> r = t.read(LogicalDatastoreType.OPERATIONAL, fcniid).get();
 317         if (!r.isPresent()) {
 318             LOG.warn("Node {} is not present", fcniid);
 319             return;
 320         }
 321         fcn = r.get();
 322
 323         if (fcn.getGroup() != null) {
 324             existingGroups = new HashSet<>(fcn.getGroup());
 325         }
 326
 327         Set<Equivalence.Wrapper<Group>> existingGroupsWrap = new HashSet<>(
 328                 Collections2.transform(existingGroups, EquivalenceFabric.GROUP_WRAPPER_FUNCTION));
 329         Set<Equivalence.Wrapper<Group>> createdGroupsWrap = new HashSet<>(
 330                 Collections2.transform(createdGroups, EquivalenceFabric.GROUP_WRAPPER_FUNCTION));
 331
 332         Sets.SetView<Equivalence.Wrapper<Group>> deletions =
 333                 Sets.difference(existingGroupsWrap, createdGroupsWrap);
 334         Sets.SetView<Equivalence.Wrapper<Group>> additions =
 335                 Sets.difference(createdGroupsWrap, existingGroupsWrap);
 336
 337         if (!deletions.isEmpty()) {
 338             for (Equivalence.Wrapper<Group> groupWrapper : deletions) {
 339                 Group g = groupWrapper.get();
 340                 if (g != null) {
 341                     LOG.debug("Deleting group {} on node {}", g.getGroupId(), nodeId);
 342                     t.delete(LogicalDatastoreType.CONFIGURATION,
 343                             createGroupPath(nodeId, g.getGroupId()));
 344                 }
 345             }
 346         }
 347         if (!additions.isEmpty()) {
 348             for (Equivalence.Wrapper<Group> groupWrapper : additions) {
 349                 Group g = groupWrapper.get();
 350                 if (g != null) {
 351                     LOG.debug("Putting node {}, group {}", nodeId, g.getGroupId());
 352                     t.put(LogicalDatastoreType.CONFIGURATION,
 353                             createGroupPath(nodeId, g.getGroupId()), g, true);
 354                 }
 355             }
 356         }
 357
 358         CheckedFuture<Void, TransactionCommitFailedException> f = t.submit();
 359         Futures.addCallback(f, new FutureCallback<Void>() {
 360
 361             @Override
 362             public void onFailure(Throwable t) {
 363                 LOG.error("Could not write group table on node {}: {}", nodeId, t);
 364             }
 365
 366             @Override
 367             public void onSuccess(Void result) {
 368                 LOG.debug("Group table on node {} updated.", nodeId);
 369             }
 370         });
 371     }
 372
 373 }