Code Review
/
controller.git
/ blobdiff
commit
grep
author
committer
pickaxe
?
search:
re
summary
|
shortlog
|
log
|
commit
|
commitdiff
|
review
|
tree
raw
|
inline
| side by side
Merge "Fix to karaf-parent to copy in dependencies."
[controller.git]
/
opendaylight
/
md-sal
/
sal-distributed-datastore
/
src
/
main
/
java
/
org
/
opendaylight
/
controller
/
cluster
/
datastore
/
ShardManager.java
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 22e2dbd47d4d7148c8e41cfe05ce8532df527466..775cae35e22843cafd223bce22eb5232a230868f 100644
(file)
--- 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
@@
-15,8
+15,6
@@
import akka.actor.OneForOneStrategy;
import akka.actor.Props;
import akka.actor.SupervisorStrategy;
import akka.cluster.ClusterEvent;
import akka.actor.Props;
import akka.actor.SupervisorStrategy;
import akka.cluster.ClusterEvent;
-import akka.event.Logging;
-import akka.event.LoggingAdapter;
import akka.japi.Creator;
import akka.japi.Function;
import akka.japi.Procedure;
import akka.japi.Creator;
import akka.japi.Function;
import akka.japi.Procedure;
@@
-41,7
+39,6
@@
import org.opendaylight.controller.cluster.common.actor.AbstractUntypedPersisten
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.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.ActorInitialized;
import org.opendaylight.controller.cluster.datastore.messages.ActorNotInitialized;
import org.opendaylight.controller.cluster.datastore.messages.FindLocalShard;
import org.opendaylight.controller.cluster.datastore.messages.ActorInitialized;
import org.opendaylight.controller.cluster.datastore.messages.ActorNotInitialized;
import org.opendaylight.controller.cluster.datastore.messages.FindLocalShard;
@@
-52,8
+49,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.cluster.datastore.messages.PrimaryFound;
import org.opendaylight.controller.cluster.datastore.messages.PrimaryNotFound;
import org.opendaylight.controller.cluster.datastore.messages.UpdateSchemaContext;
+import org.opendaylight.controller.cluster.datastore.utils.Dispatchers;
import org.opendaylight.yangtools.yang.model.api.ModuleIdentifier;
import org.opendaylight.yangtools.yang.model.api.SchemaContext;
import org.opendaylight.yangtools.yang.model.api.ModuleIdentifier;
import org.opendaylight.yangtools.yang.model.api.SchemaContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import scala.concurrent.duration.Duration;
/**
import scala.concurrent.duration.Duration;
/**
@@
-67,8
+67,7
@@
import scala.concurrent.duration.Duration;
*/
public class ShardManager extends AbstractUntypedPersistentActorWithMetering {
*/
public class ShardManager extends AbstractUntypedPersistentActorWithMetering {
- protected final LoggingAdapter LOG =
- Logging.getLogger(getContext().system(), this);
+ private final Logger LOG = LoggerFactory.getLogger(getClass());
// Stores a mapping between a member name and the address of the member
// Member names look like "member-1", "member-2" etc and are as specified
// Stores a mapping between a member name and the address of the member
// Member names look like "member-1", "member-2" etc and are as specified
@@
-88,26
+87,28
@@
public class ShardManager extends AbstractUntypedPersistentActorWithMetering {
private final Configuration configuration;
private final Configuration configuration;
- private
ShardManagerInfoMBean mBean
;
+ private
final String shardDispatcherPath
;
- private final DatastoreContext datastoreContext;
+ private ShardManagerInfo mBean;
+
+ private DatastoreContext datastoreContext;
private Collection<String> knownModules = Collections.emptySet();
private final DataPersistenceProvider dataPersistenceProvider;
/**
private Collection<String> knownModules = Collections.emptySet();
private final DataPersistenceProvider dataPersistenceProvider;
/**
- * @param type defines the kind of data that goes into shards created by this shard manager. Examples of type would be
- * configuration or operational
*/
*/
- protected ShardManager(
String type,
ClusterWrapper cluster, Configuration configuration,
+ protected ShardManager(ClusterWrapper cluster, Configuration configuration,
DatastoreContext datastoreContext) {
DatastoreContext datastoreContext) {
- 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.datastoreContext = datastoreContext;
this.dataPersistenceProvider = createDataPersistenceProvider(datastoreContext.isPersistent());
this.cluster = Preconditions.checkNotNull(cluster, "cluster should not be null");
this.configuration = Preconditions.checkNotNull(configuration, "configuration should not be null");
this.datastoreContext = datastoreContext;
this.dataPersistenceProvider = createDataPersistenceProvider(datastoreContext.isPersistent());
+ this.type = datastoreContext.getDataStoreType();
+ this.shardDispatcherPath =
+ new Dispatchers(context().system().dispatchers()).getDispatcherPath(Dispatchers.DispatcherType.Shard);
// Subscribe this actor to cluster member events
cluster.subscribeToMemberEvents(getSelf());
// Subscribe this actor to cluster member events
cluster.subscribeToMemberEvents(getSelf());
@@
-119,16
+120,22
@@
public class ShardManager extends AbstractUntypedPersistentActorWithMetering {
return (persistent) ? new PersistentDataProvider() : new NonPersistentDataProvider();
}
return (persistent) ? new PersistentDataProvider() : new NonPersistentDataProvider();
}
- public static Props props(
final String type,
+ public static Props props(
final ClusterWrapper cluster,
final Configuration configuration,
final DatastoreContext datastoreContext) {
final ClusterWrapper cluster,
final Configuration configuration,
final DatastoreContext datastoreContext) {
- Preconditions.checkNotNull(type, "type should not be null");
Preconditions.checkNotNull(cluster, "cluster should not be null");
Preconditions.checkNotNull(configuration, "configuration should not be null");
Preconditions.checkNotNull(cluster, "cluster should not be null");
Preconditions.checkNotNull(configuration, "configuration should not be null");
- return Props.create(new ShardManagerCreator(type, cluster, configuration, datastoreContext));
+ return Props.create(new ShardManagerCreator(cluster, configuration, datastoreContext));
+ }
+
+ @Override
+ public void postStop() {
+ LOG.info("Stopping ShardManager");
+
+ mBean.unregisterMBean();
}
@Override
}
@Override
@@
-147,6
+154,8
@@
public class ShardManager extends AbstractUntypedPersistentActorWithMetering {
memberRemoved((ClusterEvent.MemberRemoved) message);
} else if(message instanceof ClusterEvent.UnreachableMember) {
ignoreMessage(message);
memberRemoved((ClusterEvent.MemberRemoved) message);
} else if(message instanceof ClusterEvent.UnreachableMember) {
ignoreMessage(message);
+ } else if(message instanceof DatastoreContext) {
+ onDatastoreContext((DatastoreContext)message);
} else{
unknownMessage(message);
}
} else{
unknownMessage(message);
}
@@
-186,7
+195,7
@@
public class ShardManager extends AbstractUntypedPersistentActorWithMetering {
knownModules = ImmutableSet.copyOf(msg.getModules());
} else if (message instanceof RecoveryFailure) {
RecoveryFailure failure = (RecoveryFailure) message;
knownModules = ImmutableSet.copyOf(msg.getModules());
} else if (message instanceof RecoveryFailure) {
RecoveryFailure failure = (RecoveryFailure) message;
- LOG.error(
failure.cause(), "Recovery failed"
);
+ LOG.error(
"Recovery failed", failure.cause()
);
} else if (message instanceof RecoveryCompleted) {
LOG.info("Recovery complete : {}", persistenceId());
} else if (message instanceof RecoveryCompleted) {
LOG.info("Recovery complete : {}", persistenceId());
@@
-257,6
+266,15
@@
public class ShardManager extends AbstractUntypedPersistentActorWithMetering {
}
}
}
}
+ private void onDatastoreContext(DatastoreContext context) {
+ datastoreContext = context;
+ for (ShardInformation info : localShards.values()) {
+ if (info.getActor() != null) {
+ info.getActor().tell(datastoreContext, getSelf());
+ }
+ }
+ }
+
/**
* Notifies all the local shards of a change in the schema context
*
/**
* Notifies all the local shards of a change in the schema context
*
@@
-275,7
+293,7
@@
public class ShardManager extends AbstractUntypedPersistentActorWithMetering {
if(newModules.containsAll(knownModules)) {
if(newModules.containsAll(knownModules)) {
- LOG.
info
("New SchemaContext has a super set of current knownModules - persisting info");
+ LOG.
debug
("New SchemaContext has a super set of current knownModules - persisting info");
knownModules = ImmutableSet.copyOf(newModules);
knownModules = ImmutableSet.copyOf(newModules);
@@
-283,12
+301,12
@@
public class ShardManager extends AbstractUntypedPersistentActorWithMetering {
@Override
public void apply(SchemaContextModules param) throws Exception {
@Override
public void apply(SchemaContextModules param) throws Exception {
- LOG.
info
("Sending new SchemaContext to Shards");
+ LOG.
debug
("Sending new SchemaContext to Shards");
for (ShardInformation info : localShards.values()) {
if (info.getActor() == null) {
info.setActor(getContext().actorOf(Shard.props(info.getShardId(),
for (ShardInformation info : localShards.values()) {
if (info.getActor() == null) {
info.setActor(getContext().actorOf(Shard.props(info.getShardId(),
- info.getPeerAddresses(), datastoreContext, schemaContext)
,
- info.getShardId().toString()));
+ info.getPeerAddresses(), datastoreContext, schemaContext)
+
.withDispatcher(shardDispatcherPath),
info.getShardId().toString()));
} else {
info.getActor().tell(message, getSelf());
}
} else {
info.getActor().tell(message, getSelf());
}
@@
-297,7
+315,8
@@
public class ShardManager extends AbstractUntypedPersistentActorWithMetering {
});
} else {
});
} else {
- LOG.info("Rejecting schema context update because it is not a super set of previously known modules");
+ LOG.debug("Rejecting schema context update - not a super set of previously known modules:\nUPDATE: {}\nKNOWN: {}",
+ newModules, knownModules);
}
}
}
}
@@
-424,12
+443,7
@@
public class ShardManager extends AbstractUntypedPersistentActorWithMetering {
new Function<Throwable, SupervisorStrategy.Directive>() {
@Override
public SupervisorStrategy.Directive apply(Throwable t) {
new Function<Throwable, SupervisorStrategy.Directive>() {
@Override
public SupervisorStrategy.Directive apply(Throwable t) {
- StringBuilder sb = new StringBuilder();
- for(StackTraceElement element : t.getStackTrace()) {
- sb.append("\n\tat ")
- .append(element.toString());
- }
- LOG.warning("Supervisor Strategy of resume applied {}",sb.toString());
+ LOG.warn("Supervisor Strategy caught unexpected exception - resuming", t);
return SupervisorStrategy.resume();
}
}
return SupervisorStrategy.resume();
}
}
@@
-535,14
+549,12
@@
public class ShardManager extends AbstractUntypedPersistentActorWithMetering {
private static class ShardManagerCreator implements Creator<ShardManager> {
private static final long serialVersionUID = 1L;
private static class ShardManagerCreator implements Creator<ShardManager> {
private static final long serialVersionUID = 1L;
- final String type;
final ClusterWrapper cluster;
final Configuration configuration;
final DatastoreContext datastoreContext;
final ClusterWrapper cluster;
final Configuration configuration;
final DatastoreContext datastoreContext;
- ShardManagerCreator(
String type,
ClusterWrapper cluster,
+ ShardManagerCreator(ClusterWrapper cluster,
Configuration configuration, DatastoreContext datastoreContext) {
Configuration configuration, DatastoreContext datastoreContext) {
- this.type = type;
this.cluster = cluster;
this.configuration = configuration;
this.datastoreContext = datastoreContext;
this.cluster = cluster;
this.configuration = configuration;
this.datastoreContext = datastoreContext;
@@
-550,7
+562,7
@@
public class ShardManager extends AbstractUntypedPersistentActorWithMetering {
@Override
public ShardManager create() throws Exception {
@Override
public ShardManager create() throws Exception {
- return new ShardManager(
type,
cluster, configuration, datastoreContext);
+ return new ShardManager(cluster, configuration, datastoreContext);
}
}
}
}