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         // notified upon remove flow rpc successfully invoked
 167         if (notification == null) {
 168             return;
 169         }
 170
 171         final NodeRef node = notification.getNode();
 172         if (node == null) {
 173             LOG.debug("Notification {} has not node, ignoring it", notification);
 174             return;
 175         }
 176
 177         Node adNode;
 178         try {
 179             adNode = NodeMapping.toADNode(notification.getNode());
 180         } catch (ConstructionException e) {
 181             LOG.warn("Failed to construct AD node for {}, ignoring notification", node, e);
 182             return;
 183         }
 184         flowProgrammerPublisher.flowRemoved(adNode, ToSalConversionsUtils.toFlow(notification, adNode));
 185     }
 186
 187     @Override
 188     public void onFlowUpdated(final FlowUpdated notification) {
 189         // FIXME: unfinished?
 190     }
 191
 192     @Override
 193     public void onSwitchFlowRemoved(final SwitchFlowRemoved notification) {
 194         // notified upon remove flow message from device arrives
 195         if (notification == null) {
 196             return;
 197         }
 198
 199         final NodeRef node = notification.getNode();
 200         if (node == null) {
 201             LOG.debug("Notification {} has not node, ignoring it", notification);
 202             return;
 203         }
 204
 205         Node adNode;
 206         try {
 207             adNode = NodeMapping.toADNode(notification.getNode());
 208         } catch (ConstructionException e) {
 209             LOG.warn("Failed to construct AD node for {}, ignoring notification", node, e);
 210             return;
 211         }
 212         flowProgrammerPublisher.flowRemoved(adNode, ToSalConversionsUtils.toFlow(notification, adNode));
 213     }
 214
 215     @Override
 216     public void onNodeErrorNotification(final NodeErrorNotification notification) {
 217         // FIXME: unfinished?
 218     }
 219
 220     @Override
 221     public void onNodeExperimenterErrorNotification(final NodeExperimenterErrorNotification notification) {
 222         // FIXME: unfinished?
 223     }
 224
 225     private static final InstanceIdentifier<org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow> flowPath(
 226             final org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow flow, final NodeKey nodeKey) {
 227         return InstanceIdentifier.builder(Nodes.class)
 228                 .child(org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node.class, nodeKey)
 229                 .augmentation(FlowCapableNode.class)
 230                 .child(Table.class, new TableKey(flow.getTableId()))
 231                 .child(org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow.class, new FlowKey(flow.getId()))
 232                 .toInstance();
 233     }
 234
 235     private Future<RpcResult<TransactionStatus>> writeFlowAsync(final org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow flow, final NodeKey nodeKey) {
 236         final DataModificationTransaction modification = this.dataBrokerService.beginTransaction();
 237         modification.putConfigurationData(flowPath(flow, nodeKey), flow);
 238         return modification.commit();
 239     }
 240
 241     private Future<RpcResult<TransactionStatus>> internalAddFlowAsync(final Node node, final Flow flow, final long rid) {
 242         final Map<Flow,UUID> cache = this.getCache();
 243         UUID flowId = cache.get(flow);
 244         if (flowId != null) {
 245             this.removeFlow(node, flow);
 246         }
 247
 248         flowId = UUID.randomUUID();
 249         cache.put(flow, flowId);
 250         return this.writeFlowAsync(MDFlowMapping.toMDFlow(flow, flowId.toString()), new NodeKey(
 251                 new NodeId(NodeMapping.OPENFLOW_ID_PREFIX + node.getID())));
 252     }
 253
 254     private Future<RpcResult<TransactionStatus>> internalModifyFlowAsync(final Node node, final Flow oldFlow, final Flow newFlow, final long rid) {
 255         final Map<Flow,UUID> cache = this.getCache();
 256
 257         UUID flowId = cache.remove(oldFlow);
 258         if (flowId == null) {
 259             flowId = UUID.randomUUID();
 260             cache.put(oldFlow, flowId);
 261             LOG.warn("Could not find flow {} in cache, assigned new ID {}", oldFlow.hashCode(), flowId);
 262         }
 263
 264         cache.put(newFlow, flowId);
 265         return this.writeFlowAsync(MDFlowMapping.toMDFlow(newFlow, flowId.toString()), new NodeKey(
 266                 new NodeId(NodeMapping.OPENFLOW_ID_PREFIX + node.getID())));
 267     }
 268
 269     private Future<RpcResult<TransactionStatus>> internalRemoveFlowAsync(final Node node, final Flow adflow, final long rid) {
 270         final Map<Flow,UUID> cache = this.getCache();
 271
 272         final UUID flowId = cache.remove(adflow);
 273         if (flowId == null) {
 274             LOG.warn("Could not find flow {} in cache, nothing to do", adflow.hashCode());
 275             return null;
 276         }
 277
 278         final org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow flow = MDFlowMapping.toMDFlow(adflow, flowId.toString());
 279         final DataModificationTransaction modification = this.dataBrokerService.beginTransaction();
 280         modification.removeConfigurationData(flowPath(flow, new NodeKey(
 281                 new NodeId(NodeMapping.OPENFLOW_ID_PREFIX + node.getID()))));
 282         return modification.commit();
 283     }
 284
 285     private static Status toFutureStatus(final Future<RpcResult<TransactionStatus>> future) {
 286         if (future == null) {
 287             // FIXME: really?
 288             return FlowProgrammerAdapter.toStatus(true);
 289         }
 290
 291         try {
 292             final RpcResult<TransactionStatus> result = future.get();
 293             return FlowProgrammerAdapter.toStatus(result);
 294         } catch (final InterruptedException e) {
 295             FlowProgrammerAdapter.LOG.error("Interrupted while processing flow", e);
 296         } catch (ExecutionException e) {
 297             FlowProgrammerAdapter.LOG.error("Failed to process flow", e);
 298         }
 299
 300         return new Status(StatusCode.INTERNALERROR);
 301     }
 302
 303     @SuppressWarnings("unchecked")
 304     private Map<Flow,UUID> getCache() {
 305         final IClusterGlobalServices cgs = getClusterGlobalServices();
 306         if (cgs == null) {
 307             return new ConcurrentHashMap<Flow, UUID>();
 308         }
 309
 310         Map<Flow, UUID> cache = (Map<Flow, UUID>) cgs.getCache(FlowProgrammerAdapter.CACHE_NAME);
 311         if (cache != null) {
 312             return cache;
 313         }
 314
 315         try {
 316             return (Map<Flow, UUID>) cgs.createCache(CACHE_NAME, EnumSet.of(cacheMode.TRANSACTIONAL));
 317         } catch (CacheExistException e) {
 318             return (Map<Flow, UUID>) cgs.getCache(CACHE_NAME);
 319         } catch (CacheConfigException e) {
 320             throw new IllegalStateException("Unexpected cache configuration problem", e);
 321         }
 322     }
 323
 324 }