X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2FShard.java;h=21d74a6e1a37d49ad9ea6e53b731014e4020f01b;hb=5dd83186d64a4b10214f32ec15f007dae016f5e6;hp=9cd758ba30fdb94e85cd1703d99a8e0c55a50a17;hpb=cfc94248aa44307e7dc9aaefcb6748c478f93138;p=controller.git diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/Shard.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/Shard.java index 9cd758ba30..21d74a6e1a 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/Shard.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/Shard.java @@ -12,8 +12,6 @@ import akka.actor.ActorRef; import akka.actor.ActorSelection; import akka.actor.Cancellable; import akka.actor.Props; -import akka.event.Logging; -import akka.event.LoggingAdapter; import akka.japi.Creator; import akka.persistence.RecoveryFailure; import akka.serialization.Serialization; @@ -101,9 +99,6 @@ public class Shard extends RaftActor { // The state of this Shard private final InMemoryDOMDataStore store; - private final LoggingAdapter LOG = - Logging.getLogger(getContext().system(), this); - /// The name of this shard private final ShardIdentifier name; @@ -148,7 +143,7 @@ public class Shard extends RaftActor { this.schemaContext = schemaContext; this.dataPersistenceProvider = (datastoreContext.isPersistent()) ? new PersistentDataProvider() : new NonPersistentRaftDataProvider(); - LOG.info("Shard created : {} persistent : {}", name, datastoreContext.isPersistent()); + LOG.info("Shard created : {}, persistent : {}", name, datastoreContext.isPersistent()); store = InMemoryDOMDataStoreFactory.create(name.toString(), null, datastoreContext.getDataStoreProperties()); @@ -166,7 +161,7 @@ public class Shard extends RaftActor { } commitCoordinator = new ShardCommitCoordinator(TimeUnit.SECONDS.convert(1, TimeUnit.MINUTES), - datastoreContext.getShardTransactionCommitQueueCapacity()); + datastoreContext.getShardTransactionCommitQueueCapacity(), LOG, name.toString()); transactionCommitTimeout = TimeUnit.MILLISECONDS.convert( datastoreContext.getShardTransactionCommitTimeoutInSeconds(), TimeUnit.SECONDS); @@ -216,13 +211,13 @@ public class Shard extends RaftActor { @Override public void onReceiveRecover(final Object message) throws Exception { if(LOG.isDebugEnabled()) { - LOG.debug("onReceiveRecover: Received message {} from {}", - message.getClass().toString(), - getSender()); + LOG.debug("{}: onReceiveRecover: Received message {} from {}", persistenceId(), + message.getClass().toString(), getSender()); } if (message instanceof RecoveryFailure){ - LOG.error(((RecoveryFailure) message).cause(), "Recovery failed because of this cause"); + LOG.error("{}: Recovery failed because of this cause", + persistenceId(), ((RecoveryFailure) message).cause()); // Even though recovery failed, we still need to finish our recovery, eg send the // ActorInitialized message and start the txCommitTimeoutCheckSchedule. @@ -234,10 +229,6 @@ public class Shard extends RaftActor { @Override public void onReceiveCommand(final Object message) throws Exception { - if(LOG.isDebugEnabled()) { - LOG.debug("onReceiveCommand: Received message {} from {}", message, getSender()); - } - if (message.getClass().equals(CreateTransaction.SERIALIZABLE_CLASS)) { handleCreateTransaction(message); } else if(message instanceof ForwardedReadyTransaction) { @@ -275,8 +266,8 @@ public class Shard extends RaftActor { if(cohortEntry != null) { long elapsed = System.currentTimeMillis() - cohortEntry.getLastAccessTime(); if(elapsed > transactionCommitTimeout) { - LOG.warning("Current transaction {} has timed out after {} ms - aborting", - cohortEntry.getTransactionID(), transactionCommitTimeout); + LOG.warn("{}: Current transaction {} has timed out after {} ms - aborting", + persistenceId(), cohortEntry.getTransactionID(), transactionCommitTimeout); doAbortTransaction(cohortEntry.getTransactionID(), null); } @@ -286,7 +277,7 @@ public class Shard extends RaftActor { private void handleCommitTransaction(final CommitTransaction commit) { final String transactionID = commit.getTransactionID(); - LOG.debug("Committing transaction {}", transactionID); + LOG.debug("{}: Committing transaction {}", persistenceId(), transactionID); // Get the current in-progress cohort entry in the commitCoordinator if it corresponds to // this transaction. @@ -295,8 +286,8 @@ public class Shard extends RaftActor { // We're not the current Tx - the Tx was likely expired b/c it took too long in // between the canCommit and commit messages. IllegalStateException ex = new IllegalStateException( - String.format("Cannot commit transaction %s - it is not the current transaction", - transactionID)); + String.format("%s: Cannot commit transaction %s - it is not the current transaction", + persistenceId(), transactionID)); LOG.error(ex.getMessage()); shardMBean.incrementFailedTransactionsCount(); getSender().tell(new akka.actor.Status.Failure(ex), getSelf()); @@ -322,9 +313,9 @@ public class Shard extends RaftActor { Shard.this.persistData(getSender(), transactionID, new ModificationPayload(cohortEntry.getModification())); } - } catch (InterruptedException | ExecutionException | IOException e) { - LOG.error(e, "An exception occurred while preCommitting transaction {}", - cohortEntry.getTransactionID()); + } catch (Exception e) { + LOG.error("{} An exception occurred while preCommitting transaction {}", + persistenceId(), cohortEntry.getTransactionID(), e); shardMBean.incrementFailedTransactionsCount(); getSender().tell(new akka.actor.Status.Failure(e), getSelf()); } @@ -352,8 +343,8 @@ public class Shard extends RaftActor { // This really shouldn't happen - it likely means that persistence or replication // took so long to complete such that the cohort entry was expired from the cache. IllegalStateException ex = new IllegalStateException( - String.format("Could not finish committing transaction %s - no CohortEntry found", - transactionID)); + String.format("%s: Could not finish committing transaction %s - no CohortEntry found", + persistenceId(), transactionID)); LOG.error(ex.getMessage()); sender.tell(new akka.actor.Status.Failure(ex), getSelf()); } @@ -361,7 +352,7 @@ public class Shard extends RaftActor { return; } - LOG.debug("Finishing commit for transaction {}", cohortEntry.getTransactionID()); + LOG.debug("{}: Finishing commit for transaction {}", persistenceId(), cohortEntry.getTransactionID()); try { // We block on the future here so we don't have to worry about possibly accessing our @@ -374,24 +365,25 @@ public class Shard extends RaftActor { shardMBean.incrementCommittedTransactionCount(); shardMBean.setLastCommittedTransactionTime(System.currentTimeMillis()); - } catch (InterruptedException | ExecutionException e) { + } catch (Exception e) { sender.tell(new akka.actor.Status.Failure(e), getSelf()); - LOG.error(e, "An exception occurred while committing transaction {}", transactionID); + LOG.error("{}, An exception occurred while committing transaction {}", persistenceId(), + transactionID, e); shardMBean.incrementFailedTransactionsCount(); + } finally { + commitCoordinator.currentTransactionComplete(transactionID, true); } - - commitCoordinator.currentTransactionComplete(transactionID, true); } private void handleCanCommitTransaction(final CanCommitTransaction canCommit) { - LOG.debug("Can committing transaction {}", canCommit.getTransactionID()); + LOG.debug("{}: Can committing transaction {}", persistenceId(), canCommit.getTransactionID()); commitCoordinator.handleCanCommit(canCommit, getSender(), self()); } private void handleForwardedReadyTransaction(ForwardedReadyTransaction ready) { - LOG.debug("Readying transaction {}, client version {}", ready.getTransactionID(), - ready.getTxnClientVersion()); + LOG.debug("{}: Readying transaction {}, client version {}", persistenceId(), + ready.getTransactionID(), ready.getTxnClientVersion()); // This message is forwarded by the ShardTransaction on ready. We cache the cohort in the // commitCoordinator in preparation for the subsequent three phase commit initiated by @@ -406,7 +398,7 @@ public class Shard extends RaftActor { // to provide the compatible behavior. ActorRef replyActorPath = self(); if(ready.getTxnClientVersion() < DataStoreVersions.HELIUM_1_VERSION) { - LOG.debug("Creating BackwardsCompatibleThreePhaseCommitCohort"); + LOG.debug("{}: Creating BackwardsCompatibleThreePhaseCommitCohort", persistenceId()); replyActorPath = getContext().actorOf(BackwardsCompatibleThreePhaseCommitCohort.props( ready.getTransactionID())); } @@ -424,7 +416,7 @@ public class Shard extends RaftActor { void doAbortTransaction(final String transactionID, final ActorRef sender) { final CohortEntry cohortEntry = commitCoordinator.getCohortEntryIfCurrent(transactionID); if(cohortEntry != null) { - LOG.debug("Aborting transaction {}", transactionID); + LOG.debug("{}: Aborting transaction {}", persistenceId(), transactionID); // We don't remove the cached cohort entry here (ie pass false) in case the Tx was // aborted during replication in which case we may still commit locally if replication @@ -446,7 +438,7 @@ public class Shard extends RaftActor { @Override public void onFailure(final Throwable t) { - LOG.error(t, "An exception happened during abort"); + LOG.error("{}: An exception happened during abort", persistenceId(), t); if(sender != null) { sender.tell(new akka.actor.Status.Failure(t), self); @@ -462,10 +454,10 @@ public class Shard extends RaftActor { } else if (getLeader() != null) { getLeader().forward(message, getContext()); } else { - getSender().tell(new akka.actor.Status.Failure(new NoShardLeaderException( - "Could not find shard leader so transaction cannot be created. This typically happens" + + getSender().tell(new akka.actor.Status.Failure(new NoShardLeaderException(String.format( + "Could not find leader for shard %s so transaction cannot be created. This typically happens" + " when the system is coming up or recovering and a leader is being elected. Try again" + - " later.")), getSelf()); + " later.", persistenceId()))), getSelf()); } } @@ -556,7 +548,7 @@ public class Shard extends RaftActor { .build(); if(LOG.isDebugEnabled()) { - LOG.debug("Creating transaction : {} ", transactionId); + LOG.debug("{}: Creating transaction : {} ", persistenceId(), transactionId); } ActorRef transactionActor = createTypedTransactionActor(transactionType, transactionId, @@ -581,7 +573,7 @@ public class Shard extends RaftActor { shardMBean.setLastCommittedTransactionTime(System.currentTimeMillis()); } catch (InterruptedException | ExecutionException e) { shardMBean.incrementFailedTransactionsCount(); - LOG.error(e, "Failed to commit"); + LOG.error("{}: Failed to commit", persistenceId(), e); } } @@ -598,14 +590,14 @@ public class Shard extends RaftActor { private void registerChangeListener(final RegisterChangeListener registerChangeListener) { - LOG.debug("registerDataChangeListener for {}", registerChangeListener.getPath()); + LOG.debug("{}: registerDataChangeListener for {}", persistenceId(), registerChangeListener.getPath()); ListenerRegistration>> registration; if(isLeader()) { registration = doChangeListenerRegistration(registerChangeListener); } else { - LOG.debug("Shard is not the leader - delaying registration"); + LOG.debug("{}: Shard is not the leader - delaying registration", persistenceId()); DelayedListenerRegistration delayedReg = new DelayedListenerRegistration(registerChangeListener); @@ -616,8 +608,8 @@ public class Shard extends RaftActor { ActorRef listenerRegistration = getContext().actorOf( DataChangeListenerRegistration.props(registration)); - LOG.debug("registerDataChangeListener sending reply, listenerRegistrationPath = {} ", - listenerRegistration.path()); + LOG.debug("{}: registerDataChangeListener sending reply, listenerRegistrationPath = {} ", + persistenceId(), listenerRegistration.path()); getSender().tell(new RegisterChangeListenerReply(listenerRegistration.path()), getSelf()); } @@ -641,7 +633,7 @@ public class Shard extends RaftActor { AsyncDataChangeListener> listener = new DataChangeListenerProxy(dataChangeListenerPath); - LOG.debug("Registering for path {}", registerChangeListener.getPath()); + LOG.debug("{}: Registering for path {}", persistenceId(), registerChangeListener.getPath()); return store.registerChangeListener(registerChangeListener.getPath(), listener, registerChangeListener.getScope()); @@ -658,7 +650,7 @@ public class Shard extends RaftActor { currentLogRecoveryBatch = Lists.newArrayListWithCapacity(maxBatchSize); if(LOG.isDebugEnabled()) { - LOG.debug("{} : starting log recovery batch with max size {}", persistenceId(), maxBatchSize); + LOG.debug("{}: starting log recovery batch with max size {}", persistenceId(), maxBatchSize); } } @@ -668,40 +660,42 @@ public class Shard extends RaftActor { try { currentLogRecoveryBatch.add(((ModificationPayload) data).getModification()); } catch (ClassNotFoundException | IOException e) { - LOG.error(e, "Error extracting ModificationPayload"); + LOG.error("{}: Error extracting ModificationPayload", persistenceId(), e); } } else if (data instanceof CompositeModificationPayload) { currentLogRecoveryBatch.add(((CompositeModificationPayload) data).getModification()); } else if (data instanceof CompositeModificationByteStringPayload) { currentLogRecoveryBatch.add(((CompositeModificationByteStringPayload) data).getModification()); } else { - LOG.error("Unknown state received {} during recovery", data); + LOG.error("{}: Unknown state received {} during recovery", persistenceId(), data); } } @Override protected void applyRecoverySnapshot(final byte[] snapshotBytes) { if(recoveryCoordinator == null) { - recoveryCoordinator = new ShardRecoveryCoordinator(persistenceId(), schemaContext); + recoveryCoordinator = new ShardRecoveryCoordinator(persistenceId(), schemaContext, + LOG, name.toString()); } recoveryCoordinator.submit(snapshotBytes, store.newWriteOnlyTransaction()); if(LOG.isDebugEnabled()) { - LOG.debug("{} : submitted recovery sbapshot", persistenceId()); + LOG.debug("{}: submitted recovery sbapshot", persistenceId()); } } @Override protected void applyCurrentLogRecoveryBatch() { if(recoveryCoordinator == null) { - recoveryCoordinator = new ShardRecoveryCoordinator(persistenceId(), schemaContext); + recoveryCoordinator = new ShardRecoveryCoordinator(persistenceId(), schemaContext, + LOG, name.toString()); } recoveryCoordinator.submit(currentLogRecoveryBatch, store.newWriteOnlyTransaction()); if(LOG.isDebugEnabled()) { - LOG.debug("{} : submitted log recovery batch with size {}", persistenceId(), + LOG.debug("{}: submitted log recovery batch with size {}", persistenceId(), currentLogRecoveryBatch.size()); } } @@ -712,7 +706,7 @@ public class Shard extends RaftActor { Collection txList = recoveryCoordinator.getTransactions(); if(LOG.isDebugEnabled()) { - LOG.debug("{} : recovery complete - committing {} Tx's", persistenceId(), txList.size()); + LOG.debug("{}: recovery complete - committing {} Tx's", persistenceId(), txList.size()); } for(DOMStoreWriteTransaction tx: txList) { @@ -721,7 +715,7 @@ public class Shard extends RaftActor { shardMBean.incrementCommittedTransactionCount(); } catch (InterruptedException | ExecutionException e) { shardMBean.incrementFailedTransactionsCount(); - LOG.error(e, "Failed to commit"); + LOG.error("{}: Failed to commit", persistenceId(), e); } } } @@ -751,7 +745,7 @@ public class Shard extends RaftActor { try { applyModificationToState(clientActor, identifier, ((ModificationPayload) data).getModification()); } catch (ClassNotFoundException | IOException e) { - LOG.error(e, "Error extracting ModificationPayload"); + LOG.error("{}: Error extracting ModificationPayload", persistenceId(), e); } } else if (data instanceof CompositeModificationPayload) { @@ -763,8 +757,8 @@ public class Shard extends RaftActor { applyModificationToState(clientActor, identifier, modification); } else { - LOG.error("Unknown state received {} Class loader = {} CompositeNodeMod.ClassLoader = {}", - data, data.getClass().getClassLoader(), + LOG.error("{}: Unknown state received {} Class loader = {} CompositeNodeMod.ClassLoader = {}", + persistenceId(), data, data.getClass().getClassLoader(), CompositeModificationPayload.class.getClassLoader()); } @@ -775,8 +769,8 @@ public class Shard extends RaftActor { private void applyModificationToState(ActorRef clientActor, String identifier, Object modification) { if(modification == null) { LOG.error( - "modification is null - this is very unexpected, clientActor = {}, identifier = {}", - identifier, clientActor != null ? clientActor.path().toString() : null); + "{}: modification is null - this is very unexpected, clientActor = {}, identifier = {}", + persistenceId(), identifier, clientActor != null ? clientActor.path().toString() : null); } else if(clientActor == null) { // There's no clientActor to which to send a commit reply so we must be applying // replicated state from the leader. @@ -821,7 +815,7 @@ public class Shard extends RaftActor { // we can safely commit everything in here. We not need to worry about event notifications // as they would have already been disabled on the follower - LOG.info("Applying snapshot"); + LOG.info("{}: Applying snapshot", persistenceId()); try { DOMStoreWriteTransaction transaction = store.newWriteOnlyTransaction(); @@ -834,9 +828,9 @@ public class Shard extends RaftActor { transaction.write(DATASTORE_ROOT, node); syncCommitTransaction(transaction); } catch (InterruptedException | ExecutionException e) { - LOG.error(e, "An exception occurred when applying snapshot"); + LOG.error("{}: An exception occurred when applying snapshot", persistenceId(), e); } finally { - LOG.info("Done applying snapshot"); + LOG.info("{}: Done applying snapshot", persistenceId()); } } @@ -865,8 +859,8 @@ public class Shard extends RaftActor { for(Map.Entry entry : transactionChains.entrySet()){ if(LOG.isDebugEnabled()) { LOG.debug( - "onStateChanged: Closing transaction chain {} because shard {} is no longer the leader", - entry.getKey(), getId()); + "{}: onStateChanged: Closing transaction chain {} because shard {} is no longer the leader", + persistenceId(), entry.getKey(), getId()); } entry.getValue().close(); }