X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2FShardManager.java;h=2d793157744c221454c8a542c65ed926244efb0f;hp=f4fa7b3a97e8617f8ac474b494a08deb330a0b6d;hb=ecccb6d5b43dd73aef0d2d19349d19ee9b4728f7;hpb=69c1fa1f26d6c9130696e7916c6ebe9a9e9d635c diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardManager.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardManager.java index f4fa7b3a97..2d79315774 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardManager.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardManager.java @@ -18,9 +18,7 @@ import akka.actor.SupervisorStrategy; import akka.cluster.ClusterEvent; import akka.japi.Creator; import akka.japi.Function; -import akka.japi.Procedure; import akka.persistence.RecoveryCompleted; -import akka.persistence.RecoveryFailure; import akka.serialization.Serialization; import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Objects; @@ -28,52 +26,57 @@ import com.google.common.base.Optional; import com.google.common.base.Preconditions; import com.google.common.base.Strings; import com.google.common.base.Supplier; -import com.google.common.collect.ImmutableSet; import com.google.common.collect.Sets; import java.io.Serializable; import java.util.ArrayList; import java.util.Collection; -import java.util.Collections; import java.util.HashMap; -import java.util.HashSet; import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.Set; import java.util.concurrent.CountDownLatch; -import org.opendaylight.controller.cluster.DataPersistenceProvider; -import org.opendaylight.controller.cluster.NonPersistentDataProvider; -import org.opendaylight.controller.cluster.PersistentDataProvider; +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.identifiers.ShardIdentifier; -import org.opendaylight.controller.cluster.datastore.identifiers.ShardManagerIdentifier; 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; +import org.opendaylight.controller.cluster.datastore.messages.SwitchShardBehavior; import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext; +import org.opendaylight.controller.cluster.datastore.messages.AddShardReplica; +import org.opendaylight.controller.cluster.datastore.messages.RemoveShardReplica; import org.opendaylight.controller.cluster.datastore.utils.Dispatchers; +import org.opendaylight.controller.cluster.datastore.utils.PrimaryShardInfoFutureCache; import org.opendaylight.controller.cluster.notifications.RegisterRoleChangeListener; 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.ModuleIdentifier; 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, @@ -88,11 +91,6 @@ public class ShardManager extends AbstractUntypedPersistentActorWithMetering { private static final Logger LOG = LoggerFactory.getLogger(ShardManager.class); - // Stores a mapping between a member name and the address of the member - // Member names look like "member-1", "member-2" etc and are as specified - // in configuration - private final Map memberNameToAddress = new HashMap<>(); - // Stores a mapping between a shard name and it's corresponding information // Shard names look like inventory, topology etc and are as specified in // configuration @@ -102,8 +100,6 @@ public class ShardManager extends AbstractUntypedPersistentActorWithMetering { // A data store could be of type config/operational private final String type; - private final String shardManagerIdentifierString; - private final ClusterWrapper cluster; private final Configuration configuration; @@ -114,26 +110,32 @@ public class ShardManager extends AbstractUntypedPersistentActorWithMetering { private DatastoreContext datastoreContext; - private Collection knownModules = Collections.emptySet(); + private final CountDownLatch waitTillReadyCountdownLatch; - private final DataPersistenceProvider dataPersistenceProvider; + private final PrimaryShardInfoFutureCache primaryShardInfoCache; - private final CountDownLatch waitTillReadyCountdownLatch; + private final ShardPeerAddressResolver peerAddressResolver; + + private SchemaContext schemaContext; /** */ protected ShardManager(ClusterWrapper cluster, Configuration configuration, - DatastoreContext datastoreContext, CountDownLatch waitTillReadyCountdownLatch) { + DatastoreContext datastoreContext, CountDownLatch waitTillReadyCountdownLatch, + PrimaryShardInfoFutureCache primaryShardInfoCache) { this.cluster = Preconditions.checkNotNull(cluster, "cluster should not be null"); this.configuration = Preconditions.checkNotNull(configuration, "configuration should not be null"); this.datastoreContext = datastoreContext; - this.dataPersistenceProvider = createDataPersistenceProvider(datastoreContext.isPersistent()); this.type = datastoreContext.getDataStoreType(); - this.shardManagerIdentifierString = ShardManagerIdentifier.builder().type(type).build().toString(); this.shardDispatcherPath = new Dispatchers(context().system().dispatchers()).getDispatcherPath(Dispatchers.DispatcherType.Shard); this.waitTillReadyCountdownLatch = waitTillReadyCountdownLatch; + this.primaryShardInfoCache = primaryShardInfoCache; + + peerAddressResolver = new ShardPeerAddressResolver(type, cluster.getCurrentMemberName()); + this.datastoreContext = DatastoreContext.newBuilderFrom(datastoreContext).shardPeerAddressResolver( + peerAddressResolver).build(); // Subscribe this actor to cluster member events cluster.subscribeToMemberEvents(getSelf()); @@ -141,21 +143,20 @@ public class ShardManager extends AbstractUntypedPersistentActorWithMetering { createLocalShards(); } - protected DataPersistenceProvider createDataPersistenceProvider(boolean persistent) { - return (persistent) ? new PersistentDataProvider(this) : new NonPersistentDataProvider(); - } - public static Props props( - final ClusterWrapper cluster, - final Configuration configuration, - final DatastoreContext datastoreContext, - final CountDownLatch waitTillReadyCountdownLatch) { + final ClusterWrapper cluster, + final Configuration configuration, + final DatastoreContext datastoreContext, + final CountDownLatch waitTillReadyCountdownLatch, + final PrimaryShardInfoFutureCache primaryShardInfoCache) { Preconditions.checkNotNull(cluster, "cluster should not be null"); Preconditions.checkNotNull(configuration, "configuration should not be null"); Preconditions.checkNotNull(waitTillReadyCountdownLatch, "waitTillReadyCountdownLatch should not be null"); + Preconditions.checkNotNull(primaryShardInfoCache, "primaryShardInfoCache should not be null"); - return Props.create(new ShardManagerCreator(cluster, configuration, datastoreContext, waitTillReadyCountdownLatch)); + return Props.create(new ShardManagerCreator(cluster, configuration, datastoreContext, + waitTillReadyCountdownLatch, primaryShardInfoCache)); } @Override @@ -177,10 +178,14 @@ public class ShardManager extends AbstractUntypedPersistentActorWithMetering { 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) { - ignoreMessage(message); + memberUnreachable((ClusterEvent.UnreachableMember)message); + } else if(message instanceof ClusterEvent.ReachableMember) { + memberReachable((ClusterEvent.ReachableMember) message); } else if(message instanceof DatastoreContext) { onDatastoreContext((DatastoreContext)message); } else if(message instanceof RoleChangeNotification) { @@ -190,13 +195,68 @@ public class ShardManager extends AbstractUntypedPersistentActorWithMetering { } 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 if(message instanceof AddShardReplica){ + onAddShardReplica((AddShardReplica)message); + } else if(message instanceof RemoveShardReplica){ + onRemoveShardReplica((RemoveShardReplica)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 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; + } else { + shardDatastoreContext = DatastoreContext.newBuilderFrom(shardDatastoreContext).shardPeerAddressResolver( + peerAddressResolver).build(); + } + + ShardInformation info = new ShardInformation(moduleShardConfig.getShardName(), shardId, peerAddresses, + shardDatastoreContext, createShard.getShardPropsCreator(), peerAddressResolver); + 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 {}", @@ -212,7 +272,11 @@ public class ShardManager extends AbstractUntypedPersistentActorWithMetering { ShardInformation shardInformation = findShardInformation(leaderStateChanged.getMemberId()); if(shardInformation != null) { shardInformation.setLocalDataTree(leaderStateChanged.getLocalShardDataTree()); - shardInformation.setLeaderId(leaderStateChanged.getLeaderId()); + shardInformation.setLeaderVersion(leaderStateChanged.getLeaderPayloadVersion()); + if(shardInformation.setLeaderId(leaderStateChanged.getLeaderId())) { + primaryShardInfoCache.remove(shardInformation.getShardName()); + } + checkReady(); } else { LOG.debug("No shard found with member Id {}", leaderStateChanged.getMemberId()); @@ -324,26 +388,12 @@ public class ShardManager extends AbstractUntypedPersistentActorWithMetering { @Override protected void handleRecover(Object message) throws Exception { - if(dataPersistenceProvider.isRecoveryApplicable()) { - if (message instanceof SchemaContextModules) { - SchemaContextModules msg = (SchemaContextModules) message; - knownModules = ImmutableSet.copyOf(msg.getModules()); - } else if (message instanceof RecoveryFailure) { - RecoveryFailure failure = (RecoveryFailure) message; - LOG.error("Recovery failed", failure.cause()); - } else if (message instanceof RecoveryCompleted) { - LOG.info("Recovery complete : {}", persistenceId()); - - // Delete all the messages from the akka journal except the last one - deleteMessages(lastSequenceNr() - 1); - } - } else { - if (message instanceof RecoveryCompleted) { - LOG.info("Recovery complete : {}", persistenceId()); + if (message instanceof RecoveryCompleted) { + LOG.info("Recovery complete : {}", persistenceId()); - // Delete all the messages from the akka journal - deleteMessages(lastSequenceNr()); - } + // We no longer persist SchemaContext modules so delete all the prior messages from the akka + // journal on upgrade from Helium. + deleteMessages(lastSequenceNr()); } } @@ -384,8 +434,16 @@ public class ShardManager extends AbstractUntypedPersistentActorWithMetering { 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()); @@ -407,13 +465,11 @@ public class ShardManager extends AbstractUntypedPersistentActorWithMetering { getSender().tell(messageSupplier.get(), 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)); + private static NoShardLeaderException createNoShardLeaderException(ShardIdentifier shardId) { + return new NoShardLeaderException(null, shardId.toString()); } - private NotInitializedException createNotInitializedException(ShardIdentifier shardId) { + private static NotInitializedException createNotInitializedException(ShardIdentifier shardId) { return new NotInitializedException(String.format( "Found primary shard %s but it's not initialized yet. Please try again later", shardId)); } @@ -424,7 +480,24 @@ public class ShardManager extends AbstractUntypedPersistentActorWithMetering { LOG.debug("{}: Received MemberRemoved: memberName: {}, address: {}", persistenceId(), memberName, message.member().address()); - memberNameToAddress.remove(message.member().roles().head()); + peerAddressResolver.removePeerAddress(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()); + + peerAddressResolver.removePeerAddress(memberName); + + for(ShardInformation info : localShards.values()){ + info.peerDown(memberName, getShardIdentifier(memberName, info.getShardName()).toString(), getSelf()); + } } private void memberUp(ClusterEvent.MemberUp message) { @@ -433,19 +506,68 @@ public class ShardManager extends AbstractUntypedPersistentActorWithMetering { LOG.debug("{}: Received MemberUp: memberName: {}, address: {}", persistenceId(), memberName, message.member().address()); - memberNameToAddress.put(memberName, message.member().address()); + addPeerAddress(memberName, message.member().address()); + + checkReady(); + } + + private void addPeerAddress(String memberName, Address address) { + peerAddressResolver.addPeerAddress(memberName, address); 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, peerAddressResolver.getShardActorAddress(shardName, memberName), getSelf()); + + info.peerUp(memberName, peerId, getSelf()); } + } - checkReady(); + private void memberReachable(ClusterEvent.ReachableMember message) { + String memberName = message.member().roles().head(); + LOG.debug("Received ReachableMember: memberName {}, address: {}", memberName, message.member().address()); + + addPeerAddress(memberName, message.member().address()); + + markMemberAvailable(memberName); + } + + private void memberUnreachable(ClusterEvent.UnreachableMember message) { + String memberName = message.member().roles().head(); + LOG.debug("Received UnreachableMember: memberName {}, address: {}", memberName, message.member().address()); + + markMemberUnavailable(memberName); + } + + private void markMemberUnavailable(final String memberName) { + for(ShardInformation info : localShards.values()){ + String leaderId = info.getLeaderId(); + if(leaderId != null && leaderId.contains(memberName)) { + LOG.debug("Marking Leader {} as unavailable.", leaderId); + info.setLeaderAvailable(false); + + primaryShardInfoCache.remove(info.getShardName()); + } + + info.peerDown(memberName, getShardIdentifier(memberName, info.getShardName()).toString(), getSelf()); + } + } + + private void markMemberAvailable(final String memberName) { + for(ShardInformation info : localShards.values()){ + String leaderId = info.getLeaderId(); + if(leaderId != null && leaderId.contains(memberName)) { + LOG.debug("Marking Leader {} as available.", leaderId); + info.setLeaderAvailable(true); + } + + info.peerUp(memberName, getShardIdentifier(memberName, info.getShardName()).toString(), getSelf()); + } } private void onDatastoreContext(DatastoreContext context) { - datastoreContext = context; + datastoreContext = DatastoreContext.newBuilderFrom(context).shardPeerAddressResolver( + peerAddressResolver).build(); for (ShardInformation info : localShards.values()) { if (info.getActor() != null) { info.getActor().tell(datastoreContext, getSelf()); @@ -453,48 +575,38 @@ public class ShardManager extends AbstractUntypedPersistentActorWithMetering { } } + 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(); - Set allModuleIdentifiers = schemaContext.getAllModuleIdentifiers(); - Set newModules = new HashSet<>(128); + LOG.debug("Got updated SchemaContext: # of modules {}", schemaContext.getAllModuleIdentifiers().size()); - for(ModuleIdentifier moduleIdentifier : allModuleIdentifiers){ - String s = moduleIdentifier.getNamespace().toString(); - newModules.add(s); - } - - if(newModules.containsAll(knownModules)) { - - LOG.debug("New SchemaContext has a super set of current knownModules - persisting info"); - - knownModules = ImmutableSet.copyOf(newModules); - - dataPersistenceProvider.persist(new SchemaContextModules(newModules), new Procedure() { - - @Override - public void apply(SchemaContextModules param) throws Exception { - LOG.debug("Sending new SchemaContext to Shards"); - for (ShardInformation info : localShards.values()) { - if (info.getActor() == null) { - info.setActor(newShardActor(schemaContext, info)); - } else { - info.getActor().tell(message, getSelf()); - } - } - } - - }); - } else { - LOG.debug("Rejecting schema context update - not a super set of previously known modules:\nUPDATE: {}\nKNOWN: {}", - newModules, knownModules); + for (ShardInformation info : localShards.values()) { + if (info.getActor() == null) { + LOG.debug("Creating Shard {}", info.getShardId()); + info.setActor(newShardActor(schemaContext, info)); + } else { + info.getActor().tell(message, getSelf()); + } } - } @VisibleForTesting @@ -504,9 +616,8 @@ public class ShardManager extends AbstractUntypedPersistentActorWithMetering { @VisibleForTesting protected ActorRef newShardActor(final SchemaContext schemaContext, ShardInformation info) { - return getContext().actorOf(Shard.props(info.getShardId(), - info.getPeerAddresses(), datastoreContext, schemaContext) - .withDispatcher(shardDispatcherPath), info.getShardId().toString()); + return getContext().actorOf(info.newProps(schemaContext) + .withDispatcher(shardDispatcherPath), info.getShardId().toString()); } private void findPrimary(FindPrimary message) { @@ -524,30 +635,26 @@ public class ShardManager extends AbstractUntypedPersistentActorWithMetering { 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); - } + if(LOG.isDebugEnabled()) { + LOG.debug("{}: Found primary for {}: {}", persistenceId(), shardName, found); + } - return found; + return found; } }); return; } - for(Map.Entry entry: memberNameToAddress.entrySet()) { - if(!cluster.getCurrentMemberName().equals(entry.getKey())) { - String path = getShardManagerActorPathBuilder(entry.getValue()).toString(); - - LOG.debug("{}: findPrimary for {} forwarding to remote ShardManager {}", persistenceId(), - shardName, path); + for(String address: peerAddressResolver.getShardManagerPeerActorAddresses()) { + LOG.debug("{}: findPrimary for {} forwarding to remote ShardManager {}", persistenceId(), + shardName, address); - getContext().actorSelection(path).forward(new RemoteFindPrimary(shardName, - message.isWaitUntilReady()), getContext()); - return; - } + getContext().actorSelection(address).forward(new RemoteFindPrimary(shardName, + message.isWaitUntilReady()), getContext()); + return; } LOG.debug("{}: No shard found for {}", persistenceId(), shardName); @@ -556,23 +663,6 @@ public class ShardManager extends AbstractUntypedPersistentActorWithMetering { String.format("No primary shard found for %s.", shardName)), getSelf()); } - private StringBuilder getShardManagerActorPathBuilder(Address address) { - StringBuilder builder = new StringBuilder(); - builder.append(address.toString()).append("/user/").append(shardManagerIdentifierString); - return builder; - } - - private String getShardActorPath(String shardName, String memberName) { - Address address = memberNameToAddress.get(memberName); - if(address != null) { - StringBuilder builder = getShardManagerActorPathBuilder(address); - builder.append("/") - .append(getShardIdentifier(memberName, shardName)); - return builder.toString(); - } - return null; - } - /** * Construct the name of the shard actor given the name of the member on * which the shard resides and the name of the shard @@ -582,7 +672,7 @@ public class ShardManager extends AbstractUntypedPersistentActorWithMetering { * @return */ private ShardIdentifier getShardIdentifier(String memberName, String shardName){ - return ShardIdentifier.builder().memberName(memberName).shardName(shardName).type(type).build(); + return peerAddressResolver.getShardIdentifier(memberName, shardName); } /** @@ -592,40 +682,40 @@ public class ShardManager extends AbstractUntypedPersistentActorWithMetering { */ private void createLocalShards() { String memberName = this.cluster.getCurrentMemberName(); - List memberShardNames = - this.configuration.getMemberShardNames(memberName); + Collection memberShardNames = this.configuration.getMemberShardNames(memberName); + ShardPropsCreator shardPropsCreator = new DefaultShardPropsCreator(); List localShardActorNames = new ArrayList<>(); for(String shardName : memberShardNames){ ShardIdentifier shardId = getShardIdentifier(memberName, shardName); Map 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, peerAddressResolver)); } - mBean = ShardManagerInfo.createShardManagerMBean("shard-manager-" + this.type, - datastoreContext.getDataStoreMXBeanType(), localShardActorNames); + 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 getPeerAddresses(String shardName){ - + private Map getPeerAddresses(String shardName) { + Collection members = configuration.getMembersFromShardName(shardName); Map peerAddresses = new HashMap<>(); - List 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); - peerAddresses.put(shardId.toString(), path); + String address = peerAddressResolver.getShardActorAddress(shardName, memberName); + peerAddresses.put(shardId.toString(), address); } } return peerAddresses; @@ -635,14 +725,14 @@ public class ShardManager extends AbstractUntypedPersistentActorWithMetering { public SupervisorStrategy supervisorStrategy() { return new OneForOneStrategy(10, Duration.create("1 minute"), - new Function() { - @Override - public SupervisorStrategy.Directive apply(Throwable t) { - LOG.warn("Supervisor Strategy caught unexpected exception - resuming", t); - return SupervisorStrategy.resume(); - } + new Function() { + @Override + public SupervisorStrategy.Directive apply(Throwable t) { + LOG.warn("Supervisor Strategy caught unexpected exception - resuming", t); + return SupervisorStrategy.resume(); } - ); + } + ); } @@ -652,18 +742,50 @@ public class ShardManager extends AbstractUntypedPersistentActorWithMetering { } @VisibleForTesting - Collection getKnownModules() { - return knownModules; + ShardManagerInfoMBean getMBean(){ + return mBean; } - @VisibleForTesting - DataPersistenceProvider getDataPersistenceProvider() { - return dataPersistenceProvider; + private void onAddShardReplica (AddShardReplica shardReplicaMsg) { + String shardName = shardReplicaMsg.getShardName(); + + // verify the local shard replica is already available in the controller node + if (localShards.containsKey(shardName)) { + LOG.debug ("Local shard {} already available in the controller node", shardName); + getSender().tell(new akka.actor.Status.Failure( + new IllegalArgumentException(String.format("Local shard %s already exists", + shardName))), getSelf()); + return; + } + // verify the shard with the specified name is present in the cluster configuration + if (!(this.configuration.isShardConfigured(shardName))) { + LOG.debug ("No module configuration exists for shard {}", shardName); + getSender().tell(new akka.actor.Status.Failure(new IllegalArgumentException( + String.format("No module configuration exists for shard %s", + shardName))), getSelf()); + return; + } + + // Create the localShard + getSender().tell(new akka.actor.Status.Success(true), getSelf()); + return; } - @VisibleForTesting - ShardManagerInfoMBean getMBean(){ - return mBean; + private void onRemoveShardReplica (RemoveShardReplica shardReplicaMsg) { + String shardName = shardReplicaMsg.getShardName(); + boolean deleteStatus = false; + + // verify the local shard replica is available in the controller node + if (!localShards.containsKey(shardName)) { + LOG.debug ("Local shard replica {} is not available in the controller node", shardName); + getSender().tell(new akka.actor.Status.Failure( + new IllegalArgumentException(String.format("Local shard %s not available", + shardName))), getSelf()); + return; + } + // call RemoveShard for the shardName + getSender().tell(new akka.actor.Status.Success(true), getSelf()); + return; } @VisibleForTesting @@ -672,8 +794,9 @@ public class ShardManager extends AbstractUntypedPersistentActorWithMetering { private final String shardName; private ActorRef actor; private ActorPath actorPath; - private final Map peerAddresses; + private final Map initialPeerAddresses; private Optional localShardDataTree; + private boolean leaderAvailable = false; // flag that determines if the actor is ready for business private boolean actorInitialized = false; @@ -683,12 +806,25 @@ public class ShardManager extends AbstractUntypedPersistentActorWithMetering { private final Set onShardInitializedSet = Sets.newHashSet(); private String role ; private String leaderId; + private short leaderVersion; + + private final DatastoreContext datastoreContext; + private final ShardPropsCreator shardPropsCreator; + private final ShardPeerAddressResolver addressResolver; private ShardInformation(String shardName, ShardIdentifier shardId, - Map peerAddresses) { + Map initialPeerAddresses, DatastoreContext datastoreContext, + ShardPropsCreator shardPropsCreator, ShardPeerAddressResolver addressResolver) { this.shardName = shardName; this.shardId = shardId; - this.peerAddresses = peerAddresses; + this.initialPeerAddresses = initialPeerAddresses; + this.datastoreContext = datastoreContext; + this.shardPropsCreator = shardPropsCreator; + this.addressResolver = addressResolver; + } + + Props newProps(SchemaContext schemaContext) { + return shardPropsCreator.newProps(shardId, initialPeerAddresses, datastoreContext, schemaContext); } String getShardName() { @@ -720,26 +856,30 @@ public class ShardManager extends AbstractUntypedPersistentActorWithMetering { return localShardDataTree; } - Map getPeerAddresses() { - return peerAddresses; - } - void updatePeerAddress(String peerId, String peerAddress, ActorRef sender){ - LOG.info("updatePeerAddress for peer {} with address {}", peerId, - peerAddress); - if(peerAddresses.containsKey(peerId)){ - peerAddresses.put(peerId, peerAddress); - - if(actor != null) { - if(LOG.isDebugEnabled()) { - LOG.debug("Sending PeerAddressResolved for peer {} with address {} to {}", - peerId, peerAddress, actor.path()); - } + LOG.info("updatePeerAddress for peer {} with address {}", peerId, peerAddress); - actor.tell(new PeerAddressResolved(peerId.toString(), peerAddress), sender); + if(actor != null) { + if(LOG.isDebugEnabled()) { + LOG.debug("Sending PeerAddressResolved for peer {} with address {} to {}", + peerId, peerAddress, actor.path()); } - notifyOnShardInitializedCallbacks(); + actor.tell(new PeerAddressResolved(peerId, peerAddress), sender); + } + + notifyOnShardInitializedCallbacks(); + } + + void peerDown(String memberName, String peerId, ActorRef sender) { + if(actor != null) { + actor.tell(new PeerDown(memberName, peerId), sender); + } + } + + void peerUp(String memberName, String peerId, ActorRef sender) { + if(actor != null) { + actor.tell(new PeerUp(memberName, peerId), sender); } } @@ -748,7 +888,8 @@ public class ShardManager extends AbstractUntypedPersistentActorWithMetering { } boolean isShardReadyWithLeaderId() { - return isShardReady() && (isLeader() || peerAddresses.get(leaderId) != null); + return leaderAvailable && isShardReady() && !RaftState.IsolatedLeader.name().equals(role) && + (isLeader() || addressResolver.resolve(leaderId) != null); } boolean isShardInitialized() { @@ -763,7 +904,7 @@ public class ShardManager extends AbstractUntypedPersistentActorWithMetering { if(isLeader()) { return Serialization.serializedActorPath(getActor()); } else { - return peerAddresses.get(leaderId); + return addressResolver.resolve(leaderId); } } @@ -826,10 +967,31 @@ public class ShardManager extends AbstractUntypedPersistentActorWithMetering { return false; } - void setLeaderId(String leaderId) { + boolean setLeaderId(String leaderId) { + boolean changed = !Objects.equal(this.leaderId, leaderId); this.leaderId = leaderId; - + if(leaderId != null) { + this.leaderAvailable = true; + } notifyOnShardInitializedCallbacks(); + + return changed; + } + + String getLeaderId() { + return leaderId; + } + + void setLeaderAvailable(boolean leaderAvailable) { + this.leaderAvailable = leaderAvailable; + } + + short getLeaderVersion() { + return leaderVersion; + } + + void setLeaderVersion(short leaderVersion) { + this.leaderVersion = leaderVersion; } } @@ -840,18 +1002,21 @@ public class ShardManager extends AbstractUntypedPersistentActorWithMetering { final Configuration configuration; final DatastoreContext datastoreContext; private final CountDownLatch waitTillReadyCountdownLatch; + private final PrimaryShardInfoFutureCache primaryShardInfoCache; - ShardManagerCreator(ClusterWrapper cluster, - Configuration configuration, DatastoreContext datastoreContext, CountDownLatch waitTillReadyCountdownLatch) { + ShardManagerCreator(ClusterWrapper cluster, Configuration configuration, DatastoreContext datastoreContext, + CountDownLatch waitTillReadyCountdownLatch, PrimaryShardInfoFutureCache primaryShardInfoCache) { this.cluster = cluster; this.configuration = configuration; this.datastoreContext = datastoreContext; this.waitTillReadyCountdownLatch = waitTillReadyCountdownLatch; + this.primaryShardInfoCache = primaryShardInfoCache; } @Override public ShardManager create() throws Exception { - return new ShardManager(cluster, configuration, datastoreContext, waitTillReadyCountdownLatch); + return new ShardManager(cluster, configuration, datastoreContext, waitTillReadyCountdownLatch, + primaryShardInfoCache); } } @@ -906,6 +1071,11 @@ public class ShardManager extends AbstractUntypedPersistentActorWithMetering { } } + /** + * We no longer persist SchemaContextModules but keep this class around for now for backwards + * compatibility so we don't get de-serialization failures on upgrade from Helium. + */ + @Deprecated static class SchemaContextModules implements Serializable { private static final long serialVersionUID = -8884620101025936590L;