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=3396eb556456116005459540c66a78119ca64767;hp=64c6821120f94f99a389c12700757a7b8c7266f5;hb=d206d27042eef2185c875f85cf6eac61a1bd77c4;hpb=b584e686fdeba863643f80c0894d7fbd2dcaa540 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..3396eb5564 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; @@ -26,8 +30,11 @@ import org.opendaylight.controller.cluster.datastore.messages.PeerAddressResolve import org.opendaylight.controller.cluster.datastore.messages.PrimaryFound; import org.opendaylight.controller.cluster.datastore.messages.PrimaryNotFound; import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext; +import org.opendaylight.controller.md.sal.dom.store.impl.InMemoryDOMDataStoreConfigProperties; + import scala.concurrent.duration.Duration; +import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -61,15 +68,21 @@ public class ShardManager extends AbstractUntypedActor { private final Configuration configuration; + private ShardManagerInfoMBean mBean; + + private final InMemoryDOMDataStoreConfigProperties dataStoreProperties; + /** * @param type defines the kind of data that goes into shards created by this shard manager. Examples of type would be * configuration or operational */ - private ShardManager(String type, ClusterWrapper cluster, Configuration configuration) { + private ShardManager(String type, ClusterWrapper cluster, Configuration configuration, + InMemoryDOMDataStoreConfigProperties dataStoreProperties) { this.type = Preconditions.checkNotNull(type, "type should not be null"); this.cluster = Preconditions.checkNotNull(cluster, "cluster should not be null"); this.configuration = Preconditions.checkNotNull(configuration, "configuration should not be null"); + this.dataStoreProperties = dataStoreProperties; // Subscribe this actor to cluster member events cluster.subscribeToMemberEvents(getSelf()); @@ -81,12 +94,18 @@ public class ShardManager extends AbstractUntypedActor { public static Props props(final String type, final ClusterWrapper cluster, - final Configuration configuration) { + final Configuration configuration, + final InMemoryDOMDataStoreConfigProperties dataStoreProperties) { + + 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 public ShardManager create() throws Exception { - return new ShardManager(type, cluster, configuration); + return new ShardManager(type, cluster, configuration, dataStoreProperties); } }); } @@ -108,7 +127,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 +141,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 +156,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 +175,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 +188,9 @@ public class ShardManager extends AbstractUntypedActor { } } + List members = + configuration.getMembersFromShardName(shardName); + if(cluster.getCurrentMemberName() != null) { members.remove(cluster.getCurrentMemberName()); } @@ -196,9 +212,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 +231,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 +245,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, dataStoreProperties), + shardId.toString()); + localShardActorNames.add(shardId.toString()); localShards.put(shardName, new ShardInformation(shardName, actor, peerAddresses)); } + mBean = ShardManagerInfo + .createShardManagerMBean("shard-manager-" + this.type, localShardActorNames); + } /** @@ -242,9 +267,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,22 +278,24 @@ 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"), new Function() { @Override public SupervisorStrategy.Directive apply(Throwable t) { + LOG.warning("Supervisor Strategy of resume applied {}",t); return SupervisorStrategy.resume(); } } @@ -280,10 +307,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 +329,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 +347,6 @@ public class ShardManager extends AbstractUntypedActor { } } } + + +