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