import com.google.common.collect.Sets;
import java.io.Serializable;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.opendaylight.controller.cluster.common.actor.AbstractUntypedPersistentActorWithMetering;
+import org.opendaylight.controller.cluster.datastore.config.Configuration;
+import org.opendaylight.controller.cluster.datastore.config.ModuleShardConfiguration;
import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException;
import org.opendaylight.controller.cluster.datastore.exceptions.NotInitializedException;
import org.opendaylight.controller.cluster.datastore.exceptions.PrimaryNotFoundException;
import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shardmanager.ShardManagerInfo;
import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shardmanager.ShardManagerInfoMBean;
import org.opendaylight.controller.cluster.datastore.messages.ActorInitialized;
+import org.opendaylight.controller.cluster.datastore.messages.CreateShard;
+import org.opendaylight.controller.cluster.datastore.messages.CreateShardReply;
import org.opendaylight.controller.cluster.datastore.messages.FindLocalShard;
import org.opendaylight.controller.cluster.datastore.messages.FindPrimary;
import org.opendaylight.controller.cluster.datastore.messages.LocalPrimaryShardFound;
import org.opendaylight.controller.cluster.datastore.messages.LocalShardFound;
import org.opendaylight.controller.cluster.datastore.messages.LocalShardNotFound;
import org.opendaylight.controller.cluster.datastore.messages.PeerAddressResolved;
+import org.opendaylight.controller.cluster.datastore.messages.PeerDown;
+import org.opendaylight.controller.cluster.datastore.messages.PeerUp;
import org.opendaylight.controller.cluster.datastore.messages.RemoteFindPrimary;
import org.opendaylight.controller.cluster.datastore.messages.RemotePrimaryShardFound;
import org.opendaylight.controller.cluster.datastore.messages.ShardLeaderStateChanged;
private final PrimaryShardInfoFutureCache primaryShardInfoCache;
+ private SchemaContext schemaContext;
+
/**
*/
protected ShardManager(ClusterWrapper cluster, Configuration configuration,
onActorInitialized(message);
} else if (message instanceof ClusterEvent.MemberUp){
memberUp((ClusterEvent.MemberUp) message);
+ } else if (message instanceof ClusterEvent.MemberExited){
+ memberExited((ClusterEvent.MemberExited) message);
} else if(message instanceof ClusterEvent.MemberRemoved) {
memberRemoved((ClusterEvent.MemberRemoved) message);
} else if(message instanceof ClusterEvent.UnreachableMember) {
onLeaderStateChanged((ShardLeaderStateChanged) message);
} else if(message instanceof SwitchShardBehavior){
onSwitchShardBehavior((SwitchShardBehavior) message);
+ } else if(message instanceof CreateShard) {
+ onCreateShard((CreateShard)message);
} else {
unknownMessage(message);
}
}
+ private void onCreateShard(CreateShard createShard) {
+ Object reply;
+ try {
+ ModuleShardConfiguration moduleShardConfig = createShard.getModuleShardConfig();
+ if(localShards.containsKey(moduleShardConfig.getShardName())) {
+ throw new IllegalStateException(String.format("Shard with name %s already exists",
+ moduleShardConfig.getShardName()));
+ }
+
+ configuration.addModuleShardConfiguration(moduleShardConfig);
+
+ ShardIdentifier shardId = getShardIdentifier(cluster.getCurrentMemberName(), moduleShardConfig.getShardName());
+ Map<String, String> peerAddresses = getPeerAddresses(moduleShardConfig.getShardName()/*,
+ moduleShardConfig.getShardMemberNames()*/);
+
+ LOG.debug("onCreateShard: shardId: {}, memberNames: {}. peerAddresses: {}", shardId,
+ moduleShardConfig.getShardMemberNames(), peerAddresses);
+
+ DatastoreContext shardDatastoreContext = createShard.getDatastoreContext();
+ if(shardDatastoreContext == null) {
+ shardDatastoreContext = datastoreContext;
+ }
+
+ ShardInformation info = new ShardInformation(moduleShardConfig.getShardName(), shardId, peerAddresses,
+ shardDatastoreContext, createShard.getShardPropsCreator());
+ localShards.put(info.getShardName(), info);
+
+ mBean.addLocalShard(shardId.toString());
+
+ if(schemaContext != null) {
+ info.setActor(newShardActor(schemaContext, info));
+ }
+
+ reply = new CreateShardReply();
+ } catch (Exception e) {
+ LOG.error("onCreateShard failed", e);
+ reply = new akka.actor.Status.Failure(e);
+ }
+
+ if(getSender() != null && !getContext().system().deadLetters().equals(getSender())) {
+ getSender().tell(reply, getSelf());
+ }
+ }
+
private void checkReady(){
if (isReadyWithLeaderId()) {
LOG.info("{}: All Shards are ready - data store {} is ready, available count is {}",
LOG.debug("{}: Received MemberRemoved: memberName: {}, address: {}", persistenceId(), memberName,
message.member().address());
- memberNameToAddress.remove(message.member().roles().head());
+ memberNameToAddress.remove(memberName);
+
+ for(ShardInformation info : localShards.values()){
+ info.peerDown(memberName, getShardIdentifier(memberName, info.getShardName()).toString(), getSelf());
+ }
+ }
+
+ private void memberExited(ClusterEvent.MemberExited message) {
+ String memberName = message.member().roles().head();
+
+ LOG.debug("{}: Received MemberExited: memberName: {}, address: {}", persistenceId(), memberName,
+ message.member().address());
+
+ memberNameToAddress.remove(memberName);
+
+ for(ShardInformation info : localShards.values()){
+ info.peerDown(memberName, getShardIdentifier(memberName, info.getShardName()).toString(), getSelf());
+ }
}
private void memberUp(ClusterEvent.MemberUp message) {
for(ShardInformation info : localShards.values()){
String shardName = info.getShardName();
- info.updatePeerAddress(getShardIdentifier(memberName, shardName).toString(),
- getShardActorPath(shardName, memberName), getSelf());
+ String peerId = getShardIdentifier(memberName, shardName).toString();
+ info.updatePeerAddress(peerId, getShardActorPath(shardName, memberName), getSelf());
+
+ info.peerUp(memberName, peerId, getSelf());
}
checkReady();
primaryShardInfoCache.remove(info.getShardName());
}
+
+ info.peerDown(memberName, getShardIdentifier(memberName, info.getShardName()).toString(), getSelf());
}
}
LOG.debug("Marking Leader {} as available.", leaderId);
info.setLeaderAvailable(true);
}
+
+ info.peerUp(memberName, getShardIdentifier(memberName, info.getShardName()).toString(), getSelf());
}
}
* @param message
*/
private void updateSchemaContext(final Object message) {
- final SchemaContext schemaContext = ((UpdateSchemaContext) message).getSchemaContext();
+ schemaContext = ((UpdateSchemaContext) message).getSchemaContext();
LOG.debug("Got updated SchemaContext: # of modules {}", schemaContext.getAllModuleIdentifiers().size());
@VisibleForTesting
protected ActorRef newShardActor(final SchemaContext schemaContext, ShardInformation info) {
- return getContext().actorOf(Shard.props(info.getShardId(),
- info.getPeerAddresses(), datastoreContext, schemaContext)
+ return getContext().actorOf(info.newProps(schemaContext)
.withDispatcher(shardDispatcherPath), info.getShardId().toString());
}
*/
private void createLocalShards() {
String memberName = this.cluster.getCurrentMemberName();
- List<String> memberShardNames =
- this.configuration.getMemberShardNames(memberName);
+ Collection<String> memberShardNames = this.configuration.getMemberShardNames(memberName);
+ ShardPropsCreator shardPropsCreator = new DefaultShardPropsCreator();
List<String> localShardActorNames = new ArrayList<>();
for(String shardName : memberShardNames){
ShardIdentifier shardId = getShardIdentifier(memberName, shardName);
Map<String, String> peerAddresses = getPeerAddresses(shardName);
localShardActorNames.add(shardId.toString());
- localShards.put(shardName, new ShardInformation(shardName, shardId, peerAddresses));
+ localShards.put(shardName, new ShardInformation(shardName, shardId, peerAddresses, datastoreContext,
+ shardPropsCreator));
}
- mBean = ShardManagerInfo.createShardManagerMBean("shard-manager-" + this.type,
+ mBean = ShardManagerInfo.createShardManagerMBean(memberName, "shard-manager-" + this.type,
datastoreContext.getDataStoreMXBeanType(), localShardActorNames);
mBean.setShardManager(this);
* Given the name of the shard find the addresses of all it's peers
*
* @param shardName
- * @return
*/
- private Map<String, String> getPeerAddresses(String shardName){
-
+ private Map<String, String> getPeerAddresses(String shardName) {
+ Collection<String> members = configuration.getMembersFromShardName(shardName);
Map<String, String> peerAddresses = new HashMap<>();
- List<String> members = this.configuration.getMembersFromShardName(shardName);
-
String currentMemberName = this.cluster.getCurrentMemberName();
- for(String memberName : members){
- if(!currentMemberName.equals(memberName)){
+ for(String memberName : members) {
+ if(!currentMemberName.equals(memberName)) {
ShardIdentifier shardId = getShardIdentifier(memberName, shardName);
- String path = getShardActorPath(shardName, currentMemberName);
+ String path = getShardActorPath(shardName, memberName);
peerAddresses.put(shardId.toString(), path);
}
}
private String leaderId;
private short leaderVersion;
+ private final DatastoreContext datastoreContext;
+ private final ShardPropsCreator shardPropsCreator;
+
private ShardInformation(String shardName, ShardIdentifier shardId,
- Map<String, String> peerAddresses) {
+ Map<String, String> peerAddresses, DatastoreContext datastoreContext,
+ ShardPropsCreator shardPropsCreator) {
this.shardName = shardName;
this.shardId = shardId;
this.peerAddresses = peerAddresses;
+ this.datastoreContext = datastoreContext;
+ this.shardPropsCreator = shardPropsCreator;
+ }
+
+ Props newProps(SchemaContext schemaContext) {
+ return shardPropsCreator.newProps(shardId, peerAddresses, datastoreContext, schemaContext);
}
String getShardName() {
peerId, peerAddress, actor.path());
}
- actor.tell(new PeerAddressResolved(peerId.toString(), peerAddress), sender);
+ actor.tell(new PeerAddressResolved(peerId, peerAddress), sender);
}
notifyOnShardInitializedCallbacks();
}
}
+ void peerDown(String memberName, String peerId, ActorRef sender) {
+ if(peerAddresses.containsKey(peerId) && actor != null) {
+ actor.tell(new PeerDown(memberName, peerId), sender);
+ }
+ }
+
+ void peerUp(String memberName, String peerId, ActorRef sender) {
+ if(peerAddresses.containsKey(peerId) && actor != null) {
+ actor.tell(new PeerUp(memberName, peerId), sender);
+ }
+ }
+
boolean isShardReady() {
return !RaftState.Candidate.name().equals(role) && !Strings.isNullOrEmpty(role);
}
Code Review / controller.git / blobdiff