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.Map;
import java.util.Set;
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.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.RemoteFindPrimary;
import org.opendaylight.controller.cluster.datastore.messages.RemotePrimaryShardFound;
import org.opendaylight.controller.cluster.datastore.messages.ShardLeaderStateChanged;
+import org.opendaylight.controller.cluster.datastore.messages.SwitchShardBehavior;
import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext;
import org.opendaylight.controller.cluster.datastore.utils.Dispatchers;
import org.opendaylight.controller.cluster.datastore.utils.PrimaryShardInfoFutureCache;
import org.opendaylight.controller.cluster.notifications.RoleChangeNotification;
import org.opendaylight.controller.cluster.raft.RaftState;
import org.opendaylight.controller.cluster.raft.base.messages.FollowerInitialSyncUpStatus;
+import org.opendaylight.controller.cluster.raft.base.messages.SwitchBehavior;
import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree;
import org.opendaylight.yangtools.yang.model.api.SchemaContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.concurrent.duration.Duration;
+import scala.concurrent.duration.FiniteDuration;
/**
* The ShardManager has the following jobs,
private final PrimaryShardInfoFutureCache primaryShardInfoCache;
+ private SchemaContext schemaContext;
+
/**
*/
protected ShardManager(ClusterWrapper cluster, Configuration configuration,
} else if(message instanceof ShardNotInitializedTimeout) {
onShardNotInitializedTimeout((ShardNotInitializedTimeout)message);
} else if(message instanceof ShardLeaderStateChanged) {
- onLeaderStateChanged((ShardLeaderStateChanged)message);
+ 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 {
+ if(localShards.containsKey(createShard.getShardName())) {
+ throw new IllegalStateException(String.format("Shard with name %s already exists",
+ createShard.getShardName()));
+ }
+
+ ShardIdentifier shardId = getShardIdentifier(cluster.getCurrentMemberName(), createShard.getShardName());
+ Map<String, String> peerAddresses = getPeerAddresses(createShard.getShardName(), createShard.getMemberNames());
+
+ LOG.debug("onCreateShard: shardId: {}, peerAddresses: {}", shardId, peerAddresses);
+
+ DatastoreContext shardDatastoreContext = createShard.getDatastoreContext();
+ if(shardDatastoreContext == null) {
+ shardDatastoreContext = datastoreContext;
+ }
+
+ ShardInformation info = new ShardInformation(createShard.getShardName(), shardId, peerAddresses,
+ shardDatastoreContext, createShard.getShardPropsCreator());
+ localShards.put(createShard.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 {}",
ShardInformation shardInformation = findShardInformation(leaderStateChanged.getMemberId());
if(shardInformation != null) {
shardInformation.setLocalDataTree(leaderStateChanged.getLocalShardDataTree());
+ shardInformation.setLeaderVersion(leaderStateChanged.getLeaderPayloadVersion());
if(shardInformation.setLeaderId(leaderStateChanged.getLeaderId())) {
primaryShardInfoCache.remove(shardInformation.getShardName());
}
LOG.debug("{}: Scheduling timer to wait for shard {}", persistenceId(), shardInformation.getShardName());
+ FiniteDuration timeout = datastoreContext.getShardInitializationTimeout().duration();
+ if(shardInformation.isShardInitialized()) {
+ // If the shard is already initialized then we'll wait enough time for the shard to
+ // elect a leader, ie 2 times the election timeout.
+ timeout = FiniteDuration.create(datastoreContext.getShardRaftConfig()
+ .getElectionTimeOutInterval().toMillis() * 2, TimeUnit.MILLISECONDS);
+ }
+
Cancellable timeoutSchedule = getContext().system().scheduler().scheduleOnce(
- datastoreContext.getShardInitializationTimeout().duration(), getSelf(),
+ timeout, getSelf(),
new ShardNotInitializedTimeout(shardInformation, onShardInitialized, sender),
getContext().dispatcher(), getSelf());
}
private NoShardLeaderException createNoShardLeaderException(ShardIdentifier shardId) {
- return new NoShardLeaderException(String.format(
- "Could not find a leader for shard %s. This typically happens when the system is coming up or " +
- "recovering and a leader is being elected. Try again later.", shardId));
+ return new NoShardLeaderException(null, shardId.toString());
}
private NotInitializedException createNotInitializedException(ShardIdentifier shardId) {
}
}
+ private void onSwitchShardBehavior(SwitchShardBehavior message) {
+ ShardIdentifier identifier = ShardIdentifier.builder().fromShardIdString(message.getShardName()).build();
+
+ ShardInformation shardInformation = localShards.get(identifier.getShardName());
+
+ if(shardInformation != null && shardInformation.getActor() != null) {
+ shardInformation.getActor().tell(
+ new SwitchBehavior(RaftState.valueOf(message.getNewState()), message.getTerm()), getSelf());
+ } else {
+ LOG.warn("Could not switch the behavior of shard {} to {} - shard is not yet available",
+ message.getShardName(), message.getNewState());
+ }
+ }
+
/**
* Notifies all the local shards of a change in the schema context
*
* @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());
}
String primaryPath = info.getSerializedLeaderActor();
Object found = canReturnLocalShardState && info.isLeader() ?
new LocalPrimaryShardFound(primaryPath, info.getLocalShardDataTree().get()) :
- new RemotePrimaryShardFound(primaryPath);
+ new RemotePrimaryShardFound(primaryPath, info.getLeaderVersion());
if(LOG.isDebugEnabled()) {
LOG.debug("{}: Found primary for {}: {}", persistenceId(), shardName, found);
*/
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);
}
/**
* @param shardName
* @return
*/
- private Map<String, String> getPeerAddresses(String shardName){
+ private Map<String, String> getPeerAddresses(String shardName) {
+ return getPeerAddresses(shardName, configuration.getMembersFromShardName(shardName));
+ }
- Map<String, String> peerAddresses = new HashMap<>();
+ private Map<String, String> getPeerAddresses(String shardName, Collection<String> members) {
- List<String> members = this.configuration.getMembersFromShardName(shardName);
+ Map<String, String> peerAddresses = new HashMap<>();
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);
peerAddresses.put(shardId.toString(), path);
private final Set<OnShardInitialized> onShardInitializedSet = Sets.newHashSet();
private String role ;
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() {
}
boolean isShardReadyWithLeaderId() {
- return leaderAvailable && isShardReady() && (isLeader() || peerAddresses.get(leaderId) != null);
+ return leaderAvailable && isShardReady() && !RaftState.IsolatedLeader.name().equals(role) &&
+ (isLeader() || peerAddresses.get(leaderId) != null);
}
boolean isShardInitialized() {
return changed;
}
- public String getLeaderId() {
+ String getLeaderId() {
return leaderId;
}
- public void setLeaderAvailable(boolean leaderAvailable) {
+ void setLeaderAvailable(boolean leaderAvailable) {
this.leaderAvailable = leaderAvailable;
}
+
+ short getLeaderVersion() {
+ return leaderVersion;
+ }
+
+ void setLeaderVersion(short leaderVersion) {
+ this.leaderVersion = leaderVersion;
+ }
}
private static class ShardManagerCreator implements Creator<ShardManager> {
Code Review / controller.git / blobdiff