X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;ds=sidebyside;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2Fshardmanager%2FShardManager.java;h=bb289bc82e365177e55d320ffbbbe92d1cd8f404;hb=refs%2Fchanges%2F90%2F56590%2F4;hp=5bfa1851eedbdcf76222137fcec52eab2a746b9b;hpb=149feb98f151186975fe42bab5853e05aafd4b51;p=controller.git diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/shardmanager/ShardManager.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/shardmanager/ShardManager.java index 5bfa1851ee..bb289bc82e 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/shardmanager/ShardManager.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/shardmanager/ShardManager.java @@ -64,6 +64,7 @@ import org.opendaylight.controller.cluster.datastore.exceptions.NotInitializedEx import org.opendaylight.controller.cluster.datastore.exceptions.PrimaryNotFoundException; import org.opendaylight.controller.cluster.datastore.identifiers.ShardIdentifier; import org.opendaylight.controller.cluster.datastore.messages.ActorInitialized; +import org.opendaylight.controller.cluster.datastore.messages.AddPrefixShardReplica; import org.opendaylight.controller.cluster.datastore.messages.AddShardReplica; import org.opendaylight.controller.cluster.datastore.messages.ChangeShardMembersVotingStatus; import org.opendaylight.controller.cluster.datastore.messages.CreateShard; @@ -75,6 +76,7 @@ import org.opendaylight.controller.cluster.datastore.messages.LocalShardFound; import org.opendaylight.controller.cluster.datastore.messages.LocalShardNotFound; import org.opendaylight.controller.cluster.datastore.messages.RemoteFindPrimary; import org.opendaylight.controller.cluster.datastore.messages.RemotePrimaryShardFound; +import org.opendaylight.controller.cluster.datastore.messages.RemovePrefixShardReplica; import org.opendaylight.controller.cluster.datastore.messages.RemoveShardReplica; import org.opendaylight.controller.cluster.datastore.messages.ShardLeaderStateChanged; import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext; @@ -108,6 +110,7 @@ import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType; import org.opendaylight.controller.md.sal.dom.api.DOMDataTreeChangeListener; import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier; import org.opendaylight.yangtools.concepts.ListenerRegistration; +import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier; import org.opendaylight.yangtools.yang.model.api.SchemaContext; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -161,6 +164,8 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { private final Set shardReplicaOperationsInProgress = new HashSet<>(); + private final Map> shardActorStoppingFutures = new HashMap<>(); + private final String persistenceId; private final AbstractDataStore dataStore; @@ -249,6 +254,8 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { onCreateShard((CreateShard)message); } else if (message instanceof AddShardReplica) { onAddShardReplica((AddShardReplica) message); + } else if (message instanceof AddPrefixShardReplica) { + onAddPrefixShardReplica((AddPrefixShardReplica) message); } else if (message instanceof PrefixShardCreated) { onPrefixShardCreated((PrefixShardCreated) message); } else if (message instanceof PrefixShardRemoved) { @@ -264,6 +271,8 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { onAddServerFailure(msg.shardName, msg.failureMessage, msg.failure, getSender(), msg.removeShardOnFailure); } else if (message instanceof RemoveShardReplica) { onRemoveShardReplica((RemoveShardReplica) message); + } else if (message instanceof RemovePrefixShardReplica) { + onRemovePrefixShardReplica((RemovePrefixShardReplica) message); } else if (message instanceof WrappedShardResponse) { onWrappedShardResponse((WrappedShardResponse) message); } else if (message instanceof GetSnapshot) { @@ -291,7 +300,7 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { } private void onInitConfigListener() { - LOG.debug("{}: Initializing config listener.", persistenceId()); + LOG.debug("{}: Initializing config listener on {}", persistenceId(), cluster.getCurrentMemberName()); final org.opendaylight.mdsal.common.api.LogicalDatastoreType type = org.opendaylight.mdsal.common.api.LogicalDatastoreType @@ -374,6 +383,46 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { } } + private void removePrefixShardReplica(final RemovePrefixShardReplica contextMessage, final String shardName, + final String primaryPath, final ActorRef sender) { + if (isShardReplicaOperationInProgress(shardName, sender)) { + return; + } + + shardReplicaOperationsInProgress.add(shardName); + + final ShardIdentifier shardId = getShardIdentifier(contextMessage.getMemberName(), shardName); + + final DatastoreContext datastoreContext = newShardDatastoreContextBuilder(shardName).build(); + + //inform ShardLeader to remove this shard as a replica by sending an RemoveServer message + LOG.debug("{}: Sending RemoveServer message to peer {} for shard {}", persistenceId(), + primaryPath, shardId); + + Timeout removeServerTimeout = new Timeout(datastoreContext.getShardLeaderElectionTimeout().duration()); + Future futureObj = ask(getContext().actorSelection(primaryPath), + new RemoveServer(shardId.toString()), removeServerTimeout); + + futureObj.onComplete(new OnComplete() { + @Override + public void onComplete(Throwable failure, Object response) { + if (failure != null) { + shardReplicaOperationsInProgress.remove(shardName); + String msg = String.format("RemoveServer request to leader %s for shard %s failed", + primaryPath, shardName); + + LOG.debug("{}: {}", persistenceId(), msg, failure); + + // FAILURE + sender.tell(new Status.Failure(new RuntimeException(msg, failure)), self()); + } else { + // SUCCESS + self().tell(new WrappedShardResponse(shardId, response, primaryPath), sender); + } + } + }, new Dispatchers(context().system().dispatchers()).getDispatcher(Dispatchers.DispatcherType.Client)); + } + private void removeShardReplica(RemoveShardReplica contextMessage, final String shardName, final String primaryPath, final ActorRef sender) { if (isShardReplicaOperationInProgress(shardName, sender)) { @@ -415,16 +464,41 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { } private void onShardReplicaRemoved(ServerRemoved message) { - final ShardIdentifier shardId = new ShardIdentifier.Builder().fromShardIdString(message.getServerId()).build(); - final ShardInformation shardInformation = localShards.remove(shardId.getShardName()); + removeShard(new ShardIdentifier.Builder().fromShardIdString(message.getServerId()).build()); + } + + @SuppressWarnings("checkstyle:IllegalCatch") + private void removeShard(final ShardIdentifier shardId) { + final String shardName = shardId.getShardName(); + final ShardInformation shardInformation = localShards.remove(shardName); if (shardInformation == null) { LOG.debug("{} : Shard replica {} is not present in list", persistenceId(), shardId.toString()); return; - } else if (shardInformation.getActor() != null) { - LOG.debug("{} : Sending Shutdown to Shard actor {}", persistenceId(), shardInformation.getActor()); - shardInformation.getActor().tell(Shutdown.INSTANCE, self()); } - LOG.debug("{} : Local Shard replica for shard {} has been removed", persistenceId(), shardId.getShardName()); + + final ActorRef shardActor = shardInformation.getActor(); + if (shardActor != null) { + LOG.debug("{} : Sending Shutdown to Shard actor {}", persistenceId(), shardActor); + FiniteDuration duration = shardInformation.getDatastoreContext().getShardRaftConfig() + .getElectionTimeOutInterval().$times(3); + final Future stopFuture = Patterns.gracefulStop(shardActor, duration, Shutdown.INSTANCE); + shardActorStoppingFutures.put(shardName, stopFuture); + stopFuture.onComplete(new OnComplete() { + @Override + public void onComplete(Throwable failure, Boolean result) { + if (failure == null) { + LOG.debug("{} : Successfully shut down Shard actor {}", persistenceId(), shardActor); + } else { + LOG.warn("{}: Failed to shut down Shard actor {}", persistenceId(), shardActor, failure); + } + + self().tell((RunnableMessage) () -> shardActorStoppingFutures.remove(shardName), + ActorRef.noSender()); + } + }, new Dispatchers(context().system().dispatchers()).getDispatcher(Dispatchers.DispatcherType.Client)); + } + + LOG.debug("{} : Local Shard replica for shard {} has been removed", persistenceId(), shardName); persistShardList(); } @@ -485,11 +559,14 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { LOG.debug("{}: onPrefixShardCreated: {}", persistenceId(), message); final PrefixShardConfiguration config = message.getConfiguration(); - - final ShardIdentifier shardId = ClusterUtils.getShardIdentifier(cluster.getCurrentMemberName(), - config.getPrefix()); + final ShardIdentifier shardId = getShardIdentifier(cluster.getCurrentMemberName(), + ClusterUtils.getCleanShardName(config.getPrefix().getRootIdentifier())); final String shardName = shardId.getShardName(); + if (isPreviousShardActorStopInProgress(shardName, message)) { + return; + } + if (localShards.containsKey(shardName)) { LOG.debug("{}: Received create for an already existing shard {}", persistenceId(), shardName); final PrefixShardConfiguration existing = @@ -504,6 +581,26 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { doCreatePrefixShard(config, shardId, shardName); } + private boolean isPreviousShardActorStopInProgress(final String shardName, final Object messageToDefer) { + final Future stopFuture = shardActorStoppingFutures.get(shardName); + if (stopFuture == null) { + return false; + } + + LOG.debug("{} : Stop is in progress for shard {} - adding Future callback to defer {}", persistenceId(), + shardName, messageToDefer); + final ActorRef sender = getSender(); + stopFuture.onComplete(new OnComplete() { + @Override + public void onComplete(Throwable failure, Boolean result) { + LOG.debug("{} : Stop complete for shard {} - re-queing {}", persistenceId(), shardName, messageToDefer); + self().tell(messageToDefer, sender); + } + }, new Dispatchers(context().system().dispatchers()).getDispatcher(Dispatchers.DispatcherType.Client)); + + return true; + } + private void doCreatePrefixShard(PrefixShardConfiguration config, ShardIdentifier shardId, String shardName) { configuration.addPrefixShardConfiguration(config); @@ -512,7 +609,7 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { .storeRoot(config.getPrefix().getRootIdentifier()); DatastoreContext shardDatastoreContext = builder.build(); - final Map peerAddresses = Collections.emptyMap(); + final Map peerAddresses = getPeerAddresses(shardName); final boolean isActiveMember = true; LOG.debug("{} doCreatePrefixShard: shardId: {}, memberNames: {}, peerAddresses: {}, isActiveMember: {}", @@ -532,23 +629,11 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { LOG.debug("{}: onPrefixShardRemoved : {}", persistenceId(), message); final DOMDataTreeIdentifier prefix = message.getPrefix(); - final ShardIdentifier shardId = ClusterUtils.getShardIdentifier(cluster.getCurrentMemberName(), prefix); - final ShardInformation shard = localShards.remove(shardId.getShardName()); + final ShardIdentifier shardId = getShardIdentifier(cluster.getCurrentMemberName(), + ClusterUtils.getCleanShardName(prefix.getRootIdentifier())); configuration.removePrefixShardConfiguration(prefix); - - if (shard == null) { - LOG.warn("{}: Received removal for unconfigured shard: {}, ignoring.. ", persistenceId(), prefix); - return; - } - - if (shard.getActor() != null) { - LOG.debug("{} : Sending Shutdown to Shard actor {}", persistenceId(), shard.getActor()); - shard.getActor().tell(Shutdown.INSTANCE, self()); - } - - LOG.debug("{}: Local Shard replica for shard {} has been removed", persistenceId(), shardId.getShardName()); - persistShardList(); + removeShard(shardId); } private void doCreateShard(final CreateShard createShard) { @@ -1015,9 +1100,9 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { } @VisibleForTesting - protected ActorRef newShardActor(final SchemaContext schemaContext, ShardInformation info) { - return getContext().actorOf(info.newProps(schemaContext) - .withDispatcher(shardDispatcherPath), info.getShardId().toString()); + protected ActorRef newShardActor(final SchemaContext shardSchemaContext, final ShardInformation info) { + return getContext().actorOf(info.newProps(shardSchemaContext).withDispatcher(shardDispatcherPath), + info.getShardId().toString()); } private void findPrimary(FindPrimary message) { @@ -1188,6 +1273,40 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { return false; } + private void onAddPrefixShardReplica(final AddPrefixShardReplica message) { + LOG.debug("{}: onAddPrefixShardReplica: {}", persistenceId(), message); + + final ShardIdentifier shardId = getShardIdentifier(cluster.getCurrentMemberName(), + ClusterUtils.getCleanShardName(message.getShardPrefix())); + final String shardName = shardId.getShardName(); + + // Create the localShard + if (schemaContext == null) { + String msg = String.format( + "No SchemaContext is available in order to create a local shard instance for %s", shardName); + LOG.debug("{}: {}", persistenceId(), msg); + getSender().tell(new Status.Failure(new IllegalStateException(msg)), getSelf()); + return; + } + + findPrimary(shardName, new AutoFindPrimaryFailureResponseHandler(getSender(), shardName, persistenceId(), + getSelf()) { + @Override + public void onRemotePrimaryShardFound(RemotePrimaryShardFound response) { + final RunnableMessage runnable = (RunnableMessage) () -> addPrefixShard(getShardName(), + message.getShardPrefix(), response, getSender()); + if (!isPreviousShardActorStopInProgress(getShardName(), runnable)) { + getSelf().tell(runnable, getTargetActor()); + } + } + + @Override + public void onLocalPrimaryFound(LocalPrimaryShardFound message) { + sendLocalReplicaAlreadyExistsReply(getShardName(), getTargetActor()); + } + }); + } + private void onAddShardReplica(final AddShardReplica shardReplicaMsg) { final String shardName = shardReplicaMsg.getShardName(); @@ -1214,15 +1333,17 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { getSelf()) { @Override public void onRemotePrimaryShardFound(RemotePrimaryShardFound response) { - getSelf().tell((RunnableMessage) () -> addShard(getShardName(), response, getSender()), - getTargetActor()); + final RunnableMessage runnable = (RunnableMessage) () -> + addShard(getShardName(), response, getSender()); + if (!isPreviousShardActorStopInProgress(getShardName(), runnable)) { + getSelf().tell(runnable, getTargetActor()); + } } @Override public void onLocalPrimaryFound(LocalPrimaryShardFound message) { sendLocalReplicaAlreadyExistsReply(getShardName(), getTargetActor()); } - }); } @@ -1232,6 +1353,39 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { sender.tell(new Status.Failure(new AlreadyExistsException(msg)), getSelf()); } + private void addPrefixShard(final String shardName, final YangInstanceIdentifier shardPrefix, + final RemotePrimaryShardFound response, final ActorRef sender) { + if (isShardReplicaOperationInProgress(shardName, sender)) { + return; + } + + shardReplicaOperationsInProgress.add(shardName); + + final ShardInformation shardInfo; + final boolean removeShardOnFailure; + ShardInformation existingShardInfo = localShards.get(shardName); + if (existingShardInfo == null) { + removeShardOnFailure = true; + ShardIdentifier shardId = getShardIdentifier(cluster.getCurrentMemberName(), shardName); + + final Builder builder = newShardDatastoreContextBuilder(shardName); + builder.storeRoot(shardPrefix).customRaftPolicyImplementation(DisableElectionsRaftPolicy.class.getName()); + + DatastoreContext datastoreContext = builder.build(); + + shardInfo = new ShardInformation(shardName, shardId, getPeerAddresses(shardName), datastoreContext, + Shard.builder(), peerAddressResolver); + shardInfo.setActiveMember(false); + localShards.put(shardName, shardInfo); + shardInfo.setActor(newShardActor(schemaContext, shardInfo)); + } else { + removeShardOnFailure = false; + shardInfo = existingShardInfo; + } + + execAddShard(shardName, shardInfo, response, removeShardOnFailure, sender); + } + private void addShard(final String shardName, final RemotePrimaryShardFound response, final ActorRef sender) { if (isShardReplicaOperationInProgress(shardName, sender)) { return; @@ -1259,16 +1413,26 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { shardInfo = existingShardInfo; } - String localShardAddress = peerAddressResolver.getShardActorAddress(shardName, cluster.getCurrentMemberName()); + execAddShard(shardName, shardInfo, response, removeShardOnFailure, sender); + } + + private void execAddShard(final String shardName, + final ShardInformation shardInfo, + final RemotePrimaryShardFound response, + final boolean removeShardOnFailure, + final ActorRef sender) { + + final String localShardAddress = + peerAddressResolver.getShardActorAddress(shardName, cluster.getCurrentMemberName()); //inform ShardLeader to add this shard as a replica by sending an AddServer message LOG.debug("{}: Sending AddServer message to peer {} for shard {}", persistenceId(), response.getPrimaryPath(), shardInfo.getShardId()); - Timeout addServerTimeout = new Timeout(shardInfo.getDatastoreContext() + final Timeout addServerTimeout = new Timeout(shardInfo.getDatastoreContext() .getShardLeaderElectionTimeout().duration()); - Future futureObj = ask(getContext().actorSelection(response.getPrimaryPath()), - new AddServer(shardInfo.getShardId().toString(), localShardAddress, true), addServerTimeout); + final Future futureObj = ask(getContext().actorSelection(response.getPrimaryPath()), + new AddServer(shardInfo.getShardId().toString(), localShardAddress, true), addServerTimeout); futureObj.onComplete(new OnComplete() { @Override @@ -1277,7 +1441,7 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { LOG.debug("{}: AddServer request to {} for {} failed", persistenceId(), response.getPrimaryPath(), shardName, failure); - String msg = String.format("AddServer request to leader %s for shard %s failed", + final String msg = String.format("AddServer request to leader %s for shard %s failed", response.getPrimaryPath(), shardName); self().tell(new ForwardedAddServerFailure(shardName, msg, failure, removeShardOnFailure), sender); } else { @@ -1379,6 +1543,32 @@ class ShardManager extends AbstractUntypedPersistentActorWithMetering { }); } + private void onRemovePrefixShardReplica(final RemovePrefixShardReplica message) { + LOG.debug("{}: onRemovePrefixShardReplica: {}", persistenceId(), message); + + final ShardIdentifier shardId = getShardIdentifier(cluster.getCurrentMemberName(), + ClusterUtils.getCleanShardName(message.getShardPrefix())); + final String shardName = shardId.getShardName(); + + findPrimary(shardName, new AutoFindPrimaryFailureResponseHandler(getSender(), + shardName, persistenceId(), getSelf()) { + @Override + public void onRemotePrimaryShardFound(RemotePrimaryShardFound response) { + doRemoveShardReplicaAsync(response.getPrimaryPath()); + } + + @Override + public void onLocalPrimaryFound(LocalPrimaryShardFound response) { + doRemoveShardReplicaAsync(response.getPrimaryPath()); + } + + private void doRemoveShardReplicaAsync(final String primaryPath) { + getSelf().tell((RunnableMessage) () -> removePrefixShardReplica(message, getShardName(), + primaryPath, getSender()), getTargetActor()); + } + }); + } + private void persistShardList() { List shardList = new ArrayList<>(localShards.keySet()); for (ShardInformation shardInfo : localShards.values()) {