2 * Copyright (c) 2014 Cisco Systems, Inc. and others. All rights reserved.
4 * This program and the accompanying materials are made available under the
5 * terms of the Eclipse Public License v1.0 which accompanies this distribution,
6 * and is available at http://www.eclipse.org/legal/epl-v10.html
9 package org.opendaylight.controller.cluster.datastore;
11 import akka.actor.ActorPath;
12 import akka.actor.ActorRef;
13 import akka.actor.Address;
14 import akka.actor.OneForOneStrategy;
15 import akka.actor.Props;
16 import akka.actor.SupervisorStrategy;
17 import akka.cluster.ClusterEvent;
18 import akka.japi.Creator;
19 import akka.japi.Function;
20 import com.google.common.base.Preconditions;
21 import org.opendaylight.controller.cluster.datastore.identifiers.ShardIdentifier;
22 import org.opendaylight.controller.cluster.datastore.identifiers.ShardManagerIdentifier;
23 import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shardmanager.ShardManagerInfo;
24 import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shardmanager.ShardManagerInfoMBean;
25 import org.opendaylight.controller.cluster.datastore.messages.FindLocalShard;
26 import org.opendaylight.controller.cluster.datastore.messages.FindPrimary;
27 import org.opendaylight.controller.cluster.datastore.messages.LocalShardFound;
28 import org.opendaylight.controller.cluster.datastore.messages.LocalShardNotFound;
29 import org.opendaylight.controller.cluster.datastore.messages.PeerAddressResolved;
30 import org.opendaylight.controller.cluster.datastore.messages.PrimaryFound;
31 import org.opendaylight.controller.cluster.datastore.messages.PrimaryNotFound;
32 import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext;
33 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
34 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
35 import scala.concurrent.duration.Duration;
37 import java.util.ArrayList;
38 import java.util.HashMap;
39 import java.util.List;
43 * The ShardManager has the following jobs,
45 * <li> Create all the local shard replicas that belong on this cluster member
46 * <li> Find the address of the local shard
47 * <li> Find the primary replica for any given shard
48 * <li> Monitor the cluster members and store their addresses
51 public class ShardManager extends AbstractUntypedActor {
53 // Stores a mapping between a member name and the address of the member
54 // Member names look like "member-1", "member-2" etc and are as specified
56 private final Map<String, Address> memberNameToAddress = new HashMap<>();
58 // Stores a mapping between a shard name and it's corresponding information
59 // Shard names look like inventory, topology etc and are as specified in
61 private final Map<String, ShardInformation> localShards = new HashMap<>();
63 // The type of a ShardManager reflects the type of the datastore itself
64 // A data store could be of type config/operational
65 private final String type;
67 private final ClusterWrapper cluster;
69 private final Configuration configuration;
71 private ShardManagerInfoMBean mBean;
73 private final DatastoreContext datastoreContext;
76 * @param type defines the kind of data that goes into shards created by this shard manager. Examples of type would be
77 * configuration or operational
79 private ShardManager(String type, ClusterWrapper cluster, Configuration configuration,
80 DatastoreContext datastoreContext) {
82 this.type = Preconditions.checkNotNull(type, "type should not be null");
83 this.cluster = Preconditions.checkNotNull(cluster, "cluster should not be null");
84 this.configuration = Preconditions.checkNotNull(configuration, "configuration should not be null");
85 this.datastoreContext = datastoreContext;
87 // Subscribe this actor to cluster member events
88 cluster.subscribeToMemberEvents(getSelf());
90 //createLocalShards(null);
93 public static Props props(final String type,
94 final ClusterWrapper cluster,
95 final Configuration configuration,
96 final DatastoreContext datastoreContext) {
98 Preconditions.checkNotNull(type, "type should not be null");
99 Preconditions.checkNotNull(cluster, "cluster should not be null");
100 Preconditions.checkNotNull(configuration, "configuration should not be null");
102 return Props.create(new ShardManagerCreator(type, cluster, configuration, datastoreContext));
106 public void handleReceive(Object message) throws Exception {
107 if (message.getClass().equals(FindPrimary.SERIALIZABLE_CLASS)) {
109 FindPrimary.fromSerializable(message));
110 } else if(message instanceof FindLocalShard){
111 findLocalShard((FindLocalShard) message);
112 } else if (message instanceof UpdateSchemaContext) {
113 updateSchemaContext(message);
114 } else if (message instanceof ClusterEvent.MemberUp){
115 memberUp((ClusterEvent.MemberUp) message);
116 } else if(message instanceof ClusterEvent.MemberRemoved) {
117 memberRemoved((ClusterEvent.MemberRemoved) message);
118 } else if(message instanceof ClusterEvent.UnreachableMember) {
119 ignoreMessage(message);
121 unknownMessage(message);
126 private void findLocalShard(FindLocalShard message) {
127 ShardInformation shardInformation =
128 localShards.get(message.getShardName());
130 if(shardInformation != null){
131 getSender().tell(new LocalShardFound(shardInformation.getActor()), getSelf());
135 getSender().tell(new LocalShardNotFound(message.getShardName()),
139 private void memberRemoved(ClusterEvent.MemberRemoved message) {
140 memberNameToAddress.remove(message.member().roles().head());
143 private void memberUp(ClusterEvent.MemberUp message) {
144 String memberName = message.member().roles().head();
146 memberNameToAddress.put(memberName , message.member().address());
148 for(ShardInformation info : localShards.values()){
149 String shardName = info.getShardName();
150 info.updatePeerAddress(getShardIdentifier(memberName, shardName),
151 getShardActorPath(shardName, memberName));
156 * Notifies all the local shards of a change in the schema context
160 private void updateSchemaContext(Object message) {
161 SchemaContext schemaContext = ((UpdateSchemaContext) message).getSchemaContext();
163 if(localShards.size() == 0){
164 createLocalShards(schemaContext);
166 for (ShardInformation info : localShards.values()) {
167 info.getActor().tell(message, getSelf());
172 private void findPrimary(FindPrimary message) {
173 String shardName = message.getShardName();
175 // First see if the there is a local replica for the shard
176 ShardInformation info = localShards.get(shardName);
178 ActorPath shardPath = info.getActorPath();
179 if (shardPath != null) {
182 new PrimaryFound(shardPath.toString()).toSerializable(),
188 List<String> members =
189 configuration.getMembersFromShardName(shardName);
191 if(cluster.getCurrentMemberName() != null) {
192 members.remove(cluster.getCurrentMemberName());
195 // There is no way for us to figure out the primary (for now) so assume
196 // that one of the remote nodes is a primary
197 for(String memberName : members) {
198 Address address = memberNameToAddress.get(memberName);
201 getShardActorPath(shardName, memberName);
202 getSender().tell(new PrimaryFound(path).toSerializable(), getSelf());
206 getSender().tell(new PrimaryNotFound(shardName).toSerializable(), getSelf());
209 private String getShardActorPath(String shardName, String memberName) {
210 Address address = memberNameToAddress.get(memberName);
211 if(address != null) {
212 StringBuilder builder = new StringBuilder();
213 builder.append(address.toString())
215 .append(ShardManagerIdentifier.builder().type(type).build().toString())
217 .append(getShardIdentifier(memberName, shardName));
218 return builder.toString();
224 * Construct the name of the shard actor given the name of the member on
225 * which the shard resides and the name of the shard
231 private ShardIdentifier getShardIdentifier(String memberName, String shardName){
232 return ShardIdentifier.builder().memberName(memberName).shardName(shardName).type(type).build();
236 * Create shards that are local to the member on which the ShardManager
240 private void createLocalShards(SchemaContext schemaContext) {
241 String memberName = this.cluster.getCurrentMemberName();
242 List<String> memberShardNames =
243 this.configuration.getMemberShardNames(memberName);
245 List<String> localShardActorNames = new ArrayList<>();
246 for(String shardName : memberShardNames){
247 ShardIdentifier shardId = getShardIdentifier(memberName, shardName);
248 Map<ShardIdentifier, String> peerAddresses = getPeerAddresses(shardName);
249 ActorRef actor = getContext()
250 .actorOf(Shard.props(shardId, peerAddresses, datastoreContext, schemaContext).
251 withMailbox(ActorContext.MAILBOX), shardId.toString());
252 localShardActorNames.add(shardId.toString());
253 localShards.put(shardName, new ShardInformation(shardName, actor, peerAddresses));
256 mBean = ShardManagerInfo.createShardManagerMBean("shard-manager-" + this.type,
257 datastoreContext.getDataStoreMXBeanType(), localShardActorNames);
261 * Given the name of the shard find the addresses of all it's peers
266 private Map<ShardIdentifier, String> getPeerAddresses(String shardName){
268 Map<ShardIdentifier, String> peerAddresses = new HashMap<>();
270 List<String> members =
271 this.configuration.getMembersFromShardName(shardName);
273 String currentMemberName = this.cluster.getCurrentMemberName();
275 for(String memberName : members){
276 if(!currentMemberName.equals(memberName)){
277 ShardIdentifier shardId = getShardIdentifier(memberName,
280 getShardActorPath(shardName, currentMemberName);
281 peerAddresses.put(shardId, path);
284 return peerAddresses;
288 public SupervisorStrategy supervisorStrategy() {
290 return new OneForOneStrategy(10, Duration.create("1 minute"),
291 new Function<Throwable, SupervisorStrategy.Directive>() {
293 public SupervisorStrategy.Directive apply(Throwable t) {
294 StringBuilder sb = new StringBuilder();
295 for(StackTraceElement element : t.getStackTrace()) {
297 .append(element.toString());
299 LOG.warning("Supervisor Strategy of resume applied {}",sb.toString());
300 return SupervisorStrategy.resume();
307 private class ShardInformation {
308 private final String shardName;
309 private final ActorRef actor;
310 private final ActorPath actorPath;
311 private final Map<ShardIdentifier, String> peerAddresses;
313 private ShardInformation(String shardName, ActorRef actor,
314 Map<ShardIdentifier, String> peerAddresses) {
315 this.shardName = shardName;
317 this.actorPath = actor.path();
318 this.peerAddresses = peerAddresses;
321 public String getShardName() {
325 public ActorRef getActor(){
329 public ActorPath getActorPath() {
333 public void updatePeerAddress(ShardIdentifier peerId, String peerAddress){
334 LOG.info("updatePeerAddress for peer {} with address {}", peerId,
336 if(peerAddresses.containsKey(peerId)){
337 peerAddresses.put(peerId, peerAddress);
340 "Sending PeerAddressResolved for peer {} with address {} to {}",
341 peerId, peerAddress, actor.path());
344 .tell(new PeerAddressResolved(peerId, peerAddress),
351 private static class ShardManagerCreator implements Creator<ShardManager> {
352 private static final long serialVersionUID = 1L;
355 final ClusterWrapper cluster;
356 final Configuration configuration;
357 final DatastoreContext datastoreContext;
359 ShardManagerCreator(String type, ClusterWrapper cluster,
360 Configuration configuration, DatastoreContext datastoreContext) {
362 this.cluster = cluster;
363 this.configuration = configuration;
364 this.datastoreContext = datastoreContext;
368 public ShardManager create() throws Exception {
369 return new ShardManager(type, cluster, configuration, datastoreContext);