opendaylight/md-sal/compatibility/sal-compatibility/src/main/java/org/opendaylight/controller/sal/compatibility/FlowProgrammerAdapter.java

   1 /**
   2  * Copyright (c) 2014 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 package org.opendaylight.controller.sal.compatibility;
   9
  10 import java.util.EnumSet;
  11 import java.util.Map;
  12 import java.util.UUID;
  13 import java.util.concurrent.ConcurrentHashMap;
  14 import java.util.concurrent.ExecutionException;
  15 import java.util.concurrent.Future;
  16
  17 import org.opendaylight.controller.clustering.services.CacheConfigException;
  18 import org.opendaylight.controller.clustering.services.CacheExistException;
  19 import org.opendaylight.controller.clustering.services.IClusterGlobalServices;
  20 import org.opendaylight.controller.clustering.services.IClusterServices.cacheMode;
  21 import org.opendaylight.controller.md.sal.common.api.TransactionStatus;
  22 import org.opendaylight.controller.sal.binding.api.data.DataBrokerService;
  23 import org.opendaylight.controller.sal.binding.api.data.DataModificationTransaction;
  24 import org.opendaylight.controller.sal.core.ConstructionException;
  25 import org.opendaylight.controller.sal.core.Node;
  26 import org.opendaylight.controller.sal.flowprogrammer.Flow;
  27 import org.opendaylight.controller.sal.flowprogrammer.IPluginInFlowProgrammerService;
  28 import org.opendaylight.controller.sal.flowprogrammer.IPluginOutFlowProgrammerService;
  29 import org.opendaylight.controller.sal.utils.Status;
  30 import org.opendaylight.controller.sal.utils.StatusCode;
  31 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
  32 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.Table;
  33 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.TableKey;
  34 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.FlowKey;
  35 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.FlowAdded;
  36 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.FlowRemoved;
  37 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.FlowUpdated;
  38 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.NodeErrorNotification;
  39 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.NodeExperimenterErrorNotification;
  40 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.SalFlowListener;
  41 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.SalFlowService;
  42 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.SwitchFlowRemoved;
  43 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
  44 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRef;
  45 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.Nodes;
  46 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.NodeKey;
  47 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  48 import org.opendaylight.yangtools.yang.common.RpcResult;
  49 import org.slf4j.Logger;
  50 import org.slf4j.LoggerFactory;
  51
  52 public class FlowProgrammerAdapter implements IPluginInFlowProgrammerService, SalFlowListener {
  53     private final static Logger LOG = LoggerFactory.getLogger(FlowProgrammerAdapter.class);
  54
  55     // Note: clustering services manipulate this
  56     private final Map<Flow, UUID> flowToFlowId = new ConcurrentHashMap<Flow, UUID>();
  57     private final static String CACHE_NAME = "flowprogrammeradapter.flowtoid";
  58
  59     // These are injected via Apache DM (see ComponentActivator)
  60     private IPluginOutFlowProgrammerService flowProgrammerPublisher;
  61     private IClusterGlobalServices clusterGlobalServices;
  62     private DataBrokerService dataBrokerService;
  63     private SalFlowService delegate;
  64
  65     public SalFlowService getDelegate() {
  66         return this.delegate;
  67     }
  68
  69     public void setDelegate(final SalFlowService delegate) {
  70         this.delegate = delegate;
  71     }
  72
  73     public DataBrokerService getDataBrokerService() {
  74         return this.dataBrokerService;
  75     }
  76
  77     public void setDataBrokerService(final DataBrokerService dataBrokerService) {
  78         this.dataBrokerService = dataBrokerService;
  79     }
  80
  81     public IPluginOutFlowProgrammerService getFlowProgrammerPublisher() {
  82         return this.flowProgrammerPublisher;
  83     }
  84
  85     public void setFlowProgrammerPublisher(final IPluginOutFlowProgrammerService flowProgrammerPublisher) {
  86         this.flowProgrammerPublisher = flowProgrammerPublisher;
  87     }
  88
  89     public IClusterGlobalServices getClusterGlobalServices() {
  90         return this.clusterGlobalServices;
  91     }
  92
  93     public void setClusterGlobalServices(final IClusterGlobalServices clusterGlobalServices) {
  94         this.clusterGlobalServices = clusterGlobalServices;
  95     }
  96
  97     @Override
  98     public Status addFlow(final Node node, final Flow flow) {
  99         return toFutureStatus(internalAddFlowAsync(node, flow, 0));
 100     }
 101
 102     @Override
 103     public Status modifyFlow(final Node node, final Flow oldFlow, final Flow newFlow) {
 104         return toFutureStatus(internalModifyFlowAsync(node, oldFlow, newFlow, 0));
 105     }
 106
 107     @Override
 108     public Status removeFlow(final Node node, final Flow flow) {
 109         return toFutureStatus(internalRemoveFlowAsync(node, flow, 0));
 110     }
 111
 112     @Override
 113     public Status addFlowAsync(final Node node, final Flow flow, final long rid) {
 114         // FIXME is this correct? What if the future fails?
 115         this.internalAddFlowAsync(node, flow, rid);
 116         return FlowProgrammerAdapter.toStatus(true);
 117     }
 118
 119     @Override
 120     public Status modifyFlowAsync(final Node node, final Flow oldFlow, final Flow newFlow, final long rid) {
 121         // FIXME is this correct? What if the future fails?
 122         this.internalModifyFlowAsync(node, oldFlow, newFlow, rid);
 123         return FlowProgrammerAdapter.toStatus(true);
 124     }
 125
 126     @Override
 127     public Status removeFlowAsync(final Node node, final Flow flow, final long rid) {
 128         // FIXME is this correct? What if the future fails?
 129         this.internalRemoveFlowAsync(node, flow, rid);
 130         return FlowProgrammerAdapter.toStatus(true);
 131     }
 132
 133     @Override
 134     public Status removeAllFlows(final Node node) {
 135         // FIXME: unfinished?
 136         return new Status(StatusCode.SUCCESS);
 137     }
 138
 139     @Override
 140     public Status syncSendBarrierMessage(final Node node) {
 141         // FIXME: unfinished?
 142         return null;
 143     }
 144
 145     @Override
 146     public Status asyncSendBarrierMessage(final Node node) {
 147         // FIXME: unfinished?
 148         return null;
 149     }
 150
 151     private static Status toStatus(final boolean successful) {
 152         return new Status(successful ? StatusCode.SUCCESS : StatusCode.INTERNALERROR);
 153     }
 154
 155     public static Status toStatus(final RpcResult<? extends Object> result) {
 156         return toStatus(result.isSuccessful());
 157     }
 158
 159     @Override
 160     public void onFlowAdded(final FlowAdded notification) {
 161         // FIXME: unfinished?
 162     }
 163
 164     @Override
 165     public void onFlowRemoved(final FlowRemoved notification) {
 166         if (notification == null) {
 167             return;
 168         }
 169
 170         final NodeRef node = notification.getNode();
 171         if (node == null) {
 172             LOG.debug("Notification {} has not node, ignoring it", notification);
 173             return;
 174         }
 175
 176         Node adNode;
 177         try {
 178             adNode = NodeMapping.toADNode(notification.getNode());
 179         } catch (ConstructionException e) {
 180             LOG.warn("Failed to construct AD node for {}, ignoring notification", node, e);
 181             return;
 182         }
 183         flowProgrammerPublisher.flowRemoved(adNode, ToSalConversionsUtils.toFlow(notification, adNode));
 184     }
 185
 186     @Override
 187     public void onFlowUpdated(final FlowUpdated notification) {
 188         // FIXME: unfinished?
 189     }
 190
 191     @Override
 192     public void onSwitchFlowRemoved(final SwitchFlowRemoved notification) {
 193         // FIXME: unfinished?
 194     }
 195
 196     @Override
 197     public void onNodeErrorNotification(final NodeErrorNotification notification) {
 198         // FIXME: unfinished?
 199     }
 200
 201     @Override
 202     public void onNodeExperimenterErrorNotification(final NodeExperimenterErrorNotification notification) {
 203         // FIXME: unfinished?
 204     }
 205
 206     private static final InstanceIdentifier<org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow> flowPath(
 207             final org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow flow, final NodeKey nodeKey) {
 208         return InstanceIdentifier.builder(Nodes.class)
 209                 .child(org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node.class, nodeKey)
 210                 .augmentation(FlowCapableNode.class)
 211                 .child(Table.class, new TableKey(flow.getTableId()))
 212                 .child(org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow.class, new FlowKey(flow.getId()))
 213                 .toInstance();
 214     }
 215
 216     private Future<RpcResult<TransactionStatus>> writeFlowAsync(final org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow flow, final NodeKey nodeKey) {
 217         final DataModificationTransaction modification = this.dataBrokerService.beginTransaction();
 218         modification.putConfigurationData(flowPath(flow, nodeKey), flow);
 219         return modification.commit();
 220     }
 221
 222     private Future<RpcResult<TransactionStatus>> internalAddFlowAsync(final Node node, final Flow flow, final long rid) {
 223         final Map<Flow,UUID> cache = this.getCache();
 224         UUID flowId = cache.get(flow);
 225         if (flowId != null) {
 226             this.removeFlow(node, flow);
 227         }
 228
 229         flowId = UUID.randomUUID();
 230         cache.put(flow, flowId);
 231         return this.writeFlowAsync(MDFlowMapping.toMDFlow(flow, flowId.toString()), new NodeKey(new NodeId(node.getNodeIDString())));
 232     }
 233
 234     private Future<RpcResult<TransactionStatus>> internalModifyFlowAsync(final Node node, final Flow oldFlow, final Flow newFlow, final long rid) {
 235         final Map<Flow,UUID> cache = this.getCache();
 236
 237         UUID flowId = cache.remove(oldFlow);
 238         if (flowId == null) {
 239             flowId = UUID.randomUUID();
 240             cache.put(oldFlow, flowId);
 241             LOG.warn("Could not find flow {} in cache, assigned new ID {}", oldFlow.hashCode(), flowId);
 242         }
 243
 244         cache.put(newFlow, flowId);
 245         return this.writeFlowAsync(MDFlowMapping.toMDFlow(newFlow, flowId.toString()), new NodeKey(new NodeId(node.getNodeIDString())));
 246     }
 247
 248     private Future<RpcResult<TransactionStatus>> internalRemoveFlowAsync(final Node node, final Flow adflow, final long rid) {
 249         final Map<Flow,UUID> cache = this.getCache();
 250
 251         final UUID flowId = cache.remove(adflow);
 252         if (flowId == null) {
 253             LOG.warn("Could not find flow {} in cache, nothing to do", adflow.hashCode());
 254             return null;
 255         }
 256
 257         final org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow flow = MDFlowMapping.toMDFlow(adflow, flowId.toString());
 258         final DataModificationTransaction modification = this.dataBrokerService.beginTransaction();
 259         modification.removeConfigurationData(flowPath(flow, new NodeKey(new NodeId(node.getNodeIDString()))));
 260         return modification.commit();
 261     }
 262
 263     private static Status toFutureStatus(final Future<RpcResult<TransactionStatus>> future) {
 264         if (future == null) {
 265             // FIXME: really?
 266             return FlowProgrammerAdapter.toStatus(true);
 267         }
 268
 269         try {
 270             final RpcResult<TransactionStatus> result = future.get();
 271             return FlowProgrammerAdapter.toStatus(result);
 272         } catch (final InterruptedException e) {
 273             FlowProgrammerAdapter.LOG.error("Interrupted while processing flow", e);
 274         } catch (ExecutionException e) {
 275             FlowProgrammerAdapter.LOG.error("Failed to process flow", e);
 276         }
 277
 278         return new Status(StatusCode.INTERNALERROR);
 279     }
 280
 281     @SuppressWarnings("unchecked")
 282     private Map<Flow,UUID> getCache() {
 283         final IClusterGlobalServices cgs = getClusterGlobalServices();
 284         if (cgs == null) {
 285             return new ConcurrentHashMap<Flow, UUID>();
 286         }
 287
 288         Map<Flow, UUID> cache = (Map<Flow, UUID>) cgs.getCache(FlowProgrammerAdapter.CACHE_NAME);
 289         if (cache != null) {
 290             return cache;
 291         }
 292
 293         try {
 294             return (Map<Flow, UUID>) cgs.createCache(CACHE_NAME, EnumSet.of(cacheMode.TRANSACTIONAL));
 295         } catch (CacheExistException e) {
 296             return (Map<Flow, UUID>) cgs.getCache(CACHE_NAME);
 297         } catch (CacheConfigException e) {
 298             throw new IllegalStateException("Unexpected cache configuration problem", e);
 299         }
 300     }
 301
 302 }