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.Props;
15 import akka.event.Logging;
16 import akka.event.LoggingAdapter;
17 import akka.japi.Creator;
18 import org.opendaylight.controller.cluster.datastore.messages.FindPrimary;
19 import org.opendaylight.controller.cluster.datastore.messages.PrimaryFound;
20 import org.opendaylight.controller.cluster.datastore.messages.PrimaryNotFound;
21 import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext;
23 import java.util.HashMap;
24 import java.util.List;
28 * The ShardManager has the following jobs,
30 * <li> Create all the local shard replicas that belong on this cluster member
31 * <li> Find the primary replica for any given shard
32 * <li> Engage in shard replica elections which decide which replica should be the primary
35 * <h3>>Creation of Shard replicas</h3
37 * When the ShardManager is constructed it reads the cluster configuration to find out which shard replicas
38 * belong on this member. It finds out the name of the current cluster member from the Akka Clustering Service.
41 * <h3> Replica Elections </h3>
44 * The Shard Manager uses multiple cues to initiate election.
45 * <li> When a member of the cluster dies
46 * <li> When a local shard replica dies
47 * <li> When a local shard replica comes alive
50 public class ShardManager extends AbstractUntypedActor {
52 // Stores a mapping between a shard name and the address of the current primary
53 private final Map<String, Address> shardNameToPrimaryAddress =
56 // Stores a mapping between a member name and the address of the member
57 private final Map<String, Address> memberNameToAddress = new HashMap<>();
59 // Stores a mapping between the shard name and all the members on which a replica of that shard are available
60 private final Map<String, List<String>> shardNameToMembers =
63 private final LoggingAdapter log =
64 Logging.getLogger(getContext().system(), this);
67 private final Map<String, ActorPath> localShards = new HashMap<>();
70 private final ClusterWrapper cluster;
73 * @param type defines the kind of data that goes into shards created by this shard manager. Examples of type would be
74 * configuration or operational
76 private ShardManager(String type, ClusterWrapper cluster, Configuration configuration) {
77 this.cluster = cluster;
78 String memberName = cluster.getCurrentMemberName();
79 List<String> memberShardNames =
80 configuration.getMemberShardNames(memberName);
82 for(String shardName : memberShardNames){
83 ActorRef actor = getContext()
84 .actorOf(Shard.props(memberName + "-shard-" + shardName + "-" + type),
85 memberName + "-shard-" + shardName + "-" + type);
86 ActorPath path = actor.path();
87 localShards.put(shardName, path);
91 public static Props props(final String type,
92 final ClusterWrapper cluster,
93 final Configuration configuration) {
94 return Props.create(new Creator<ShardManager>() {
97 public ShardManager create() throws Exception {
98 return new ShardManager(type, cluster, configuration);
104 public void handleReceive(Object message) throws Exception {
105 if (message instanceof FindPrimary) {
106 FindPrimary msg = ((FindPrimary) message);
107 String shardName = msg.getShardName();
110 if (Shard.DEFAULT_NAME.equals(shardName)) {
111 ActorPath defaultShardPath = localShards.get(shardName);
112 if(defaultShardPath == null){
113 throw new IllegalStateException("local default shard not found");
115 getSender().tell(new PrimaryFound(defaultShardPath.toString()),
118 getSender().tell(new PrimaryNotFound(shardName), getSelf());
120 } else if (message instanceof UpdateSchemaContext) {
121 for(ActorPath path : localShards.values()){
122 getContext().system().actorSelection(path)