2 * Copyright (c) 2015 Cisco Systems, Inc. and others. All rights reserved.
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
9 package org.opendaylight.groupbasedpolicy.renderer.ofoverlay;
11 import static org.opendaylight.groupbasedpolicy.renderer.ofoverlay.flow.FlowUtils.createGroupPath;
12 import static org.opendaylight.groupbasedpolicy.renderer.ofoverlay.flow.FlowUtils.createNodePath;
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;
21 import java.util.concurrent.ConcurrentHashMap;
22 import java.util.concurrent.ConcurrentMap;
23 import java.util.concurrent.ExecutionException;
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;
56 public class OfWriter {
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<>();
64 private static final Logger LOG = LoggerFactory.getLogger(OfWriter.class);
66 public Table getTableForNode(NodeId nodeId, short tableId) {
67 return getTableBuilderForNode(nodeId, tableId).build();
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>()));
76 return this.flowMap.get(tableIid);
79 public boolean groupExists(NodeId nodeId, long groupId) {
80 return (getGroupForNode(nodeId, groupId) != null);
84 * Gets group (or null if group does not exist) for node
86 * @param nodeId NodeId
88 * @return Group or null
90 private Group getGroupForNode(NodeId nodeId, long groupId) {
91 InstanceIdentifier<Group> giid = FlowUtils.createGroupPath(nodeId, groupId);
92 if (this.groupByIid.get(giid) == null) {
95 return this.groupByIid.get(giid).build();
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>
105 * @param nodeId NodeId
106 * @param groupId GroupId
107 * @see OfWriter#writeGroup(NodeId, GroupId, GroupTypes, String, String, Boolean)
109 public void writeGroup(NodeId nodeId, GroupId groupId) {
110 writeGroup(nodeId, groupId, GroupTypes.GroupAll, null, null, null);
114 * Writes a new group for OVS
116 * @param nodeId NodeId
117 * @param groupId GroupId
118 * @param groupTypes GroupTypes
119 * @param containerName String
120 * @param groupName String
121 * @param barrier Boolean
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);
129 GroupBuilder gb = new GroupBuilder().setGroupId(groupId)
131 .setContainerName(containerName)
132 .setGroupName(groupName)
133 .setGroupType(groupTypes)
134 .setBuckets(new BucketsBuilder().setBucket(new ArrayList<Bucket>()).build());
136 groupByIid.put(FlowUtils.createGroupPath(nodeId, groupId), gb);
137 if (this.groupIdsByNode.get(nodeId) == null) {
138 this.groupIdsByNode.put(nodeId, new HashSet<GroupId>());
140 this.groupIdsByNode.get(nodeId).add(groupId);
144 * Writes a Bucket to Group.<br>
145 * Group has to be created previously,<br>
146 * or an IllegalStateException will be thrown.
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)
155 public void writeBucket(NodeId nodeId, GroupId groupId, Bucket bucket) {
156 Preconditions.checkNotNull(nodeId);
157 Preconditions.checkNotNull(groupId);
158 Preconditions.checkNotNull(bucket);
160 GroupBuilder gb = groupByIid.get(FlowUtils.createGroupPath(nodeId, groupId));
162 gb.getBuckets().getBucket().add(bucket);
164 LOG.error("Group {} on node {} does not exist", groupId, nodeId);
165 throw new IllegalStateException();
169 public void writeFlow(NodeId nodeId, short tableId, Flow flow) {
170 Preconditions.checkNotNull(flow);
171 Preconditions.checkNotNull(nodeId);
173 if (flow.getMatch() == null) {
174 flow = new FlowBuilder(flow).setMatch(new MatchBuilder().build()).build();
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));
182 Equivalence.Wrapper<Flow> wFlow = EquivalenceFabric.FLOW_EQUIVALENCE.wrap(flow);
184 if (!wrappedFlows.contains(wFlow)) {
185 tableBuilder.getFlow().add(flow);
187 LOG.debug("Flow already exists in OfData - {}", flow);
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
195 * Newly created flows are returned and will be used as actual in next update
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
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) {
205 for (NodeId nodeId : groupIdsByNode.keySet()) {
207 updateGroups(dataBroker, nodeId);
208 } catch (ExecutionException | InterruptedException e) {
209 LOG.error("Could not update Group table on node {}", nodeId);
213 for (Map.Entry<InstanceIdentifier<Table>, TableBuilder> newEntry : flowMap.entrySet()) {
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())) {
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());
233 private void updateFlowTable(DataBroker dataBroker, Map.Entry<InstanceIdentifier<Table>, TableBuilder> desiredFlowMap,
234 Map.Entry<InstanceIdentifier<Table>, TableBuilder> actualFlowMap)
235 throws ExecutionException, InterruptedException {
238 List<Flow> actualFlows = new ArrayList<>();
239 if (actualFlowMap != null && actualFlowMap.getValue() != null) {
240 actualFlows = actualFlowMap.getValue().getFlow();
243 List<Flow> desiredFlows = new ArrayList<>(desiredFlowMap.getValue().getFlow());
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));
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);
256 final InstanceIdentifier<Table> tableIid = desiredFlowMap.getKey();
257 ReadWriteTransaction t = dataBroker.newReadWriteTransaction();
259 if (!deletions.isEmpty()) {
260 for (Equivalence.Wrapper<Flow> wf : deletions) {
263 t.delete(LogicalDatastoreType.CONFIGURATION,
264 FlowUtils.createFlowPath(tableIid, f.getId()));
268 if (!additions.isEmpty()) {
269 for (Equivalence.Wrapper<Flow> wf : additions) {
272 if (f.getMatch() == null) {
273 f = new FlowBuilder(f).setMatch(new MatchBuilder().build()).build();
275 t.put(LogicalDatastoreType.CONFIGURATION,
276 FlowUtils.createFlowPath(tableIid, f.getId()), f, true);
280 CheckedFuture<Void, TransactionCommitFailedException> f = t.submit();
281 Futures.addCallback(f, new FutureCallback<Void>() {
284 public void onFailure(Throwable t) {
285 LOG.error("Could not write flow table {}: {}", tableIid, t);
289 public void onSuccess(Void result) {
290 LOG.debug("Flow table {} updated.", tableIid);
295 private void updateGroups(DataBroker dataBroker, final NodeId nodeId)
296 throws ExecutionException, InterruptedException {
298 if (this.groupIdsByNode.get(nodeId) == null) {
299 this.groupIdsByNode.put(nodeId, new HashSet<GroupId>());
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());
307 createdGroups.add(g);
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);
323 if (fcn.getGroup() != null) {
324 existingGroups = new HashSet<>(fcn.getGroup());
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));
332 Sets.SetView<Equivalence.Wrapper<Group>> deletions =
333 Sets.difference(existingGroupsWrap, createdGroupsWrap);
334 Sets.SetView<Equivalence.Wrapper<Group>> additions =
335 Sets.difference(createdGroupsWrap, existingGroupsWrap);
337 if (!deletions.isEmpty()) {
338 for (Equivalence.Wrapper<Group> groupWrapper : deletions) {
339 Group g = groupWrapper.get();
341 LOG.debug("Deleting group {} on node {}", g.getGroupId(), nodeId);
342 t.delete(LogicalDatastoreType.CONFIGURATION,
343 createGroupPath(nodeId, g.getGroupId()));
347 if (!additions.isEmpty()) {
348 for (Equivalence.Wrapper<Group> groupWrapper : additions) {
349 Group g = groupWrapper.get();
351 LOG.debug("Putting node {}, group {}", nodeId, g.getGroupId());
352 t.put(LogicalDatastoreType.CONFIGURATION,
353 createGroupPath(nodeId, g.getGroupId()), g, true);
358 CheckedFuture<Void, TransactionCommitFailedException> f = t.submit();
359 Futures.addCallback(f, new FutureCallback<Void>() {
362 public void onFailure(Throwable t) {
363 LOG.error("Could not write group table on node {}: {}", nodeId, t);
367 public void onSuccess(Void result) {
368 LOG.debug("Group table on node {} updated.", nodeId);