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=6162a0327ca6ab229be75b4f8b8c2c994dcb253a;hp=64c6821120f94f99a389c12700757a7b8c7266f5;hb=bc7b4edec3c868a14e0a0de3a3b8e1af2406448b;hpb=a7223085cb1b53fbe96ab5508b5bdcbff2a04224 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 64c6821120..6162a0327c 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,6 +18,10 @@ import akka.cluster.ClusterEvent; import akka.japi.Creator; import akka.japi.Function; import com.google.common.base.Preconditions; +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.FindLocalShard; import org.opendaylight.controller.cluster.datastore.messages.FindPrimary; import org.opendaylight.controller.cluster.datastore.messages.LocalShardFound; @@ -28,6 +32,7 @@ import org.opendaylight.controller.cluster.datastore.messages.PrimaryNotFound; import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext; import scala.concurrent.duration.Duration; +import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -61,6 +66,8 @@ public class ShardManager extends AbstractUntypedActor { private final Configuration configuration; + private ShardManagerInfoMBean mBean; + /** * @param type defines the kind of data that goes into shards created by this shard manager. Examples of type would be * configuration or operational @@ -82,6 +89,11 @@ public class ShardManager extends AbstractUntypedActor { public static Props props(final String type, final ClusterWrapper cluster, final Configuration configuration) { + + Preconditions.checkNotNull(type, "type should not be null"); + Preconditions.checkNotNull(cluster, "cluster should not be null"); + Preconditions.checkNotNull(configuration, "configuration should not be null"); + return Props.create(new Creator() { @Override @@ -108,7 +120,7 @@ public class ShardManager extends AbstractUntypedActor { } else if(message instanceof ClusterEvent.UnreachableMember) { ignoreMessage(message); } else{ - throw new Exception ("Not recognized message received, message="+message); + unknownMessage(message); } } @@ -122,11 +134,8 @@ public class ShardManager extends AbstractUntypedActor { return; } - getSender().tell(new LocalShardNotFound(message.getShardName()), getSelf()); - } - - private void ignoreMessage(Object message){ - LOG.debug("Unhandled message : " + message); + getSender().tell(new LocalShardNotFound(message.getShardName()), + getSelf()); } private void memberRemoved(ClusterEvent.MemberRemoved message) { @@ -140,7 +149,7 @@ public class ShardManager extends AbstractUntypedActor { for(ShardInformation info : localShards.values()){ String shardName = info.getShardName(); - info.updatePeerAddress(getShardActorName(memberName, shardName), + info.updatePeerAddress(getShardIdentifier(memberName, shardName), getShardActorPath(shardName, memberName)); } } @@ -159,9 +168,6 @@ public class ShardManager extends AbstractUntypedActor { private void findPrimary(FindPrimary message) { String shardName = message.getShardName(); - List members = - configuration.getMembersFromShardName(shardName); - // First see if the there is a local replica for the shard ShardInformation info = localShards.get(shardName); if(info != null) { @@ -175,6 +181,9 @@ public class ShardManager extends AbstractUntypedActor { } } + List members = + configuration.getMembersFromShardName(shardName); + if(cluster.getCurrentMemberName() != null) { members.remove(cluster.getCurrentMemberName()); } @@ -196,9 +205,13 @@ public class ShardManager extends AbstractUntypedActor { private String getShardActorPath(String shardName, String memberName) { Address address = memberNameToAddress.get(memberName); if(address != null) { - return address.toString() + "/user/shardmanager-" + this.type + "/" - + getShardActorName( - memberName, shardName); + StringBuilder builder = new StringBuilder(); + builder.append(address.toString()) + .append("/user/") + .append(ShardManagerIdentifier.builder().type(type).build().toString()) + .append("/") + .append(getShardIdentifier(memberName, shardName)); + return builder.toString(); } return null; } @@ -211,8 +224,8 @@ public class ShardManager extends AbstractUntypedActor { * @param shardName * @return */ - private String getShardActorName(String memberName, String shardName){ - return memberName + "-shard-" + shardName + "-" + this.type; + private ShardIdentifier getShardIdentifier(String memberName, String shardName){ + return ShardIdentifier.builder().memberName(memberName).shardName(shardName).type(type).build(); } /** @@ -225,15 +238,20 @@ public class ShardManager extends AbstractUntypedActor { List memberShardNames = this.configuration.getMemberShardNames(memberName); + List localShardActorNames = new ArrayList<>(); for(String shardName : memberShardNames){ - String shardActorName = getShardActorName(memberName, shardName); - Map peerAddresses = getPeerAddresses(shardName); + ShardIdentifier shardId = getShardIdentifier(memberName, shardName); + Map peerAddresses = getPeerAddresses(shardName); ActorRef actor = getContext() - .actorOf(Shard.props(shardActorName, peerAddresses), - shardActorName); + .actorOf(Shard.props(shardId, peerAddresses), + shardId.toString()); + localShardActorNames.add(shardId.toString()); localShards.put(shardName, new ShardInformation(shardName, actor, peerAddresses)); } + mBean = ShardManagerInfo + .createShardManagerMBean("shard-manager-" + this.type, localShardActorNames); + } /** @@ -242,9 +260,9 @@ public class ShardManager extends AbstractUntypedActor { * @param shardName * @return */ - private Map getPeerAddresses(String shardName){ + private Map getPeerAddresses(String shardName){ - Map peerAddresses = new HashMap<>(); + Map peerAddresses = new HashMap<>(); List members = this.configuration.getMembersFromShardName(shardName); @@ -253,16 +271,16 @@ public class ShardManager extends AbstractUntypedActor { for(String memberName : members){ if(!currentMemberName.equals(memberName)){ - String shardActorName = getShardActorName(memberName, shardName); + ShardIdentifier shardId = getShardIdentifier(memberName, + shardName); String path = getShardActorPath(shardName, currentMemberName); - peerAddresses.put(shardActorName, path); + peerAddresses.put(shardId, path); } } return peerAddresses; } - @Override public SupervisorStrategy supervisorStrategy() { return new OneForOneStrategy(10, Duration.create("1 minute"), @@ -280,10 +298,10 @@ public class ShardManager extends AbstractUntypedActor { private final String shardName; private final ActorRef actor; private final ActorPath actorPath; - private final Map peerAddresses; + private final Map peerAddresses; private ShardInformation(String shardName, ActorRef actor, - Map peerAddresses) { + Map peerAddresses) { this.shardName = shardName; this.actor = actor; this.actorPath = actor.path(); @@ -302,16 +320,15 @@ public class ShardManager extends AbstractUntypedActor { return actorPath; } - public Map getPeerAddresses() { - return peerAddresses; - } - - public void updatePeerAddress(String peerId, String peerAddress){ - LOG.info("updatePeerAddress for peer {} with address {}", peerId, peerAddress); + public void updatePeerAddress(ShardIdentifier peerId, String peerAddress){ + LOG.info("updatePeerAddress for peer {} with address {}", peerId, + peerAddress); if(peerAddresses.containsKey(peerId)){ peerAddresses.put(peerId, peerAddress); - LOG.info("Sending PeerAddressResolved for peer {} with address {} to {}", peerId, peerAddress, actor.path()); + LOG.debug( + "Sending PeerAddressResolved for peer {} with address {} to {}", + peerId, peerAddress, actor.path()); actor .tell(new PeerAddressResolved(peerId, peerAddress), @@ -321,3 +338,6 @@ public class ShardManager extends AbstractUntypedActor { } } } + + +