*/
package org.opendaylight.ovsdb.openstack.netvirt.providers.openflow13;
-import java.util.List;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.LinkedBlockingDeque;
-
-import org.opendaylight.controller.md.sal.binding.api.BindingTransactionChain;
import org.opendaylight.controller.md.sal.binding.api.DataBroker;
-import org.opendaylight.controller.md.sal.binding.api.ReadWriteTransaction;
+import org.opendaylight.controller.md.sal.binding.api.ReadOnlyTransaction;
import org.opendaylight.controller.md.sal.binding.api.WriteTransaction;
-import org.opendaylight.controller.md.sal.common.api.data.AsyncTransaction;
import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
-import org.opendaylight.controller.md.sal.common.api.data.TransactionChain;
-import org.opendaylight.controller.md.sal.common.api.data.TransactionChainListener;
import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
-import org.opendaylight.controller.sal.binding.api.NotificationProviderService;
+import org.opendaylight.ovsdb.openstack.netvirt.api.Southbound;
+import org.opendaylight.ovsdb.openstack.netvirt.api.Constants;
+import org.opendaylight.ovsdb.openstack.netvirt.providers.NetvirtProvidersProvider;
import org.opendaylight.ovsdb.utils.mdsal.openflow.InstructionUtils;
+import org.opendaylight.ovsdb.utils.servicehelper.ServiceHelper;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowId;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.Table;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.rev131026.instruction.list.Instruction;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.rev131026.instruction.list.InstructionBuilder;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.rev131026.instruction.list.InstructionKey;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeConnectorRemoved;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeConnectorUpdated;
import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRef;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRemoved;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeUpdated;
import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.Nodes;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.OpendaylightInventoryListener;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.NodeBuilder;
import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.NodeKey;
-import org.opendaylight.yangtools.yang.binding.DataObject;
+import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.Node;
import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
-import org.opendaylight.yangtools.yang.binding.InstanceIdentifier.InstanceIdentifierBuilder;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import com.google.common.base.Optional;
-import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import com.google.common.util.concurrent.CheckedFuture;
-import com.google.common.util.concurrent.FutureCallback;
-import com.google.common.util.concurrent.Futures;
+
+import java.util.List;
+import java.util.concurrent.ExecutionException;
+
+import org.osgi.framework.ServiceReference;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* Any ServiceInstance class that extends AbstractServiceInstance to be a part of the pipeline
* use it in any matching flows that needs to be further processed by next service in the pipeline.
*
*/
-public abstract class AbstractServiceInstance implements OpendaylightInventoryListener, Runnable, TransactionChainListener {
+public abstract class AbstractServiceInstance {
public static final String SERVICE_PROPERTY ="serviceProperty";
- private static final Logger logger = LoggerFactory.getLogger(AbstractServiceInstance.class);
-
+ private static final Logger LOG = LoggerFactory.getLogger(AbstractServiceInstance.class);
+ public static final String OPENFLOW = "openflow:";
+ private DataBroker dataBroker = null;
// OSGi Services that we are dependent on.
- private volatile MdsalConsumer mdsalConsumer;
private volatile PipelineOrchestrator orchestrator;
+ private volatile Southbound southbound;
- // Concrete Service that this AbstractServiceInstance represent
+ // Concrete Service that this AbstractServiceInstance represents
private Service service;
- private BindingTransactionChain txChain;
-
- // Process Notification in its own thread
- Thread thread = null;
- private final BlockingQueue<String> queue = new LinkedBlockingDeque<>();
-
public AbstractServiceInstance (Service service) {
this.service = service;
+ this.dataBroker = NetvirtProvidersProvider.getDataBroker();
+ }
+
+ protected void setDependencies(final ServiceReference ref, AbstractServiceInstance serviceInstance) {
+ this.orchestrator =
+ (PipelineOrchestrator) ServiceHelper.getGlobalInstance(PipelineOrchestrator.class, serviceInstance);
+ orchestrator.registerService(ref, serviceInstance);
+ this.southbound =
+ (Southbound) ServiceHelper.getGlobalInstance(Southbound.class, serviceInstance);
}
- // Let the Concrete service instance class decide if a Bride is part of the pipeline or not.
- public abstract boolean isBridgeInPipeline (String nodeId);
+ public boolean isBridgeInPipeline (Node node){
+ String bridgeName = southbound.getBridgeName(node);
+ return bridgeName != null && Constants.INTEGRATION_BRIDGE.equals(bridgeName);
+ }
public short getTable() {
return service.getTable();
this.service = service;
}
- public void start() {
- // Register for OpenFlow bridge/node Creation notification.
- NotificationProviderService notificationService = mdsalConsumer.getNotificationService();
- if (notificationService != null) {
- notificationService.registerNotificationListener(this);
- }
- this.txChain = mdsalConsumer.getDataBroker().createTransactionChain(this);
-
- // Never block a Notification thread. Process the notification in its own Thread.
- thread = new Thread(this);
- thread.setDaemon(true);
- thread.setName("AbstractServiceInstance-"+service.toString());
- thread.start();
- }
-
- private NodeBuilder createNodeBuilder(String nodeId) {
+ public NodeBuilder createNodeBuilder(String nodeId) {
NodeBuilder builder = new NodeBuilder();
builder.setId(new NodeId(nodeId));
builder.setKey(new NodeKey(builder.getId()));
return builder;
}
+ private static InstanceIdentifier<Flow> createFlowPath(FlowBuilder flowBuilder, NodeBuilder nodeBuilder) {
+ return InstanceIdentifier.builder(Nodes.class)
+ .child(org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node.class,
+ nodeBuilder.getKey())
+ .augmentation(FlowCapableNode.class)
+ .child(Table.class, new TableKey(flowBuilder.getTableId()))
+ .child(Flow.class, flowBuilder.getKey()).build();
+ }
+
+ private static InstanceIdentifier<org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node>
+ createNodePath(NodeBuilder nodeBuilder) {
+ return InstanceIdentifier.builder(Nodes.class)
+ .child(org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node.class,
+ nodeBuilder.getKey()).build();
+ }
+
/**
- * This method returns the required Pipeline Instructions to by used by any matching flows that needs
+ * This method returns the required Pipeline Instructions to by used by any matching flows that need
* to be further processed by next service in the pipeline.
*
* Important to note that this is a convenience method which returns a mutable instructionBuilder which
}
protected void writeFlow(FlowBuilder flowBuilder, NodeBuilder nodeBuilder) {
- Preconditions.checkNotNull(mdsalConsumer);
- if (mdsalConsumer == null) {
- logger.error("ERROR finding MDSAL Service. Its possible that writeFlow is called too soon ?");
- return;
- }
+ LOG.debug("writeFlow: flowBuilder: {}, nodeBuilder: {}",
+ flowBuilder.build(), nodeBuilder.build());
+ WriteTransaction modification = dataBroker.newWriteOnlyTransaction();
+ modification.put(LogicalDatastoreType.CONFIGURATION, createNodePath(nodeBuilder),
+ nodeBuilder.build(), true /*createMissingParents*/);
+ modification.put(LogicalDatastoreType.CONFIGURATION, createFlowPath(flowBuilder, nodeBuilder),
+ flowBuilder.build(), true /*createMissingParents*/);
- DataBroker dataBroker = mdsalConsumer.getDataBroker();
- if (dataBroker == null) {
- logger.error("ERROR finding reference for DataBroker. Please check MD-SAL support on the Controller.");
- return;
+ CheckedFuture<Void, TransactionCommitFailedException> commitFuture = modification.submit();
+ try {
+ commitFuture.get(); // TODO: Make it async (See bug 1362)
+ LOG.debug("Transaction success for write of Flow {}", flowBuilder.getFlowName());
+ } catch (Exception e) {
+ LOG.error(e.getMessage(), e);
+ modification.cancel();
}
+ }
- ReadWriteTransaction modification = dataBroker.newReadWriteTransaction();
- InstanceIdentifier<Flow> path1 = InstanceIdentifier.builder(Nodes.class).child(org.opendaylight.yang.gen.v1.urn.opendaylight.inventory
- .rev130819.nodes.Node.class, nodeBuilder.getKey()).augmentation(FlowCapableNode.class).child(Table.class,
- new TableKey(flowBuilder.getTableId())).child(Flow.class, flowBuilder.getKey()).build();
-
- //modification.put(LogicalDatastoreType.OPERATIONAL, path1, flowBuilder.build());
- modification.put(LogicalDatastoreType.CONFIGURATION, path1, flowBuilder.build(), true /*createMissingParents*/);
-
+ protected void removeFlow(FlowBuilder flowBuilder, NodeBuilder nodeBuilder) {
+ WriteTransaction modification = dataBroker.newWriteOnlyTransaction();
+ modification.delete(LogicalDatastoreType.CONFIGURATION, createFlowPath(flowBuilder, nodeBuilder));
CheckedFuture<Void, TransactionCommitFailedException> commitFuture = modification.submit();
try {
commitFuture.get(); // TODO: Make it async (See bug 1362)
- logger.debug("Transaction success for write of Flow "+flowBuilder.getFlowName());
- } catch (InterruptedException|ExecutionException e) {
- logger.error(e.getMessage(), e);
-
+ LOG.debug("Transaction success for deletion of Flow {}", flowBuilder.getFlowName());
+ } catch (Exception e) {
+ LOG.error(e.getMessage(), e);
+ modification.cancel();
}
}
- protected void removeFlow(FlowBuilder flowBuilder, NodeBuilder nodeBuilder) {
- Preconditions.checkNotNull(mdsalConsumer);
- if (mdsalConsumer == null) {
- logger.error("ERROR finding MDSAL Service.");
- return;
+ public Flow getFlow(FlowBuilder flowBuilder, NodeBuilder nodeBuilder) {
+ ReadOnlyTransaction readTx = dataBroker.newReadOnlyTransaction();
+ try {
+ Optional<Flow> data =
+ readTx.read(LogicalDatastoreType.CONFIGURATION, createFlowPath(flowBuilder, nodeBuilder)).get();
+ if (data.isPresent()) {
+ return data.get();
+ }
+ } catch (InterruptedException|ExecutionException e) {
+ LOG.error(e.getMessage(), e);
}
- DataBroker dataBroker = mdsalConsumer.getDataBroker();
- if (dataBroker == null) {
- logger.error("ERROR finding reference for DataBroker. Please check MD-SAL support on the Controller.");
- return;
- }
+ LOG.debug("Cannot find data for Flow {}", flowBuilder.getFlowName());
+ return null;
+ }
- WriteTransaction modification = dataBroker.newWriteOnlyTransaction();
- InstanceIdentifier<Flow> path1 = InstanceIdentifier.builder(Nodes.class)
- .child(org.opendaylight.yang.gen.v1.urn.opendaylight.inventory
- .rev130819.nodes.Node.class, nodeBuilder.getKey())
- .augmentation(FlowCapableNode.class).child(Table.class,
- new TableKey(flowBuilder.getTableId())).child(Flow.class, flowBuilder.getKey()).build();
- //modification.delete(LogicalDatastoreType.OPERATIONAL, nodeBuilderToInstanceId(nodeBuilder));
- //modification.delete(LogicalDatastoreType.OPERATIONAL, path1);
- //modification.delete(LogicalDatastoreType.CONFIGURATION, nodeBuilderToInstanceId(nodeBuilder));
- modification.delete(LogicalDatastoreType.CONFIGURATION, path1);
+ public org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node
+ getOpenFlowNode(String nodeId) {
- CheckedFuture<Void, TransactionCommitFailedException> commitFuture = modification.submit();
+ ReadOnlyTransaction readTx = dataBroker.newReadOnlyTransaction();
try {
- commitFuture.get(); // TODO: Make it async (See bug 1362)
- logger.debug("Transaction success for deletion of Flow "+flowBuilder.getFlowName());
+ Optional<org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node> data =
+ readTx.read(LogicalDatastoreType.OPERATIONAL, createNodePath(createNodeBuilder(nodeId))).get();
+ if (data.isPresent()) {
+ return data.get();
+ }
} catch (InterruptedException|ExecutionException e) {
- logger.error(e.getMessage(), e);
+ LOG.error(e.getMessage(), e);
}
+
+ LOG.debug("Cannot find data for Node {}", nodeId);
+ return null;
+ }
+
+ private long getDpid(Node node) {
+ long dpid = southbound.getDataPathId(node);
+ if (dpid == 0) {
+ LOG.warn("getDpid: dpid not found: {}", node);
+ }
+ return dpid;
}
/**
* Program Default Pipeline Flow.
*
- * @param nodeId Node on which the default pipeline flow is programmed.
+ * @param node on which the default pipeline flow is programmed.
*/
- protected void programDefaultPipelineRule(String nodeId) {
+ protected void programDefaultPipelineRule(Node node) {
+ if (!isBridgeInPipeline(node)) {
+ //LOG.trace("Bridge is not in pipeline {} ", node);
+ return;
+ }
MatchBuilder matchBuilder = new MatchBuilder();
FlowBuilder flowBuilder = new FlowBuilder();
- NodeBuilder nodeBuilder = createNodeBuilder(nodeId);
+ long dpid = getDpid(node);
+ if (dpid == 0L) {
+ LOG.info("could not find dpid: {}", node.getNodeId());
+ return;
+ }
+ String nodeName = OPENFLOW + getDpid(node);
+ NodeBuilder nodeBuilder = createNodeBuilder(nodeName);
// Create the OF Actions and Instructions
InstructionsBuilder isb = new InstructionsBuilder();
// Add InstructionsBuilder to FlowBuilder
flowBuilder.setInstructions(isb.build());
- String flowId = "DEFAULT_PIPELINE_FLOW";
+ String flowId = "DEFAULT_PIPELINE_FLOW_"+service.getTable();
flowBuilder.setId(new FlowId(flowId));
FlowKey key = new FlowKey(new FlowId(flowId));
flowBuilder.setMatch(matchBuilder.build());
flowBuilder.setIdleTimeout(0);
writeFlow(flowBuilder, nodeBuilder);
}
-
- @Override
- public void onNodeConnectorRemoved(NodeConnectorRemoved nodeConector) {
- }
-
- @Override
- public void onNodeConnectorUpdated(NodeConnectorUpdated nodeConnector) {
- }
-
- @Override
- public void onNodeRemoved(NodeRemoved node) {
- }
-
-
- @Override
- public void run() {
- try {
- for (; ; ) {
- String nodeId = queue.take();
- this.programDefaultPipelineRule(nodeId);
- }
- } catch (InterruptedException e) {
- logger.warn("Processing interrupted, terminating", e);
- }
-
- while (!queue.isEmpty()) {
- queue.poll();
- }
-
- }
-
- void enqueue(final String nodeId) {
- try {
- queue.put(nodeId);
- } catch (InterruptedException e) {
- logger.warn("Failed to enqueue operation {}", nodeId, e);
- }
- }
-
- /**
- * Process the Node update notification. Check for Openflow node and make sure if the bridge is part of the Pipeline before
- * programming the Pipeline specific flows.
- */
- @Override
- public void onNodeUpdated(NodeUpdated nodeUpdated) {
- NodeRef ref = nodeUpdated.getNodeRef();
- InstanceIdentifier<Node> identifier = (InstanceIdentifier<Node>) ref.getValue();
- logger.info("GOT NOTIFICATION FOR "+identifier.toString());
- final NodeKey key = identifier.firstKeyOf(Node.class, NodeKey.class);
- final String nodeId = key.getId().getValue();
- if (!this.isBridgeInPipeline(nodeId)) {
- logger.debug("Bridge {} is not in pipeline", nodeId);
- return;
- }
- if (key != null && key.getId().getValue().contains("openflow")) {
- InstanceIdentifierBuilder<Node> builder = ((InstanceIdentifier<Node>) ref.getValue()).builder();
- InstanceIdentifierBuilder<FlowCapableNode> augmentation = builder.augmentation(FlowCapableNode.class);
- final InstanceIdentifier<FlowCapableNode> path = augmentation.build();
- CheckedFuture readFuture = txChain.newReadWriteTransaction().read(LogicalDatastoreType.OPERATIONAL, path);
- Futures.addCallback(readFuture, new FutureCallback<Optional<? extends DataObject>>() {
- @Override
- public void onSuccess(Optional<? extends DataObject> optional) {
- if (!optional.isPresent()) {
- enqueue(nodeId);
- }
- }
-
- @Override
- public void onFailure(Throwable throwable) {
- logger.debug(String.format("Can't retrieve node data for node %s. Writing node data with table0.", nodeId));
- enqueue(nodeId);
- }
- });
- }
- }
-
- @Override
- public void onTransactionChainFailed(final TransactionChain<?, ?> chain, final AsyncTransaction<?, ?> transaction,
- final Throwable cause) {
- logger.error("Failed to export Flow Capable Inventory, Transaction {} failed.",transaction.getIdentifier(),cause);
- }
-
- @Override
- public void onTransactionChainSuccessful(final TransactionChain<?, ?> chain) {
- }
}
Code Review / ovsdb.git / blobdiff