protected Object[] getImplementations() {
return new Object[] {
dataPacketService,
+ inventory,
};
}
_instanceConfigure((ComponentActivator)imp, c, containerName);
} else if (imp instanceof DataPacketServiceAdapter) {
_instanceConfigure((DataPacketServiceAdapter)imp, c, containerName);
+ } else if (imp instanceof InventoryAndReadAdapter) {
+ _instanceConfigure((InventoryAndReadAdapter)imp, c, containerName);
} else {
throw new IllegalArgumentException(String.format("Unhandled implementation class %s", imp.getClass()));
}
.setRequired(false));
}
+ private void _instanceConfigure(final InventoryAndReadAdapter imp, final Component it, String containerName) {
+ it.setInterface(new String[] {
+ IPluginInInventoryService.class.getName(),
+ IPluginInReadService.class.getName(),
+ }, properties());
+
+ it.add(createServiceDependency()
+ .setService(IPluginOutReadService.class)
+ .setCallbacks("setReadPublisher", "unsetReadPublisher")
+ .setRequired(false));
+ it.add(createServiceDependency()
+ .setService(IPluginOutInventoryService.class)
+ .setCallbacks("setInventoryPublisher", "unsetInventoryPublisher")
+ .setRequired(false));
+ }
+
private void _configure(final TopologyAdapter imp, final Component it) {
it.setInterface(IPluginInTopologyService.class.getName(), properties());
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.AggregateFlowStatisticsUpdate;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.FlowStatisticsData;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.FlowsStatisticsUpdate;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.GetAllFlowsStatisticsFromAllFlowTablesInputBuilder;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.GetFlowStatisticsFromFlowTableInputBuilder;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.OpendaylightFlowStatisticsListener;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.statistics.rev130819.OpendaylightFlowStatisticsService;
* @param id Table id
* @return Table contents, or null if not present
*/
- private Table readConfigTable(final Node node, final short id) {
+ private Table readOperationalTable(final Node node, final short id) {
final InstanceIdentifier<Table> tableRef = InstanceIdentifier.builder(Nodes.class)
- .child(org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node.class, InventoryMapping.toNodeKey(node))
+ .child(org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node.class, NodeMapping.toNodeKey(node))
.augmentation(FlowCapableNode.class)
.child(Table.class, new TableKey(id))
.build();
- return (Table) startChange().readConfigurationData(tableRef);
+ return (Table) startChange().readOperationalData(tableRef);
}
@Override
public List<FlowOnNode> readAllFlow(final Node node, final boolean cached) {
final ArrayList<FlowOnNode> output = new ArrayList<>();
- final Table table = readConfigTable(node, OPENFLOWV10_TABLE_ID);
+ final Table table = readOperationalTable(node, OPENFLOWV10_TABLE_ID);
if (table != null) {
final List<Flow> flows = table.getFlow();
LOG.trace("Number of flows installed in table 0 of node {} : {}", node, flows.size());
}
}
- // TODO (main): Shall we send request to the switch? It will make async request to the switch.
- // Once the plugin receives a response, it will let the adaptor know through onFlowStatisticsUpdate()
- // If we assume that md-sal statistics manager will always be running, then it is not required
- // But if not, then sending request will collect the latest data for adaptor at least.
- getFlowStatisticsService().getAllFlowsStatisticsFromAllFlowTables(
- new GetAllFlowsStatisticsFromAllFlowTablesInputBuilder().setNode(NodeMapping.toNodeRef(node)).build());
return output;
}
@Override
public FlowOnNode readFlow(final Node node, final org.opendaylight.controller.sal.flowprogrammer.Flow targetFlow, final boolean cached) {
FlowOnNode ret = null;
- final Table table = readConfigTable(node, OPENFLOWV10_TABLE_ID);
+ final Table table = readOperationalTable(node, OPENFLOWV10_TABLE_ID);
if (table != null) {
final List<Flow> flows = table.getFlow();
InventoryAndReadAdapter.LOG.trace("Number of flows installed in table 0 of node {} : {}", node, flows.size());
@Override
public NodeTableStatistics readNodeTable(final NodeTable nodeTable, final boolean cached) {
NodeTableStatistics nodeStats = null;
- final Table table = readConfigTable(nodeTable.getNode(), (short) nodeTable.getID());
+ final Table table = readOperationalTable(nodeTable.getNode(), (short) nodeTable.getID());
if (table != null) {
final FlowTableStatisticsData tableStats = table.getAugmentation(FlowTableStatisticsData.class);
if (tableStats != null) {
* @return
*/
private static NodeId toNodeId(org.opendaylight.controller.sal.core.Node aDNode) {
- return new NodeId(aDNode.getType() + ":" + String.valueOf(aDNode.getID()));
+ String targetPrefix = null;
+ if (NodeIDType.OPENFLOW.equals(aDNode.getType())) {
+ targetPrefix = OPENFLOW_ID_PREFIX;
+ } else {
+ targetPrefix = aDNode.getType() + ":";
+ }
+
+ return new NodeId(targetPrefix + String.valueOf(aDNode.getID()));
+ }
+
+ /**
+ * @param aDNode
+ * @return md-sal {@link NodeKey}
+ */
+ public static NodeKey toNodeKey(org.opendaylight.controller.sal.core.Node aDNode) {
+ return new NodeKey(toNodeId(aDNode));
}
public static String toNodeConnectorType(final NodeConnectorId ncId, final NodeId nodeId) {
Assert.assertEquals(0xCC4E241C4A000000L, NodeMapping.openflowFullNodeIdToLong("14721743935839928320").longValue());
}
+ /**
+ * Test method for
+ * {@link org.opendaylight.controller.sal.compatibility.NodeMapping#toNodeKey(org.opendaylight.controller.sal.core.Node)}
+ * .
+ * @throws ConstructionException
+ */
+ @Test
+ public void testToNodeKey() throws ConstructionException {
+ org.opendaylight.controller.sal.core.Node aDNode = new org.opendaylight.controller.sal.core.Node(NodeIDType.OPENFLOW, 42L);
+ NodeKey nodeKey = NodeMapping.toNodeKey(aDNode);
+ Assert.assertEquals("openflow:42", nodeKey.getId().getValue());
+ }
+
/**
* @param nodeId
* @param portId
LOG = context.getLogger();
- if (lastIndex() >= 0) {
- context.setCommitIndex(lastIndex());
- }
-
followers = context.getPeerAddresses().keySet();
for (String followerId : followers) {
FollowerLogInformation followerLogInformation =
new FollowerLogInformationImpl(followerId,
- new AtomicLong(lastIndex()),
+ new AtomicLong(context.getCommitIndex()),
new AtomicLong(-1));
followerToLog.put(followerId, followerLogInformation);
public static class MockPayload extends Payload implements Serializable {
private String value = "";
+ public MockPayload(){
+
+ }
+
public MockPayload(String s) {
this.value = s;
}
return index;
}
}
+
+ public static class MockReplicatedLogBuilder {
+ private ReplicatedLog mockLog = new SimpleReplicatedLog();
+
+ public MockReplicatedLogBuilder createEntries(int start, int end, int term) {
+ for (int i=start; i<end; i++) {
+ this.mockLog.append(new ReplicatedLogImplEntry(i, term, new MockRaftActorContext.MockPayload("foo" + i)));
+ }
+ return this;
+ }
+
+ public MockReplicatedLogBuilder addEntry(int index, int term, MockPayload payload) {
+ this.mockLog.append(new ReplicatedLogImplEntry(index, term, payload));
+ return this;
+ }
+
+ public ReplicatedLog build() {
+ return this.mockLog;
+ }
+ }
}
import akka.actor.ActorRef;
import akka.actor.Props;
import akka.testkit.JavaTestKit;
-import akka.util.Timeout;
import com.google.protobuf.ByteString;
import junit.framework.Assert;
import org.junit.Test;
import org.opendaylight.controller.cluster.raft.messages.RequestVoteReply;
import org.opendaylight.controller.cluster.raft.utils.DoNothingActor;
import org.opendaylight.controller.cluster.raft.utils.MessageCollectorActor;
-import scala.concurrent.Await;
-import scala.concurrent.Future;
-import scala.concurrent.duration.Duration;
-import scala.concurrent.duration.FiniteDuration;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.concurrent.TimeUnit;
-import static akka.pattern.Patterns.ask;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
}
public Object executeLocalOperation(ActorRef actor, Object message) throws Exception {
- FiniteDuration operationDuration = Duration.create(5, TimeUnit.SECONDS);
- Timeout operationTimeout = new Timeout(operationDuration);
- Future<Object> future = ask(actor, message, operationTimeout);
-
- try {
- return Await.result(future, operationDuration);
- } catch (Exception e) {
- throw e;
- }
+ return MessageCollectorActor.getAllMessages(actor);
}
public ByteString getNextChunk (ByteString bs, int offset){
import org.opendaylight.controller.cluster.raft.base.messages.SendHeartBeat;
import org.opendaylight.controller.cluster.raft.base.messages.SendInstallSnapshot;
import org.opendaylight.controller.cluster.raft.messages.AppendEntries;
+import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply;
import org.opendaylight.controller.cluster.raft.messages.InstallSnapshot;
import org.opendaylight.controller.cluster.raft.messages.InstallSnapshotReply;
import org.opendaylight.controller.cluster.raft.utils.DoNothingActor;
+import org.opendaylight.controller.cluster.raft.utils.MessageCollectorActor;
+import org.opendaylight.controller.protobuff.messages.cluster.raft.AppendEntriesMessages;
import org.opendaylight.controller.protobuff.messages.cluster.raft.InstallSnapshotMessages;
import java.io.ByteArrayOutputStream;
actorContext.getReplicatedLog().removeFrom(0);
- actorContext.getReplicatedLog().append(new ReplicatedLogImplEntry(0, 1,
- new MockRaftActorContext.MockPayload("foo")));
-
- ReplicatedLogImplEntry entry =
- new ReplicatedLogImplEntry(1, 1,
- new MockRaftActorContext.MockPayload("foo"));
-
- actorContext.getReplicatedLog().append(entry);
+ actorContext.setReplicatedLog(
+ new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 2, 1)
+ .build());
Leader leader = new Leader(actorContext);
RaftState raftState = leader
- .handleMessage(senderActor, new Replicate(null, "state-id",entry));
+ .handleMessage(senderActor, new Replicate(null, "state-id",actorContext.getReplicatedLog().get(1)));
// State should not change
assertEquals(RaftState.Leader, raftState);
new ReplicatedLogImplEntry(newEntryIndex, currentTerm,
new MockRaftActorContext.MockPayload("D"));
-
RaftState raftState = leader.handleMessage(senderActor, new SendInstallSnapshot());
assertEquals(RaftState.Leader, raftState);
return null;
}
+ public static class ForwardMessageToBehaviorActor extends MessageCollectorActor {
+ private static AbstractRaftActorBehavior behavior;
+
+ public ForwardMessageToBehaviorActor(){
+
+ }
+
+ @Override public void onReceive(Object message) throws Exception {
+ super.onReceive(message);
+ behavior.handleMessage(sender(), message);
+ }
+
+ public static void setBehavior(AbstractRaftActorBehavior behavior){
+ ForwardMessageToBehaviorActor.behavior = behavior;
+ }
+ }
+
+ @Test
+ public void testLeaderCreatedWithCommitIndexLessThanLastIndex() throws Exception {
+ new JavaTestKit(getSystem()) {{
+
+ ActorRef leaderActor = getSystem().actorOf(Props.create(MessageCollectorActor.class));
+
+ MockRaftActorContext leaderActorContext =
+ new MockRaftActorContext("leader", getSystem(), leaderActor);
+
+ ActorRef followerActor = getSystem().actorOf(Props.create(ForwardMessageToBehaviorActor.class));
+
+ MockRaftActorContext followerActorContext =
+ new MockRaftActorContext("follower", getSystem(), followerActor);
+
+ Follower follower = new Follower(followerActorContext);
+
+ ForwardMessageToBehaviorActor.setBehavior(follower);
+
+ Map<String, String> peerAddresses = new HashMap();
+ peerAddresses.put(followerActor.path().toString(),
+ followerActor.path().toString());
+
+ leaderActorContext.setPeerAddresses(peerAddresses);
+
+ leaderActorContext.getReplicatedLog().removeFrom(0);
+
+ //create 3 entries
+ leaderActorContext.setReplicatedLog(
+ new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 3, 1).build());
+
+ leaderActorContext.setCommitIndex(1);
+
+ followerActorContext.getReplicatedLog().removeFrom(0);
+
+ // follower too has the exact same log entries and has the same commit index
+ followerActorContext.setReplicatedLog(
+ new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 3, 1).build());
+
+ followerActorContext.setCommitIndex(1);
+
+ Leader leader = new Leader(leaderActorContext);
+
+ leader.handleMessage(leaderActor, new SendHeartBeat());
+
+ AppendEntriesMessages.AppendEntries appendEntries =
+ (AppendEntriesMessages.AppendEntries) MessageCollectorActor
+ .getFirstMatching(followerActor, AppendEntriesMessages.AppendEntries.class);
+
+ assertNotNull(appendEntries);
+
+ assertEquals(1, appendEntries.getLeaderCommit());
+ assertEquals(1, appendEntries.getLogEntries(0).getIndex());
+ assertEquals(0, appendEntries.getPrevLogIndex());
+
+ AppendEntriesReply appendEntriesReply =
+ (AppendEntriesReply) MessageCollectorActor.getFirstMatching(
+ leaderActor, AppendEntriesReply.class);
+
+ assertNotNull(appendEntriesReply);
+
+ // follower returns its next index
+ assertEquals(2, appendEntriesReply.getLogLastIndex());
+ assertEquals(1, appendEntriesReply.getLogLastTerm());
+
+ }};
+ }
+
+
+ @Test
+ public void testLeaderCreatedWithCommitIndexLessThanFollowersCommitIndex() throws Exception {
+ new JavaTestKit(getSystem()) {{
+
+ ActorRef leaderActor = getSystem().actorOf(Props.create(MessageCollectorActor.class));
+
+ MockRaftActorContext leaderActorContext =
+ new MockRaftActorContext("leader", getSystem(), leaderActor);
+
+ ActorRef followerActor = getSystem().actorOf(
+ Props.create(ForwardMessageToBehaviorActor.class));
+
+ MockRaftActorContext followerActorContext =
+ new MockRaftActorContext("follower", getSystem(), followerActor);
+
+ Follower follower = new Follower(followerActorContext);
+
+ ForwardMessageToBehaviorActor.setBehavior(follower);
+
+ Map<String, String> peerAddresses = new HashMap();
+ peerAddresses.put(followerActor.path().toString(),
+ followerActor.path().toString());
+
+ leaderActorContext.setPeerAddresses(peerAddresses);
+
+ leaderActorContext.getReplicatedLog().removeFrom(0);
+
+ leaderActorContext.setReplicatedLog(
+ new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 3, 1).build());
+
+ leaderActorContext.setCommitIndex(1);
+
+ followerActorContext.getReplicatedLog().removeFrom(0);
+
+ followerActorContext.setReplicatedLog(
+ new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 3, 1).build());
+
+ // follower has the same log entries but its commit index > leaders commit index
+ followerActorContext.setCommitIndex(2);
+
+ Leader leader = new Leader(leaderActorContext);
+
+ leader.handleMessage(leaderActor, new SendHeartBeat());
+
+ AppendEntriesMessages.AppendEntries appendEntries =
+ (AppendEntriesMessages.AppendEntries) MessageCollectorActor
+ .getFirstMatching(followerActor, AppendEntriesMessages.AppendEntries.class);
+
+ assertNotNull(appendEntries);
+
+ assertEquals(1, appendEntries.getLeaderCommit());
+ assertEquals(1, appendEntries.getLogEntries(0).getIndex());
+ assertEquals(0, appendEntries.getPrevLogIndex());
+
+ AppendEntriesReply appendEntriesReply =
+ (AppendEntriesReply) MessageCollectorActor.getFirstMatching(
+ leaderActor, AppendEntriesReply.class);
+
+ assertNotNull(appendEntriesReply);
+
+ assertEquals(2, appendEntriesReply.getLogLastIndex());
+ assertEquals(1, appendEntriesReply.getLogLastTerm());
+
+ }};
+ }
+
private static class LeaderTestKit extends JavaTestKit {
private LeaderTestKit(ActorSystem actorSystem) {
package org.opendaylight.controller.cluster.raft.utils;
+import akka.actor.ActorRef;
import akka.actor.UntypedActor;
+import akka.pattern.Patterns;
+import akka.util.Timeout;
+import scala.concurrent.Await;
+import scala.concurrent.Future;
+import scala.concurrent.duration.Duration;
+import scala.concurrent.duration.FiniteDuration;
import java.util.ArrayList;
import java.util.List;
+import java.util.concurrent.TimeUnit;
public class MessageCollectorActor extends UntypedActor {
messages.add(message);
}
}
+
+ public static List<Object> getAllMessages(ActorRef actor) throws Exception {
+ FiniteDuration operationDuration = Duration.create(5, TimeUnit.SECONDS);
+ Timeout operationTimeout = new Timeout(operationDuration);
+ Future<Object> future = Patterns.ask(actor, "get-all-messages", operationTimeout);
+
+ try {
+ return (List<Object>) Await.result(future, operationDuration);
+ } catch (Exception e) {
+ throw e;
+ }
+ }
+
+ /**
+ * Get the first message that matches the specified class
+ * @param actor
+ * @param clazz
+ * @return
+ */
+ public static Object getFirstMatching(ActorRef actor, Class clazz) throws Exception {
+ List<Object> allMessages = getAllMessages(actor);
+
+ for(Object message : allMessages){
+ if(message.getClass().equals(clazz)){
+ return message;
+ }
+ }
+
+ return null;
+ }
+
}
--- /dev/null
+<?xml version="1.0" encoding="UTF-8"?>
+
+<blueprint xmlns="http://www.osgi.org/xmlns/blueprint/v1.0.0"
+ xmlns:cm="http://aries.apache.org/blueprint/xmlns/blueprint-cm/v1.0.0">
+
+ <command-bundle xmlns="http://karaf.apache.org/xmlns/shell/v1.1.0">
+ <command>
+ <action class="org.opendaylight.controller.xsql.xsql">
+ </action>
+ </command>
+ </command-bundle>
+</blueprint>
org.eclipse.persistence.jaxb.rs,
com.sun.jersey.spi.container.servlet,
javax.ws.rs,
+ javax.ws.rs.ext,
javax.ws.rs.core,
javax.xml.bind.annotation,
javax.xml.bind,
import org.codehaus.enunciate.jaxrs.StatusCodes;
import org.opendaylight.controller.networkconfig.neutron.INeutronLoadBalancerPoolAware;
import org.opendaylight.controller.networkconfig.neutron.INeutronLoadBalancerPoolCRUD;
-import org.opendaylight.controller.networkconfig.neutron.INeutronLoadBalancerPoolMemberCRUD;
import org.opendaylight.controller.networkconfig.neutron.NeutronCRUDInterfaces;
import org.opendaylight.controller.networkconfig.neutron.NeutronLoadBalancerPool;
import org.opendaylight.controller.networkconfig.neutron.NeutronLoadBalancerPoolMember;
/**
* For now, the LB pool member data is maintained with the INeutronLoadBalancerPoolCRUD,
- * although there may be an overlap with INeutronLoadBalancerPoolMemberCRUD's cache.
- * TODO: Consolidate and maintain a single copy
+ * and not duplicated within the INeutronLoadBalancerPoolMemberCRUD's cache.
*/
@Path("/pools")
service.neutronLoadBalancerPoolDeleted(singleton);
}
}
-
- /*
- * remove corresponding members from the member cache too
- */
- INeutronLoadBalancerPoolMemberCRUD loadBalancerPoolMemberInterface = NeutronCRUDInterfaces.getINeutronLoadBalancerPoolMemberCRUD(this);
- if (loadBalancerPoolMemberInterface != null) {
- List<NeutronLoadBalancerPoolMember> allLoadBalancerPoolMembers = new
- ArrayList<NeutronLoadBalancerPoolMember>(loadBalancerPoolMemberInterface.getAllNeutronLoadBalancerPoolMembers());
- Iterator<NeutronLoadBalancerPoolMember> i = allLoadBalancerPoolMembers.iterator();
- while (i.hasNext()) {
- NeutronLoadBalancerPoolMember member = i.next();
- if (member.getPoolID() == loadBalancerPoolUUID)
- loadBalancerPoolMemberInterface.removeNeutronLoadBalancerPoolMember(member.getPoolMemberID());
- }
- }
return Response.status(204).build();
}
}
Code Review / controller.git / commitdiff