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.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.utils.mdsal.openflow.InstructionUtils;
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.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.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;
/**
* 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 {
+public abstract class AbstractServiceInstance implements OpendaylightInventoryListener, Runnable, TransactionChainListener {
public static final String SERVICE_PROPERTY ="serviceProperty";
- private Service service;
+ private static final Logger logger = LoggerFactory.getLogger(AbstractServiceInstance.class);
+
+ // OSGi Services that we are dependent on.
+ private volatile MdsalConsumer mdsalConsumer;
private volatile PipelineOrchestrator orchestrator;
+ // Concrete Service that this AbstractServiceInstance represent
+ 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;
}
+ // Let the Concrete service instance class decide if a Bride is part of the pipeline or not.
+ public abstract boolean isBridgeInPipeline (String nodeId);
+
public int 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) {
NodeBuilder builder = new NodeBuilder();
builder.setId(new NodeId(nodeId));
}
}
- private static final Logger logger = LoggerFactory.getLogger(AbstractServiceInstance.class);
- private volatile MdsalConsumer mdsalConsumer;
-
- public void writeFlow(FlowBuilder flowBuilder, NodeBuilder nodeBuilder) {
+ private 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 ?");
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) {
+ }
}
package org.opendaylight.ovsdb.openstack.netvirt.providers.openflow13;
+import java.util.Collection;
+
+import org.apache.felix.dm.Component;
import org.opendaylight.controller.md.sal.binding.api.DataBroker;
import org.opendaylight.controller.sal.binding.api.BindingAwareBroker;
import org.opendaylight.controller.sal.binding.api.BindingAwareBroker.ConsumerContext;
+import org.opendaylight.controller.sal.binding.api.BindingAwareBroker.ProviderContext;
import org.opendaylight.controller.sal.binding.api.BindingAwareConsumer;
-
-import org.apache.felix.dm.Component;
+import org.opendaylight.controller.sal.binding.api.BindingAwareProvider;
+import org.opendaylight.controller.sal.binding.api.NotificationProviderService;
+import org.opendaylight.yangtools.yang.binding.RpcService;
import org.osgi.framework.BundleContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-public class MdsalConsumerImpl implements BindingAwareConsumer, MdsalConsumer {
+public class MdsalConsumerImpl implements BindingAwareConsumer, MdsalConsumer, BindingAwareProvider {
private BundleContext ctx = null;
private volatile BindingAwareBroker broker;
private ConsumerContext consumerContext = null;
private DataBroker dataBroker;
+ private NotificationProviderService notificationService;
static final Logger logger = LoggerFactory.getLogger(MdsalConsumerImpl.class);
this.ctx = c.getDependencyManager().getBundleContext();
logger.info("Open vSwitch OpenFlow 1.3 Neutron Networking Provider Registered with MD-SAL");
broker.registerConsumer(this, this.ctx);
+ broker.registerProvider(this, this.ctx);
}
void destroy() {
public DataBroker getDataBroker() {
return dataBroker;
}
+ @Override
+ public NotificationProviderService getNotificationService() {
+ return notificationService;
+ }
+
+ @Override
+ public Collection<? extends ProviderFunctionality> getFunctionality() {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ @Override
+ public Collection<? extends RpcService> getImplementations() {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+ @Override
+ public void onSessionInitiated(ProviderContext session) {
+ notificationService = session.getSALService(NotificationProviderService.class);
+ }
}
Code Review / ovsdb.git / commitdiff