Skip to content

RATIS-810. Add metrics for StateMachine ApplyLog. #1278

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
nandakumar131 wants to merge 1 commit into
base: master
Choose a base branch
from
Draft

RATIS-810. Add metrics for StateMachine ApplyLog. #1278

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 16 additions & 0 deletions ratis-server/src/main/java/org/apache/ratis/server/impl/StateMachineMetrics.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,9 +22,12 @@
import org.apache.ratis.metrics.RatisMetricRegistry;
import org.apache.ratis.metrics.RatisMetrics;
import org.apache.ratis.metrics.Timekeeper;
import org.apache.ratis.proto.RaftProtos.LogEntryProto.LogEntryBodyCase;
import org.apache.ratis.server.raftlog.RaftLogIndex;
import org.apache.ratis.statemachine.StateMachine;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.LongSupplier;

/**
Expand All @@ -39,6 +42,11 @@ public final class StateMachineMetrics extends RatisMetrics {
public static final String STATEMACHINE_APPLY_COMPLETED_GAUGE = "applyCompletedIndex";
public static final String STATEMACHINE_TAKE_SNAPSHOT_TIMER = "takeSnapshot";

/** Time taken for the State Machine applyLog operation to complete execution. */
public static final String STATEMACHINE_APPLY_LOG_EXECUTION_TIME = "%sApplyLogExecutionTime";

private final Map<LogEntryBodyCase, Timekeeper> applyLogTimers = new ConcurrentHashMap<>();

public static StateMachineMetrics getStateMachineMetrics(
RaftServerImpl server, RaftLogIndex appliedIndex,
StateMachine stateMachine) {
Expand Down Expand Up @@ -72,4 +80,12 @@ public Timekeeper getTakeSnapshotTimer() {
return takeSnapshotTimer;
}

private Timekeeper newApplyLogExecutionTimer(LogEntryBodyCase logType) {
return getRegistry().timer(String.format(STATEMACHINE_APPLY_LOG_EXECUTION_TIME,
logType.name().toLowerCase()));
}

public Timekeeper getApplyLogExecutionTimer(LogEntryBodyCase logType) {
return applyLogTimers.computeIfAbsent(logType, this::newApplyLogExecutionTimer);
}
}
7 changes: 5 additions & 2 deletions ratis-server/src/main/java/org/apache/ratis/server/impl/StateMachineUpdater.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -250,7 +250,8 @@ private CompletableFuture<Void> applyLog(CompletableFuture<Void> applyLogFutures
} else {
LOG.debug("{}: applying nextIndex={}", this, nextIndex);
}

final Timekeeper.Context applyLogTimerContext = stateMachineMetrics.get().getApplyLogExecutionTimer(
next.getLogEntryBodyCase()).time();
final CompletableFuture<Message> f = server.applyLogToStateMachine(next);
final long incremented = appliedIndex.incrementAndGet(debugIndexChange);
Preconditions.assertTrue(incremented == nextIndex);
Expand All @@ -259,10 +260,12 @@ private CompletableFuture<Void> applyLog(CompletableFuture<Void> applyLogFutures
LOG.error("Exception while {}: applying txn index={}, nextLog={}", this, nextIndex,
LogProtoUtils.toLogEntryString(next), ex);
return null;
});
})
.whenComplete((m, e) -> applyLogTimerContext.stop());
applyLogFutures = applyLogFutures.thenCombine(exceptionHandledFuture, (v, message) -> null);
f.thenAccept(m -> notifyAppliedIndex(incremented));
} else {
applyLogTimerContext.stop();
notifyAppliedIndex(incremented);
}
} else {
Expand Down
Loading