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 scala.concurrent.duration.Duration;
35 import java.util.ArrayList;
36 import java.util.HashMap;
37 import java.util.List;
41 * The ShardManager has the following jobs,
43 * <li> Create all the local shard replicas that belong on this cluster member
44 * <li> Find the address of the local shard
45 * <li> Find the primary replica for any given shard
46 * <li> Monitor the cluster members and store their addresses
49 public class ShardManager extends AbstractUntypedActor {
51 // Stores a mapping between a member name and the address of the member
52 // Member names look like "member-1", "member-2" etc and are as specified
54 private final Map<String, Address> memberNameToAddress = new HashMap<>();
56 // Stores a mapping between a shard name and it's corresponding information
57 // Shard names look like inventory, topology etc and are as specified in
59 private final Map<String, ShardInformation> localShards = new HashMap<>();
61 // The type of a ShardManager reflects the type of the datastore itself
62 // A data store could be of type config/operational
63 private final String type;
65 private final ClusterWrapper cluster;
67 private final Configuration configuration;
69 private ShardManagerInfoMBean mBean;
72 * @param type defines the kind of data that goes into shards created by this shard manager. Examples of type would be
73 * configuration or operational
75 private ShardManager(String type, ClusterWrapper cluster, Configuration configuration) {
77 this.type = Preconditions.checkNotNull(type, "type should not be null");
78 this.cluster = Preconditions.checkNotNull(cluster, "cluster should not be null");
79 this.configuration = Preconditions.checkNotNull(configuration, "configuration should not be null");
81 // Subscribe this actor to cluster member events
82 cluster.subscribeToMemberEvents(getSelf());
84 // Create all the local Shards and make them a child of the ShardManager
85 // TODO: This may need to be initiated when we first get the schema context
89 public static Props props(final String type,
90 final ClusterWrapper cluster,
91 final Configuration configuration) {
93 Preconditions.checkNotNull(type, "type should not be null");
94 Preconditions.checkNotNull(cluster, "cluster should not be null");
95 Preconditions.checkNotNull(configuration, "configuration should not be null");
97 return Props.create(new Creator<ShardManager>() {
100 public ShardManager create() throws Exception {
101 return new ShardManager(type, cluster, configuration);
108 public void handleReceive(Object message) throws Exception {
109 if (message.getClass().equals(FindPrimary.SERIALIZABLE_CLASS)) {
111 FindPrimary.fromSerializable(message));
112 } else if(message instanceof FindLocalShard){
113 findLocalShard((FindLocalShard) message);
114 } else if (message instanceof UpdateSchemaContext) {
115 updateSchemaContext(message);
116 } else if (message instanceof ClusterEvent.MemberUp){
117 memberUp((ClusterEvent.MemberUp) message);
118 } else if(message instanceof ClusterEvent.MemberRemoved) {
119 memberRemoved((ClusterEvent.MemberRemoved) message);
120 } else if(message instanceof ClusterEvent.UnreachableMember) {
121 ignoreMessage(message);
123 unknownMessage(message);
128 private void findLocalShard(FindLocalShard message) {
129 ShardInformation shardInformation =
130 localShards.get(message.getShardName());
132 if(shardInformation != null){
133 getSender().tell(new LocalShardFound(shardInformation.getActor()), getSelf());
137 getSender().tell(new LocalShardNotFound(message.getShardName()),
141 private void memberRemoved(ClusterEvent.MemberRemoved message) {
142 memberNameToAddress.remove(message.member().roles().head());
145 private void memberUp(ClusterEvent.MemberUp message) {
146 String memberName = message.member().roles().head();
148 memberNameToAddress.put(memberName , message.member().address());
150 for(ShardInformation info : localShards.values()){
151 String shardName = info.getShardName();
152 info.updatePeerAddress(getShardIdentifier(memberName, shardName),
153 getShardActorPath(shardName, memberName));
158 * Notifies all the local shards of a change in the schema context
162 private void updateSchemaContext(Object message) {
163 for(ShardInformation info : localShards.values()){
164 info.getActor().tell(message,getSelf());
168 private void findPrimary(FindPrimary message) {
169 String shardName = message.getShardName();
171 // First see if the there is a local replica for the shard
172 ShardInformation info = localShards.get(shardName);
174 ActorPath shardPath = info.getActorPath();
175 if (shardPath != null) {
178 new PrimaryFound(shardPath.toString()).toSerializable(),
184 List<String> members =
185 configuration.getMembersFromShardName(shardName);
187 if(cluster.getCurrentMemberName() != null) {
188 members.remove(cluster.getCurrentMemberName());
191 // There is no way for us to figure out the primary (for now) so assume
192 // that one of the remote nodes is a primary
193 for(String memberName : members) {
194 Address address = memberNameToAddress.get(memberName);
197 getShardActorPath(shardName, memberName);
198 getSender().tell(new PrimaryFound(path).toSerializable(), getSelf());
202 getSender().tell(new PrimaryNotFound(shardName).toSerializable(), getSelf());
205 private String getShardActorPath(String shardName, String memberName) {
206 Address address = memberNameToAddress.get(memberName);
207 if(address != null) {
208 StringBuilder builder = new StringBuilder();
209 builder.append(address.toString())
211 .append(ShardManagerIdentifier.builder().type(type).build().toString())
213 .append(getShardIdentifier(memberName, shardName));
214 return builder.toString();
220 * Construct the name of the shard actor given the name of the member on
221 * which the shard resides and the name of the shard
227 private ShardIdentifier getShardIdentifier(String memberName, String shardName){
228 return ShardIdentifier.builder().memberName(memberName).shardName(shardName).type(type).build();
232 * Create shards that are local to the member on which the ShardManager
236 private void createLocalShards() {
237 String memberName = this.cluster.getCurrentMemberName();
238 List<String> memberShardNames =
239 this.configuration.getMemberShardNames(memberName);
241 List<String> localShardActorNames = new ArrayList<>();
242 for(String shardName : memberShardNames){
243 ShardIdentifier shardId = getShardIdentifier(memberName, shardName);
244 Map<ShardIdentifier, String> peerAddresses = getPeerAddresses(shardName);
245 ActorRef actor = getContext()
246 .actorOf(Shard.props(shardId, peerAddresses),
248 localShardActorNames.add(shardId.toString());
249 localShards.put(shardName, new ShardInformation(shardName, actor, peerAddresses));
252 mBean = ShardManagerInfo
253 .createShardManagerMBean("shard-manager-" + this.type, localShardActorNames);
258 * Given the name of the shard find the addresses of all it's peers
263 private Map<ShardIdentifier, String> getPeerAddresses(String shardName){
265 Map<ShardIdentifier, String> peerAddresses = new HashMap<>();
267 List<String> members =
268 this.configuration.getMembersFromShardName(shardName);
270 String currentMemberName = this.cluster.getCurrentMemberName();
272 for(String memberName : members){
273 if(!currentMemberName.equals(memberName)){
274 ShardIdentifier shardId = getShardIdentifier(memberName,
277 getShardActorPath(shardName, currentMemberName);
278 peerAddresses.put(shardId, path);
281 return peerAddresses;
285 public SupervisorStrategy supervisorStrategy() {
286 return new OneForOneStrategy(10, Duration.create("1 minute"),
287 new Function<Throwable, SupervisorStrategy.Directive>() {
289 public SupervisorStrategy.Directive apply(Throwable t) {
290 return SupervisorStrategy.resume();
297 private class ShardInformation {
298 private final String shardName;
299 private final ActorRef actor;
300 private final ActorPath actorPath;
301 private final Map<ShardIdentifier, String> peerAddresses;
303 private ShardInformation(String shardName, ActorRef actor,
304 Map<ShardIdentifier, String> peerAddresses) {
305 this.shardName = shardName;
307 this.actorPath = actor.path();
308 this.peerAddresses = peerAddresses;
311 public String getShardName() {
315 public ActorRef getActor(){
319 public ActorPath getActorPath() {
323 public void updatePeerAddress(ShardIdentifier peerId, String peerAddress){
324 LOG.info("updatePeerAddress for peer {} with address {}", peerId,
326 if(peerAddresses.containsKey(peerId)){
327 peerAddresses.put(peerId, peerAddress);
330 "Sending PeerAddressResolved for peer {} with address {} to {}",
331 peerId, peerAddress, actor.path());
334 .tell(new PeerAddressResolved(peerId, peerAddress),