package org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard;
+import akka.actor.ActorRef;
+import akka.pattern.Patterns;
+import akka.util.Timeout;
+import com.google.common.base.Stopwatch;
+import com.google.common.cache.Cache;
+import com.google.common.cache.CacheBuilder;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
+import org.opendaylight.controller.cluster.raft.client.messages.FollowerInfo;
+import org.opendaylight.controller.cluster.raft.client.messages.GetOnDemandRaftState;
+import org.opendaylight.controller.cluster.raft.client.messages.OnDemandRaftState;
import org.opendaylight.controller.md.sal.common.util.jmx.AbstractMXBean;
import org.opendaylight.controller.md.sal.common.util.jmx.QueuedNotificationManagerMXBeanImpl;
-import org.opendaylight.controller.md.sal.common.util.jmx.ThreadExecutorStats;
import org.opendaylight.controller.md.sal.common.util.jmx.ThreadExecutorStatsMXBeanImpl;
import org.opendaylight.controller.md.sal.dom.store.impl.InMemoryDOMDataStore;
-import org.opendaylight.yangtools.util.concurrent.ListenerNotificationQueueStats;
import org.opendaylight.yangtools.util.concurrent.QueuedNotificationManager;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import scala.concurrent.Await;
/**
* Maintains statistics for a shard.
public class ShardStats extends AbstractMXBean implements ShardStatsMXBean {
public static String JMX_CATEGORY_SHARD = "Shards";
- private final AtomicLong committedTransactionsCount = new AtomicLong();
+ private static final Logger LOG = LoggerFactory.getLogger(ShardStats.class);
- private final AtomicLong readOnlyTransactionCount = new AtomicLong();
+ private static final SimpleDateFormat DATE_FORMAT = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
- private final AtomicLong writeOnlyTransactionCount = new AtomicLong();
+ private static final Cache<String, OnDemandRaftState> onDemandRaftStateCache =
+ CacheBuilder.newBuilder().expireAfterWrite(2, TimeUnit.SECONDS).build();
- private final AtomicLong readWriteTransactionCount = new AtomicLong();
+ private long committedTransactionsCount;
- private String leader;
+ private long readOnlyTransactionCount;
- private String raftState;
+ private long writeOnlyTransactionCount;
- private volatile long lastLogTerm = -1L;
+ private long readWriteTransactionCount;
- private volatile long lastLogIndex = -1L;
+ private long lastCommittedTransactionTime;
- private volatile long currentTerm = -1L;
-
- private volatile long commitIndex = -1L;
-
- private volatile long lastApplied = -1L;
-
- private volatile long lastCommittedTransactionTime;
-
- private final AtomicLong failedTransactionsCount = new AtomicLong();
+ private long failedTransactionsCount;
private final AtomicLong failedReadTransactionsCount = new AtomicLong();
- private final AtomicLong abortTransactionsCount = new AtomicLong();
+ private long abortTransactionsCount;
private ThreadExecutorStatsMXBeanImpl notificationExecutorStatsBean;
private QueuedNotificationManagerMXBeanImpl notificationManagerStatsBean;
- private volatile long dataSize = 0;
+ private boolean followerInitialSyncStatus = false;
- private final SimpleDateFormat sdf =
- new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
+ private ActorRef shardActor;
- private boolean followerInitialSyncStatus = false;
+ private String statRetrievalError;
+
+ private String statRetrievalTime;
+
+ private long leadershipChangeCount;
+
+ private long lastLeadershipChangeTime;
public ShardStats(final String shardName, final String mxBeanType) {
super(shardName, mxBeanType, JMX_CATEGORY_SHARD);
this.notificationExecutorStatsBean = ThreadExecutorStatsMXBeanImpl.create(manager.getExecutor());
}
+ public void setShardActor(ActorRef shardActor) {
+ this.shardActor = shardActor;
+ }
+
+ private OnDemandRaftState getOnDemandRaftState() {
+ String name = getShardName();
+ OnDemandRaftState state = onDemandRaftStateCache.getIfPresent(name);
+ if(state == null) {
+ statRetrievalError = null;
+ statRetrievalTime = null;
+
+ if(shardActor != null) {
+ Timeout timeout = new Timeout(10, TimeUnit.SECONDS);
+ try {
+ Stopwatch timer = Stopwatch.createStarted();
+
+ state = (OnDemandRaftState) Await.result(Patterns.ask(shardActor,
+ GetOnDemandRaftState.INSTANCE, timeout), timeout.duration());
+
+ statRetrievalTime = timer.stop().toString();
+ onDemandRaftStateCache.put(name, state);
+ } catch (Exception e) {
+ statRetrievalError = e.toString();
+ }
+ }
+
+ state = state != null ? state : OnDemandRaftState.builder().build();
+ }
+
+ return state;
+ }
+
@Override
public String getShardName() {
return getMBeanName();
@Override
public long getCommittedTransactionsCount() {
- return committedTransactionsCount.get();
+ return committedTransactionsCount;
}
@Override
public String getLeader() {
- return leader;
+ return getOnDemandRaftState().getLeader();
}
@Override
public String getRaftState() {
- return raftState;
+ return getOnDemandRaftState().getRaftState();
}
@Override
public long getReadOnlyTransactionCount() {
- return readOnlyTransactionCount.get();
+ return readOnlyTransactionCount;
}
@Override
public long getWriteOnlyTransactionCount() {
- return writeOnlyTransactionCount.get();
+ return writeOnlyTransactionCount;
}
@Override
public long getReadWriteTransactionCount() {
- return readWriteTransactionCount.get();
+ return readWriteTransactionCount;
}
@Override
public long getLastLogIndex() {
- return lastLogIndex;
+ return getOnDemandRaftState().getLastLogIndex();
}
@Override
public long getLastLogTerm() {
- return lastLogTerm;
+ return getOnDemandRaftState().getLastLogTerm();
}
@Override
public long getCurrentTerm() {
- return currentTerm;
+ return getOnDemandRaftState().getCurrentTerm();
}
@Override
public long getCommitIndex() {
- return commitIndex;
+ return getOnDemandRaftState().getCommitIndex();
}
@Override
public long getLastApplied() {
- return lastApplied;
+ return getOnDemandRaftState().getLastApplied();
}
@Override
- public String getLastCommittedTransactionTime() {
-
- return sdf.format(new Date(lastCommittedTransactionTime));
+ public long getLastIndex() {
+ return getOnDemandRaftState().getLastIndex();
}
@Override
- public long getFailedTransactionsCount() {
- return failedTransactionsCount.get();
+ public long getLastTerm() {
+ return getOnDemandRaftState().getLastTerm();
}
@Override
- public long getFailedReadTransactionsCount() {
- return failedReadTransactionsCount.get();
+ public long getSnapshotIndex() {
+ return getOnDemandRaftState().getSnapshotIndex();
}
@Override
- public long getAbortTransactionsCount() {
- return abortTransactionsCount.get();
+ public long getSnapshotTerm() {
+ return getOnDemandRaftState().getSnapshotTerm();
}
- public long incrementCommittedTransactionCount() {
- return committedTransactionsCount.incrementAndGet();
+ @Override
+ public long getReplicatedToAllIndex() {
+ return getOnDemandRaftState().getReplicatedToAllIndex();
}
- public long incrementReadOnlyTransactionCount() {
- return readOnlyTransactionCount.incrementAndGet();
+ @Override
+ public String getVotedFor() {
+ return getOnDemandRaftState().getVotedFor();
}
- public long incrementWriteOnlyTransactionCount() {
- return writeOnlyTransactionCount.incrementAndGet();
+ @Override
+ public boolean isSnapshotCaptureInitiated() {
+ return getOnDemandRaftState().isSnapshotCaptureInitiated();
}
- public long incrementReadWriteTransactionCount() {
- return readWriteTransactionCount.incrementAndGet();
+ @Override
+ public String getLastCommittedTransactionTime() {
+ return DATE_FORMAT.format(new Date(lastCommittedTransactionTime));
}
- public long incrementFailedTransactionsCount() {
- return failedTransactionsCount.incrementAndGet();
+ @Override
+ public long getFailedTransactionsCount() {
+ return failedTransactionsCount;
}
- public long incrementFailedReadTransactionsCount() {
- return failedReadTransactionsCount.incrementAndGet();
+ @Override
+ public long getFailedReadTransactionsCount() {
+ return failedReadTransactionsCount.get();
}
- public long incrementAbortTransactionsCount ()
- {
- return abortTransactionsCount.incrementAndGet();
+ @Override
+ public long getAbortTransactionsCount() {
+ return abortTransactionsCount;
}
- public void setLeader(final String leader) {
- this.leader = leader;
+ public long incrementCommittedTransactionCount() {
+ return ++committedTransactionsCount;
}
- public void setRaftState(final String raftState) {
- this.raftState = raftState;
+ public long incrementReadOnlyTransactionCount() {
+ return ++readOnlyTransactionCount;
}
- public void setLastLogTerm(final long lastLogTerm) {
- this.lastLogTerm = lastLogTerm;
+ public long incrementWriteOnlyTransactionCount() {
+ return ++writeOnlyTransactionCount;
}
- public void setLastLogIndex(final long lastLogIndex) {
- this.lastLogIndex = lastLogIndex;
+ public long incrementReadWriteTransactionCount() {
+ return ++readWriteTransactionCount;
}
- public void setCurrentTerm(final long currentTerm) {
- this.currentTerm = currentTerm;
+ public long incrementFailedTransactionsCount() {
+ return ++failedTransactionsCount;
}
- public void setCommitIndex(final long commitIndex) {
- this.commitIndex = commitIndex;
+ public long incrementFailedReadTransactionsCount() {
+ return failedReadTransactionsCount.incrementAndGet();
}
- public void setLastApplied(final long lastApplied) {
- this.lastApplied = lastApplied;
+ public long incrementAbortTransactionsCount ()
+ {
+ return ++abortTransactionsCount;
}
public void setLastCommittedTransactionTime(final long lastCommittedTransactionTime) {
this.lastCommittedTransactionTime = lastCommittedTransactionTime;
}
- public void setInMemoryJournalDataSize(long dataSize){
- this.dataSize = dataSize;
- }
-
@Override
public long getInMemoryJournalDataSize(){
- return dataSize;
- }
-
- @Override
- public ThreadExecutorStats getDataStoreExecutorStats() {
- // FIXME: this particular thing does not work, as it really is DS-specific
- return null;
+ return getOnDemandRaftState().getInMemoryJournalDataSize();
}
@Override
- public ThreadExecutorStats getNotificationMgrExecutorStats() {
- return notificationExecutorStatsBean.toThreadExecutorStats();
- }
-
- @Override
- public List<ListenerNotificationQueueStats> getCurrentNotificationMgrListenerQueueStats() {
- return notificationManagerStatsBean.getCurrentListenerQueueStats();
- }
-
- @Override
- public int getMaxNotificationMgrListenerQueueSize() {
- return notificationManagerStatsBean.getMaxListenerQueueSize();
+ public long getInMemoryJournalLogSize() {
+ return getOnDemandRaftState().getInMemoryJournalLogSize();
}
/**
*/
@Override
public void resetTransactionCounters(){
- committedTransactionsCount.set(0);
+ committedTransactionsCount = 0;
- readOnlyTransactionCount.set(0);
+ readOnlyTransactionCount = 0;
- writeOnlyTransactionCount.set(0);
+ writeOnlyTransactionCount = 0;
- readWriteTransactionCount.set(0);
+ readWriteTransactionCount = 0;
lastCommittedTransactionTime = 0;
- failedTransactionsCount.set(0);
+ failedTransactionsCount = 0;
failedReadTransactionsCount.set(0);
- abortTransactionsCount.set(0);
+ abortTransactionsCount = 0;
}
public boolean getFollowerInitialSyncStatus() {
return followerInitialSyncStatus;
}
+
+ @Override
+ public List<FollowerInfo> getFollowerInfo() {
+ return getOnDemandRaftState().getFollowerInfoList();
+ }
+
+ @Override
+ public String getPeerAddresses() {
+ StringBuilder builder = new StringBuilder();
+ int i = 0;
+ for(Map.Entry<String, String> e: getOnDemandRaftState().getPeerAddresses().entrySet()) {
+ if(i++ > 0) {
+ builder.append(", ");
+ }
+
+ builder.append(e.getKey()).append(": ").append(e.getValue());
+ }
+
+ return builder.toString();
+ }
+
+ @Override
+ public String getStatRetrievalTime() {
+ getOnDemandRaftState();
+ return statRetrievalTime;
+ }
+
+ @Override
+ public String getStatRetrievalError() {
+ getOnDemandRaftState();
+ return statRetrievalError;
+ }
+
+ @Override
+ public long getLeadershipChangeCount() {
+ return leadershipChangeCount;
+ }
+
+ public void incrementLeadershipChangeCount() {
+ leadershipChangeCount++;
+ lastLeadershipChangeTime = System.currentTimeMillis();
+ }
+
+ @Override
+ public String getLastLeadershipChangeTime() {
+ return DATE_FORMAT.format(new Date(lastLeadershipChangeTime));
+ }
}