diff --git a/ratis-common/src/main/java/org/apache/ratis/util/UncheckedAutoCloseable.java b/ratis-common/src/main/java/org/apache/ratis/util/UncheckedAutoCloseable.java index cc7315934a..9f2ba0ede0 100644 --- a/ratis-common/src/main/java/org/apache/ratis/util/UncheckedAutoCloseable.java +++ b/ratis-common/src/main/java/org/apache/ratis/util/UncheckedAutoCloseable.java @@ -1,4 +1,4 @@ -/** +/* * Licensed to the Apache Software Foundation (ASF) under one * or more contributor license agreements. See the NOTICE file * distributed with this work for additional information @@ -21,6 +21,7 @@ * The same as {@link AutoCloseable} * except that the close method does not throw {@link Exception}. */ +@FunctionalInterface public interface UncheckedAutoCloseable extends AutoCloseable { @Override void close(); diff --git a/ratis-grpc/src/main/java/org/apache/ratis/grpc/metrics/GrpcServerMetrics.java b/ratis-grpc/src/main/java/org/apache/ratis/grpc/metrics/GrpcServerMetrics.java index df55514f93..1ffe2ee4a1 100644 --- a/ratis-grpc/src/main/java/org/apache/ratis/grpc/metrics/GrpcServerMetrics.java +++ b/ratis-grpc/src/main/java/org/apache/ratis/grpc/metrics/GrpcServerMetrics.java @@ -22,7 +22,7 @@ import org.apache.ratis.metrics.RatisMetricRegistry; import org.apache.ratis.metrics.RatisMetrics; -import org.apache.ratis.thirdparty.com.codahale.metrics.Timer; +import org.apache.ratis.metrics.Timekeeper; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; @@ -62,6 +62,9 @@ public class GrpcServerMetrics extends RatisMetrics { private final Map requestNotLeader = new ConcurrentHashMap<>(); private final Map requestInconsistency = new ConcurrentHashMap<>(); + private final Map heartbeatLatency = new ConcurrentHashMap<>(); + private final Map appendLogLatency = new ConcurrentHashMap<>(); + public GrpcServerMetrics(String serverId) { registry = getMetricRegistryForGrpcServer(serverId); @@ -77,10 +80,11 @@ private RatisMetricRegistry getMetricRegistryForGrpcServer(String serverId) { RATIS_GRPC_METRICS_COMP_NAME, RATIS_GRPC_METRICS_DESC)); } - public Timer getGrpcLogAppenderLatencyTimer(String follower, - boolean isHeartbeat) { - return registry.timer(String.format(RATIS_GRPC_METRICS_LOG_APPENDER_LATENCY + getHeartbeatSuffix(isHeartbeat), - follower)); + public Timekeeper getGrpcLogAppenderLatencyTimer(String follower, boolean isHeartbeat) { + final Map map = isHeartbeat ? heartbeatLatency : appendLogLatency; + final String name = map.computeIfAbsent(follower, + key -> String.format(RATIS_GRPC_METRICS_LOG_APPENDER_LATENCY + getHeartbeatSuffix(isHeartbeat), key)); + return registry.timer(name); } public void onRequestRetry() { diff --git a/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcLogAppender.java b/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcLogAppender.java index f83656cb03..21477d569f 100644 --- a/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcLogAppender.java +++ b/ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcLogAppender.java @@ -21,6 +21,7 @@ import org.apache.ratis.grpc.GrpcConfigKeys; import org.apache.ratis.grpc.GrpcUtil; import org.apache.ratis.grpc.metrics.GrpcServerMetrics; +import org.apache.ratis.metrics.Timekeeper; import org.apache.ratis.proto.RaftProtos.InstallSnapshotResult; import org.apache.ratis.protocol.RaftPeer; import org.apache.ratis.protocol.RaftPeerId; @@ -49,8 +50,6 @@ import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; -import org.apache.ratis.thirdparty.com.codahale.metrics.Timer; - /** * A new log appender implementation using grpc bi-directional stream API. */ @@ -280,10 +279,10 @@ private void appendLog(boolean excludeLogEntries) throws IOException { private void sendRequest(AppendEntriesRequest request, AppendEntriesRequestProto proto) { CodeInjectionForTesting.execute(GrpcService.GRPC_SEND_SERVER_REQUEST, getServer().getId(), null, proto); - request.startRequestTimer(); resetHeartbeatTrigger(); final boolean sent = Optional.ofNullable(appendLogRequestObserver) .map(observer -> { + request.startRequestTimer(); observer.onNext(proto); return true; }).isPresent(); @@ -293,8 +292,6 @@ private void sendRequest(AppendEntriesRequest request, AppendEntriesRequestProto () -> timeoutAppendRequest(request.getCallId(), request.isHeartbeat()), LOG, () -> "Timeout check failed for append entry request: " + request); getFollower().updateLastRpcSendTime(request.isHeartbeat()); - } else { - request.stopRequestTimer(); } } @@ -696,8 +693,8 @@ private TermIndex shouldNotifyToInstallSnapshot() { } static class AppendEntriesRequest { - private final Timer timer; - private volatile Timer.Context timerContext; + private final Timekeeper timer; + private volatile Timekeeper.Context timerContext; private final long callId; private final TermIndex previousLog; diff --git a/ratis-grpc/src/test/java/org/apache/ratis/grpc/server/TestGrpcServerMetrics.java b/ratis-grpc/src/test/java/org/apache/ratis/grpc/server/TestGrpcServerMetrics.java index dad04e32e7..04f8ded95d 100644 --- a/ratis-grpc/src/test/java/org/apache/ratis/grpc/server/TestGrpcServerMetrics.java +++ b/ratis-grpc/src/test/java/org/apache/ratis/grpc/server/TestGrpcServerMetrics.java @@ -29,6 +29,7 @@ import java.util.function.Consumer; +import org.apache.ratis.metrics.impl.DefaultTimekeeperImpl; import org.apache.ratis.server.metrics.ServerMetricsTestUtils; import org.apache.ratis.grpc.metrics.GrpcServerMetrics; import org.apache.ratis.metrics.RatisMetricRegistry; @@ -70,17 +71,14 @@ public void testGrpcLogAppenderLatencyTimer() throws Exception { GrpcLogAppender.AppendEntriesRequest req = new GrpcLogAppender.AppendEntriesRequest(proto.build(), followerId, grpcServerMetrics); - Assert.assertEquals(0L, ratisMetricRegistry.timer(String.format( - RATIS_GRPC_METRICS_LOG_APPENDER_LATENCY + GrpcServerMetrics - .getHeartbeatSuffix(heartbeat), followerId.toString())) - .getSnapshot().getMax()); + final String format = RATIS_GRPC_METRICS_LOG_APPENDER_LATENCY + GrpcServerMetrics.getHeartbeatSuffix(heartbeat); + final String name = String.format(format, followerId); + final DefaultTimekeeperImpl t = (DefaultTimekeeperImpl) ratisMetricRegistry.timer(name); + Assert.assertEquals(0L, t.getTimer().getSnapshot().getMax()); req.startRequestTimer(); Thread.sleep(1000L); req.stopRequestTimer(); - Assert.assertTrue(ratisMetricRegistry.timer(String.format( - RATIS_GRPC_METRICS_LOG_APPENDER_LATENCY + GrpcServerMetrics - .getHeartbeatSuffix(heartbeat), followerId.toString())) - .getSnapshot().getMax() > 1000L); + Assert.assertTrue(t.getTimer().getSnapshot().getMax() > 1000L); } } diff --git a/ratis-metrics/src/main/java/org/apache/ratis/metrics/RatisMetricRegistry.java b/ratis-metrics/src/main/java/org/apache/ratis/metrics/RatisMetricRegistry.java index fc29f45a90..9ebd8f0a5f 100644 --- a/ratis-metrics/src/main/java/org/apache/ratis/metrics/RatisMetricRegistry.java +++ b/ratis-metrics/src/main/java/org/apache/ratis/metrics/RatisMetricRegistry.java @@ -23,14 +23,13 @@ import org.apache.ratis.thirdparty.com.codahale.metrics.Metric; import org.apache.ratis.thirdparty.com.codahale.metrics.MetricRegistry; import org.apache.ratis.thirdparty.com.codahale.metrics.MetricSet; -import org.apache.ratis.thirdparty.com.codahale.metrics.Timer; import org.apache.ratis.thirdparty.com.codahale.metrics.jmx.JmxReporter; import org.apache.ratis.thirdparty.com.google.common.annotations.VisibleForTesting; import java.util.function.Supplier; public interface RatisMetricRegistry { - Timer timer(String name); + Timekeeper timer(String name); LongCounter counter(String name); @@ -38,8 +37,6 @@ public interface RatisMetricRegistry { void gauge(String name, Supplier> gaugeSupplier); - Timer timer(String name, MetricRegistry.MetricSupplier supplier); - Histogram histogram(String name); Meter meter(String name); diff --git a/ratis-metrics/src/main/java/org/apache/ratis/metrics/RatisMetrics.java b/ratis-metrics/src/main/java/org/apache/ratis/metrics/RatisMetrics.java index 00271f5c81..2793797a11 100644 --- a/ratis-metrics/src/main/java/org/apache/ratis/metrics/RatisMetrics.java +++ b/ratis-metrics/src/main/java/org/apache/ratis/metrics/RatisMetrics.java @@ -54,4 +54,8 @@ public void unregister() { public RatisMetricRegistry getRegistry() { return registry; } + + protected Timekeeper getTimer(String timerName) { + return getRegistry().timer(timerName); + } } diff --git a/ratis-metrics/src/main/java/org/apache/ratis/metrics/Timekeeper.java b/ratis-metrics/src/main/java/org/apache/ratis/metrics/Timekeeper.java new file mode 100644 index 0000000000..a7ac11843c --- /dev/null +++ b/ratis-metrics/src/main/java/org/apache/ratis/metrics/Timekeeper.java @@ -0,0 +1,45 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + *

+ * http://www.apache.org/licenses/LICENSE-2.0 + *

+ * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.ratis.metrics; + +import org.apache.ratis.util.UncheckedAutoCloseable; + +import java.util.Optional; + +@FunctionalInterface +public interface Timekeeper { + UncheckedAutoCloseable NOOP = () -> {}; + + static UncheckedAutoCloseable start(Timekeeper timekeeper) { + return Optional.ofNullable(timekeeper) + .map(Timekeeper::time) + .map(Context::toAutoCloseable) + .orElse(NOOP); + } + + @FunctionalInterface + interface Context { + long stop(); + + default UncheckedAutoCloseable toAutoCloseable() { + return this::stop; + } + } + + Context time(); +} diff --git a/ratis-metrics/src/main/java/org/apache/ratis/metrics/impl/DefaultTimekeeperImpl.java b/ratis-metrics/src/main/java/org/apache/ratis/metrics/impl/DefaultTimekeeperImpl.java new file mode 100644 index 0000000000..abe7f897ad --- /dev/null +++ b/ratis-metrics/src/main/java/org/apache/ratis/metrics/impl/DefaultTimekeeperImpl.java @@ -0,0 +1,42 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.ratis.metrics.impl; + +import org.apache.ratis.metrics.Timekeeper; +import org.apache.ratis.thirdparty.com.codahale.metrics.Timer; + +/** + * The default implementation of {@link Timekeeper} by the shaded {@link Timer}. + */ +public class DefaultTimekeeperImpl implements Timekeeper { + private final Timer timer; + + DefaultTimekeeperImpl(Timer timer) { + this.timer = timer; + } + + public Timer getTimer() { + return timer; + } + + @Override + public Context time() { + final Timer.Context context = timer.time(); + return context::stop; + } +} diff --git a/ratis-metrics/src/main/java/org/apache/ratis/metrics/impl/RatisMetricRegistryImpl.java b/ratis-metrics/src/main/java/org/apache/ratis/metrics/impl/RatisMetricRegistryImpl.java index e77aa7b4c6..10c2adb757 100644 --- a/ratis-metrics/src/main/java/org/apache/ratis/metrics/impl/RatisMetricRegistryImpl.java +++ b/ratis-metrics/src/main/java/org/apache/ratis/metrics/impl/RatisMetricRegistryImpl.java @@ -20,6 +20,7 @@ import org.apache.ratis.metrics.LongCounter; import org.apache.ratis.metrics.MetricRegistryInfo; import org.apache.ratis.metrics.RatisMetricRegistry; +import org.apache.ratis.metrics.Timekeeper; import org.apache.ratis.thirdparty.com.codahale.metrics.ConsoleReporter; import org.apache.ratis.thirdparty.com.codahale.metrics.Counter; import org.apache.ratis.thirdparty.com.codahale.metrics.Gauge; @@ -29,7 +30,6 @@ import org.apache.ratis.thirdparty.com.codahale.metrics.MetricFilter; import org.apache.ratis.thirdparty.com.codahale.metrics.MetricRegistry; import org.apache.ratis.thirdparty.com.codahale.metrics.MetricSet; -import org.apache.ratis.thirdparty.com.codahale.metrics.Timer; import org.apache.ratis.thirdparty.com.codahale.metrics.jmx.JmxReporter; import org.apache.ratis.thirdparty.com.google.common.annotations.VisibleForTesting; @@ -54,8 +54,8 @@ public RatisMetricRegistryImpl(MetricRegistryInfo info) { } @Override - public Timer timer(String name) { - return metricRegistry.timer(getMetricName(name)); + public Timekeeper timer(String name) { + return new DefaultTimekeeperImpl(metricRegistry.timer(getMetricName(name))); } static LongCounter toLongCounter(Counter c) { @@ -96,10 +96,6 @@ public void gauge(String name, Supplier> gaugeSupplier) { metricRegistry.gauge(getMetricName(name), () -> toGauge(gaugeSupplier.get())); } - @Override public Timer timer(String name, MetricRegistry.MetricSupplier supplier) { - return metricRegistry.timer(getMetricName(name), supplier); - } - public SortedMap getGauges(MetricFilter filter) { return metricRegistry.getGauges(filter); } diff --git a/ratis-netty/src/main/java/org/apache/ratis/netty/metrics/NettyServerStreamRpcMetrics.java b/ratis-netty/src/main/java/org/apache/ratis/netty/metrics/NettyServerStreamRpcMetrics.java index 6c8125508a..a613acb9e4 100644 --- a/ratis-netty/src/main/java/org/apache/ratis/netty/metrics/NettyServerStreamRpcMetrics.java +++ b/ratis-netty/src/main/java/org/apache/ratis/netty/metrics/NettyServerStreamRpcMetrics.java @@ -17,10 +17,10 @@ */ package org.apache.ratis.netty.metrics; -import org.apache.ratis.thirdparty.com.codahale.metrics.Timer; import org.apache.ratis.metrics.MetricRegistryInfo; import org.apache.ratis.metrics.RatisMetricRegistry; import org.apache.ratis.metrics.RatisMetrics; +import org.apache.ratis.metrics.Timekeeper; import java.util.Locale; @@ -64,34 +64,22 @@ String getLatencyString() { } } - public static final class RequestContext { - private final Timer.Context timerContext; - - private RequestContext(Timer.Context timerContext) { - this.timerContext = timerContext; - } - - Timer.Context getTimerContext() { - return timerContext; - } - } - public final class RequestMetrics { private final RequestType type; - private final Timer timer; + private final Timekeeper timer; private RequestMetrics(RequestType type) { this.type = type; this.timer = getLatencyTimer(type); } - public RequestContext start() { + public Timekeeper.Context start() { onRequestCreate(type); - return new RequestContext(timer.time()); + return timer.time(); } - public void stop(RequestContext context, boolean success) { - context.getTimerContext().stop(); + public void stop(Timekeeper.Context context, boolean success) { + context.stop(); if (success) { onRequestSuccess(type); } else { @@ -113,7 +101,7 @@ public RequestMetrics newRequestMetrics(RequestType type) { return new RequestMetrics(type); } - public Timer getLatencyTimer(RequestType type) { + public Timekeeper getLatencyTimer(RequestType type) { return registry.timer(type.getLatencyString()); } diff --git a/ratis-netty/src/main/java/org/apache/ratis/netty/server/DataStreamManagement.java b/ratis-netty/src/main/java/org/apache/ratis/netty/server/DataStreamManagement.java index 593c2793a3..96266789b5 100644 --- a/ratis-netty/src/main/java/org/apache/ratis/netty/server/DataStreamManagement.java +++ b/ratis-netty/src/main/java/org/apache/ratis/netty/server/DataStreamManagement.java @@ -25,8 +25,8 @@ import org.apache.ratis.datastream.impl.DataStreamReplyByteBuffer; import org.apache.ratis.io.StandardWriteOption; import org.apache.ratis.io.WriteOption; +import org.apache.ratis.metrics.Timekeeper; import org.apache.ratis.netty.metrics.NettyServerStreamRpcMetrics; -import org.apache.ratis.netty.metrics.NettyServerStreamRpcMetrics.RequestContext; import org.apache.ratis.netty.metrics.NettyServerStreamRpcMetrics.RequestMetrics; import org.apache.ratis.netty.metrics.NettyServerStreamRpcMetrics.RequestType; import org.apache.ratis.proto.RaftProtos.CommitInfoProto; @@ -91,7 +91,7 @@ static class LocalStream { } CompletableFuture write(ByteBuf buf, WriteOption[] options, Executor executor) { - final RequestContext context = metrics.start(); + final Timekeeper.Context context = metrics.start(); return composeAsync(writeFuture, executor, n -> streamFuture.thenCompose(stream -> writeToAsync(buf, options, stream, executor) .whenComplete((l, e) -> metrics.stop(context, e == null)))); @@ -110,7 +110,7 @@ static class RemoteStream { } CompletableFuture write(DataStreamRequestByteBuf request, Executor executor) { - final RequestContext context = metrics.start(); + final Timekeeper.Context context = metrics.start(); return composeAsync(sendFuture, executor, n -> out.writeAsync(request.slice().nioBuffer(), request.getWriteOptions()) .whenComplete((l, e) -> metrics.stop(context, e == null))); @@ -252,7 +252,7 @@ private CompletableFuture computeDataStreamIfAbsent(RaftClientReques final MemoizedSupplier> supplier = JavaUtils.memoize( () -> { final RequestMetrics metrics = getMetrics().newRequestMetrics(RequestType.STATE_MACHINE_STREAM); - final RequestContext context = metrics.start(); + final Timekeeper.Context context = metrics.start(); return division.getStateMachine().data().stream(request) .whenComplete((r, e) -> metrics.stop(context, e == null)); }); diff --git a/ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderElection.java b/ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderElection.java index 73d8c0cdf1..f89e8b502d 100644 --- a/ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderElection.java +++ b/ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderElection.java @@ -17,6 +17,7 @@ */ package org.apache.ratis.server.impl; +import org.apache.ratis.metrics.Timekeeper; import org.apache.ratis.proto.RaftProtos.RequestVoteReplyProto; import org.apache.ratis.proto.RaftProtos.RequestVoteRequestProto; import org.apache.ratis.protocol.RaftPeer; @@ -58,20 +59,18 @@ import static org.apache.ratis.util.LifeCycle.State.RUNNING; import static org.apache.ratis.util.LifeCycle.State.STARTING; -import org.apache.ratis.thirdparty.com.codahale.metrics.Timer; - /** * For a candidate to start an election for becoming the leader. * There are two phases: Pre-Vote and Election. - * + *

* In Pre-Vote, the candidate does not change its term and try to learn * if a majority of the cluster would be willing to grant the candidate their votes * (if the candidate’s log is sufficiently up-to-date, * and the voters have not received heartbeats from a valid leader * for at least a baseline election timeout). - * + *

* Once the Pre-Vote has passed, the candidate increments its term and starts a real Election. - * + *

* See * Ongaro, D. Consensus: Bridging Theory and Practice. PhD thesis, Stanford University, 2014. * Available at https://github.com/ongardie/dissertation @@ -232,8 +231,7 @@ public void run() { return; } - final Timer.Context electionContext = server.getLeaderElectionMetrics().getLeaderElectionTimer().time(); - try { + try (AutoCloseable ignored = Timekeeper.start(server.getLeaderElectionMetrics().getLeaderElectionTimer())) { if (skipPreVote || askForVotes(Phase.PRE_VOTE)) { if (askForVotes(Phase.ELECTION)) { server.changeToLeader(); @@ -255,7 +253,6 @@ public void run() { } } finally { // Update leader election completion metric(s). - electionContext.stop(); server.getLeaderElectionMetrics().onNewLeaderElectionCompletion(); lifeCycle.checkStateAndClose(() -> {}); } diff --git a/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java b/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java index 258e94d9fa..3ab600cf73 100644 --- a/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java +++ b/ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java @@ -20,6 +20,7 @@ import org.apache.ratis.client.RaftClient; import org.apache.ratis.client.impl.ClientProtoUtils; import org.apache.ratis.conf.RaftProperties; +import org.apache.ratis.metrics.Timekeeper; import org.apache.ratis.proto.RaftProtos.*; import org.apache.ratis.proto.RaftProtos.RaftClientRequestProto.TypeCase; import org.apache.ratis.protocol.*; @@ -63,6 +64,7 @@ import org.apache.ratis.statemachine.TransactionContext; import org.apache.ratis.thirdparty.com.google.protobuf.InvalidProtocolBufferException; import org.apache.ratis.util.*; +import org.apache.ratis.util.function.CheckedSupplier; import javax.management.ObjectName; import java.io.File; @@ -92,9 +94,6 @@ import static org.apache.ratis.util.LifeCycle.State.RUNNING; import static org.apache.ratis.util.LifeCycle.State.STARTING; -import org.apache.ratis.thirdparty.com.codahale.metrics.Timer; -import org.apache.ratis.util.function.CheckedSupplier; - class RaftServerImpl implements RaftServer.Division, RaftServerProtocol, RaftServerAsynchronousProtocol, RaftClientProtocol, RaftClientAsynchronousProtocol{ @@ -815,7 +814,8 @@ public CompletableFuture submitClientRequestAsync( RaftClientRequest request) throws IOException { assertLifeCycleState(LifeCycle.States.RUNNING); LOG.debug("{}: receive client request({})", getMemberId(), request); - final Optional timer = Optional.ofNullable(raftServerMetrics.getClientRequestTimer(request.getType())); + final Timekeeper timer = raftServerMetrics.getClientRequestTimer(request.getType()); + final Optional timerContext = Optional.ofNullable(timer).map(Timekeeper::time); final CompletableFuture replyFuture; @@ -878,7 +878,7 @@ public CompletableFuture submitClientRequestAsync( final RaftClientRequest.Type type = request.getType(); replyFuture.whenComplete((clientReply, exception) -> { if (clientReply.isSuccess()) { - timer.map(Timer::time).ifPresent(Timer.Context::stop); + timerContext.ifPresent(Timekeeper.Context::stop); } if (exception != null || clientReply.getException() != null) { raftServerMetrics.incFailedRequestCount(type); @@ -1367,7 +1367,7 @@ private CompletableFuture appendEntriesAsync( final long currentTerm; final long followerCommit = state.getLog().getLastCommittedIndex(); final Optional followerState; - Timer.Context timer = raftServerMetrics.getFollowerAppendEntryTimer(isHeartbeat).time(); + final Timekeeper.Context timer = raftServerMetrics.getFollowerAppendEntryTimer(isHeartbeat).time(); synchronized (this) { // Check life cycle state again to avoid the PAUSING/PAUSED state. assertLifeCycleState(LifeCycle.States.STARTING_OR_RUNNING); diff --git a/ratis-server/src/main/java/org/apache/ratis/server/impl/StateMachineMetrics.java b/ratis-server/src/main/java/org/apache/ratis/server/impl/StateMachineMetrics.java index 596cdaf5ac..690faad47c 100644 --- a/ratis-server/src/main/java/org/apache/ratis/server/impl/StateMachineMetrics.java +++ b/ratis-server/src/main/java/org/apache/ratis/server/impl/StateMachineMetrics.java @@ -21,13 +21,12 @@ import org.apache.ratis.metrics.MetricRegistryInfo; import org.apache.ratis.metrics.RatisMetricRegistry; import org.apache.ratis.metrics.RatisMetrics; +import org.apache.ratis.metrics.Timekeeper; import org.apache.ratis.server.raftlog.RaftLogIndex; import org.apache.ratis.statemachine.StateMachine; import java.util.function.LongSupplier; -import org.apache.ratis.thirdparty.com.codahale.metrics.Timer; - /** * Metrics Registry for the State Machine Updater. One instance per group. */ @@ -66,7 +65,7 @@ private RatisMetricRegistry getMetricRegistryForStateMachine(String serverId) { RATIS_STATEMACHINE_METRICS, RATIS_STATEMACHINE_METRICS_DESC)); } - public Timer getTakeSnapshotTimer() { + public Timekeeper getTakeSnapshotTimer() { return registry.timer(STATEMACHINE_TAKE_SNAPSHOT_TIMER); } diff --git a/ratis-server/src/main/java/org/apache/ratis/server/impl/StateMachineUpdater.java b/ratis-server/src/main/java/org/apache/ratis/server/impl/StateMachineUpdater.java index bd989a389f..37f97e5892 100644 --- a/ratis-server/src/main/java/org/apache/ratis/server/impl/StateMachineUpdater.java +++ b/ratis-server/src/main/java/org/apache/ratis/server/impl/StateMachineUpdater.java @@ -18,6 +18,7 @@ package org.apache.ratis.server.impl; import org.apache.ratis.conf.RaftProperties; +import org.apache.ratis.metrics.Timekeeper; import org.apache.ratis.proto.RaftProtos.CommitInfoProto; import org.apache.ratis.protocol.Message; import org.apache.ratis.protocol.exceptions.StateMachineException; @@ -47,8 +48,6 @@ import java.util.function.Consumer; import java.util.stream.LongStream; -import org.apache.ratis.thirdparty.com.codahale.metrics.Timer; - /** * This class tracks the log entries that have been committed in a quorum and * applies them to the state machine. We let a separate thread do this work @@ -264,9 +263,9 @@ private void checkAndTakeSnapshot(MemoizedSupplier flushBatchSize) { registry.gauge(RAFT_LOG_SYNC_BATCH_SIZE, () -> flushBatchSize); } - private Timer getTimer(String timerName) { - return registry.timer(timerName); + public UncheckedAutoCloseable startFlushTimer() { + return Timekeeper.start(flushTimer); } - public Timer getFlushTimer() { - return getTimer(RAFT_LOG_FLUSH_TIME); - } - - public Timer getRaftLogSyncTimer() { - return getTimer(RAFT_LOG_SYNC_TIME); + public Timekeeper getSyncTimer() { + return syncTimer; } public void onRaftLogCacheHit() { @@ -132,32 +142,37 @@ public void onRaftLogAppendEntry() { registry.counter(RAFT_LOG_APPEND_ENTRY_COUNT).inc(); } - public Timer getRaftLogAppendEntryTimer() { - return getTimer(RAFT_LOG_APPEND_ENTRY_LATENCY); + public UncheckedAutoCloseable startAppendEntryTimer() { + return Timekeeper.start(appendEntryTimer); } - public Timer getRaftLogQueueTimer() { - return getTimer(RAFT_LOG_TASK_QUEUE_TIME); + public Timekeeper getEnqueuedTimer() { + return enqueuedTimer; } - public Timer getRaftLogEnqueueDelayTimer() { - return getTimer(RAFT_LOG_TASK_ENQUEUE_DELAY); + public UncheckedAutoCloseable startQueuingDelayTimer() { + return Timekeeper.start(queuingDelayTimer); } - public Timer getRaftLogTaskExecutionTimer(String taskName) { - return getTimer(String.format(RAFT_LOG_TASK_EXECUTION_TIME, taskName)); + private final Map, Timekeeper> classMap = new ConcurrentHashMap<>(); + private Timekeeper getTaskExecutionTimer(Class taskClass) { + return getTimer(String.format(RAFT_LOG_TASK_EXECUTION_TIME, + JavaUtils.getClassSimpleName(taskClass).toLowerCase())); + } + public UncheckedAutoCloseable startTaskExecutionTimer(Class taskClass) { + return Timekeeper.start(classMap.computeIfAbsent(taskClass, this::getTaskExecutionTimer)); } - public Timer getRaftLogReadEntryTimer() { - return getTimer(RAFT_LOG_READ_ENTRY_LATENCY); + public Timekeeper getReadEntryTimer() { + return readEntryTimer; } - public Timer getRaftLogLoadSegmentTimer() { - return getTimer(RAFT_LOG_LOAD_SEGMENT_LATENCY); + public UncheckedAutoCloseable startLoadSegmentTimer() { + return Timekeeper.start(loadSegmentTimer); } - public Timer getRaftLogPurgeTimer() { - return getTimer(RAFT_LOG_PURGE_METRIC); + public UncheckedAutoCloseable startPurgeTimer() { + return Timekeeper.start(purgeTimer); } public void onStateMachineDataWriteTimeout() { diff --git a/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/SegmentedRaftLog.java b/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/SegmentedRaftLog.java index 74d6a8c03d..80427ef428 100644 --- a/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/SegmentedRaftLog.java +++ b/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/SegmentedRaftLog.java @@ -18,6 +18,7 @@ package org.apache.ratis.server.raftlog.segmented; import org.apache.ratis.conf.RaftProperties; +import org.apache.ratis.metrics.Timekeeper; import org.apache.ratis.protocol.RaftGroupMemberId; import org.apache.ratis.server.RaftServer; import org.apache.ratis.server.RaftServerConfigKeys; @@ -50,7 +51,7 @@ import java.util.function.Consumer; import java.util.function.LongSupplier; -import org.apache.ratis.thirdparty.com.codahale.metrics.Timer; +import org.apache.ratis.util.UncheckedAutoCloseable; /** * The RaftLog implementation that writes log entries into segmented files in @@ -84,7 +85,7 @@ public class SegmentedRaftLog extends RaftLogBase { */ abstract static class Task { private final CompletableFuture future = new CompletableFuture<>(); - private Timer.Context queueTimerContext; + private Timekeeper.Context queueTimerContext; CompletableFuture getFuture() { return future; @@ -108,7 +109,7 @@ void failed(IOException e) { abstract long getEndIndex(); - void startTimerOnEnqueue(Timer queueTimer) { + void startTimerOnEnqueue(Timekeeper queueTimer) { queueTimerContext = queueTimer.time(); } @@ -237,9 +238,9 @@ private void loadLogSegments(long lastIndexInSnapshot, // so that during the initial loading we can apply part of the log // entries to the state machine boolean keepEntryInCache = (paths.size() - i++) <= cache.getMaxCachedSegments(); - final Timer.Context loadSegmentContext = getRaftLogMetrics().getRaftLogLoadSegmentTimer().time(); - cache.loadSegment(pi, keepEntryInCache, logConsumer); - loadSegmentContext.stop(); + try(UncheckedAutoCloseable ignored = getRaftLogMetrics().startLoadSegmentTimer()) { + cache.loadSegment(pi, keepEntryInCache, logConsumer); + } } // if the largest index is smaller than the last index in snapshot, we do @@ -373,12 +374,12 @@ protected CompletableFuture purgeImpl(long index) { @Override protected CompletableFuture appendEntryImpl(LogEntryProto entry) { - final Timer.Context context = getRaftLogMetrics().getRaftLogAppendEntryTimer().time(); checkLogState(); if (LOG.isTraceEnabled()) { LOG.trace("{}: appendEntry {}", getName(), LogProtoUtils.toLogEntryString(entry)); } - try(AutoCloseableLock writeLock = writeLock()) { + try(AutoCloseableLock writeLock = writeLock(); + UncheckedAutoCloseable ignored = getRaftLogMetrics().startAppendEntryTimer()) { validateLogEntry(entry); final LogSegment currentOpenSegment = cache.getOpenSegment(); if (currentOpenSegment == null) { @@ -417,8 +418,6 @@ protected CompletableFuture appendEntryImpl(LogEntryProto entry) { } catch (Exception e) { LOG.error("{}: Failed to append {}", getName(), LogProtoUtils.toLogEntryString(entry), e); throw e; - } finally { - context.stop(); } } diff --git a/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/SegmentedRaftLogReader.java b/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/SegmentedRaftLogReader.java index 98cd9022d0..7f55094018 100644 --- a/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/SegmentedRaftLogReader.java +++ b/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/SegmentedRaftLogReader.java @@ -18,6 +18,7 @@ package org.apache.ratis.server.raftlog.segmented; import org.apache.ratis.io.CorruptedFileException; +import org.apache.ratis.metrics.Timekeeper; import org.apache.ratis.protocol.exceptions.ChecksumException; import org.apache.ratis.server.metrics.SegmentedRaftLogMetrics; import org.apache.ratis.server.raftlog.RaftLog; @@ -35,8 +36,6 @@ import java.util.Optional; import java.util.zip.Checksum; -import org.apache.ratis.thirdparty.com.codahale.metrics.Timer; - class SegmentedRaftLogReader implements Closeable { static final Logger LOG = LoggerFactory.getLogger(SegmentedRaftLogReader.class); /** @@ -193,11 +192,10 @@ boolean verifyHeader() throws IOException { * exception when skipBrokenEdits is false. */ LogEntryProto readEntry() throws IOException { - Timer.Context readEntryContext = null; - try { - if (raftLogMetrics != null) { - readEntryContext = raftLogMetrics.getRaftLogReadEntryTimer().time(); - } + final Timekeeper timekeeper = Optional.ofNullable(raftLogMetrics) + .map(SegmentedRaftLogMetrics::getReadEntryTimer) + .orElse(null); + try(AutoCloseable readEntryContext = Timekeeper.start(timekeeper)) { return decodeEntry(); } catch (EOFException eof) { in.reset(); @@ -218,10 +216,6 @@ LogEntryProto readEntry() throws IOException { // broken, throw the exception instead of skipping broken entries in.reset(); throw new IOException("got unexpected exception " + e.getMessage(), e); - } finally { - if (readEntryContext != null) { - readEntryContext.stop(); - } } } diff --git a/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/SegmentedRaftLogWorker.java b/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/SegmentedRaftLogWorker.java index bc186d0ea1..ba906ab8f0 100644 --- a/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/SegmentedRaftLogWorker.java +++ b/ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/SegmentedRaftLogWorker.java @@ -17,7 +17,7 @@ */ package org.apache.ratis.server.raftlog.segmented; -import org.apache.ratis.thirdparty.com.codahale.metrics.Timer; +import org.apache.ratis.metrics.Timekeeper; import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; import org.apache.ratis.conf.RaftProperties; import org.apache.ratis.proto.RaftProtos.StateMachineLogEntryProto; @@ -148,10 +148,6 @@ synchronized void updateIndex(long i) { private volatile SegmentedRaftLogOutputStream out; private final Runnable submitUpdateCommitEvent; private final StateMachine stateMachine; - private final Timer logFlushTimer; - private final Timer raftLogSyncTimer; - private final Timer raftLogQueueingTimer; - private final Timer raftLogEnqueueingDelayTimer; private final SegmentedRaftLogMetrics raftLogMetrics; private final ByteBuffer writeBuffer; @@ -211,10 +207,6 @@ synchronized void updateIndex(long i) { metricRegistry.addDataQueueSizeGauge(queue::getNumElements); metricRegistry.addLogWorkerQueueSizeGauge(writeTasks.q::size); metricRegistry.addFlushBatchSizeGauge(() -> flushBatchSize); - this.logFlushTimer = metricRegistry.getFlushTimer(); - this.raftLogSyncTimer = metricRegistry.getRaftLogSyncTimer(); - this.raftLogQueueingTimer = metricRegistry.getRaftLogQueueTimer(); - this.raftLogEnqueueingDelayTimer = metricRegistry.getRaftLogEnqueueDelayTimer(); final int bufferSize = RaftServerConfigKeys.Log.writeBufferSize(properties).getSizeInt(); this.writeBuffer = ByteBuffer.allocateDirect(bufferSize); @@ -275,14 +267,11 @@ public String toString() { */ private Task addIOTask(Task task) { LOG.debug("{} adds IO task {}", name, task); - try { - final Timer.Context enqueueTimerContext = raftLogEnqueueingDelayTimer.time(); + try(UncheckedAutoCloseable ignored = raftLogMetrics.startQueuingDelayTimer()) { for(; !queue.offer(task, ONE_SECOND); ) { Preconditions.assertTrue(isAlive(), "the worker thread is not alive"); } - enqueueTimerContext.stop(); - task.startTimerOnEnqueue(raftLogQueueingTimer); } catch (Exception e) { if (e instanceof InterruptedException && !running) { LOG.info("Got InterruptedException when adding task " + task @@ -292,6 +281,7 @@ private Task addIOTask(Task task) { Optional.ofNullable(server).ifPresent(RaftServer.Division::close); } } + task.startTimerOnEnqueue(raftLogMetrics.getEnqueuedTimer()); return task; } @@ -312,10 +302,9 @@ private void run() { if (logIOException != null) { throw logIOException; } else { - final Timer.Context executionTimeContext = raftLogMetrics.getRaftLogTaskExecutionTimer( - JavaUtils.getClassSimpleName(task.getClass()).toLowerCase()).time(); - task.execute(); - executionTimeContext.stop(); + try (UncheckedAutoCloseable ignored = raftLogMetrics.startTaskExecutionTimer(task.getClass())) { + task.execute(); + } } } catch (IOException e) { if (task.getEndIndex() < lastWrittenIndex) { @@ -370,8 +359,7 @@ private void flushIfNecessary() throws IOException { if (shouldFlush()) { raftLogMetrics.onRaftLogFlush(); LOG.debug("{}: flush {}", name, out); - final Timer.Context timerContext = logFlushTimer.time(); - try { + try(UncheckedAutoCloseable ignored = raftLogMetrics.startFlushTimer()) { final CompletableFuture f = stateMachine != null ? stateMachine.data().flush(lastWrittenIndex) : CompletableFuture.completedFuture(null); @@ -392,19 +380,17 @@ private void flushIfNecessary() throws IOException { } updateFlushedIndexIncreasingly(); } - } finally { - timerContext.stop(); } } } private void unsafeFlushOutStream() throws IOException { - final Timer.Context logSyncTimerContext = raftLogSyncTimer.time(); + final Timekeeper.Context logSyncTimerContext = raftLogMetrics.getSyncTimer().time(); out.asyncFlush(flushExecutor).whenComplete((v, e) -> logSyncTimerContext.stop()); } private void asyncFlushOutStream(CompletableFuture stateMachineFlush) throws IOException { - final Timer.Context logSyncTimerContext = raftLogSyncTimer.time(); + final Timekeeper.Context logSyncTimerContext = raftLogMetrics.getSyncTimer().time(); final CompletableFuture f = out.asyncFlush(flushExecutor) .thenCombine(stateMachineFlush, (async, sm) -> async); flushFuture.updateAndGet(previous -> f.thenCombine(previous, (current, prev) -> current)) @@ -415,11 +401,8 @@ private void asyncFlushOutStream(CompletableFuture stateMachineFlush) thro } private void flushOutStream() throws IOException { - final Timer.Context logSyncTimerContext = raftLogSyncTimer.time(); - try { + try(UncheckedAutoCloseable ignored = Timekeeper.start(raftLogMetrics.getSyncTimer())) { out.flush(); - } finally { - logSyncTimerContext.stop(); } } @@ -428,10 +411,9 @@ private void updateFlushedIndexIncreasingly() { } private void updateFlushedIndexIncreasingly(long index) { - final long i = index; - flushIndex.updateIncreasingly(i, traceIndexChange); + flushIndex.updateIncreasingly(index, traceIndexChange); postUpdateFlushedIndex(Math.toIntExact(lastWrittenIndex - index)); - writeTasks.updateIndex(i); + writeTasks.updateIndex(index); } private void postUpdateFlushedIndex(int count) { @@ -443,7 +425,6 @@ private void postUpdateFlushedIndex(int count) { * The following several methods (startLogSegment, rollLogSegment, * writeLogEntry, and truncate) are only called by SegmentedRaftLog which is * protected by RaftServer's lock. - * * Thus all the tasks are created and added sequentially. */ void startLogSegment(long startIndex) { @@ -487,11 +468,11 @@ private PurgeLog(TruncationSegments segments) { @Override void execute() throws IOException { if (segments.getToDelete() != null) { - Timer.Context purgeLogContext = raftLogMetrics.getRaftLogPurgeTimer().time(); - for (SegmentFileInfo fileInfo : segments.getToDelete()) { - FileUtils.deleteFile(fileInfo.getFile(storage)); + try(UncheckedAutoCloseable ignored = raftLogMetrics.startPurgeTimer()) { + for (SegmentFileInfo fileInfo : segments.getToDelete()) { + FileUtils.deleteFile(fileInfo.getFile(storage)); + } } - purgeLogContext.stop(); } } diff --git a/ratis-server/src/main/java/org/apache/ratis/statemachine/impl/BaseStateMachine.java b/ratis-server/src/main/java/org/apache/ratis/statemachine/impl/BaseStateMachine.java index 629a55a67f..3c20bc5714 100644 --- a/ratis-server/src/main/java/org/apache/ratis/statemachine/impl/BaseStateMachine.java +++ b/ratis-server/src/main/java/org/apache/ratis/statemachine/impl/BaseStateMachine.java @@ -18,7 +18,6 @@ package org.apache.ratis.statemachine.impl; -import org.apache.ratis.thirdparty.com.codahale.metrics.Timer; import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; import org.apache.ratis.proto.RaftProtos; import org.apache.ratis.protocol.Message; @@ -231,19 +230,4 @@ public String toString() { return JavaUtils.getClassSimpleName(getClass()) + ":" + (!server.isDone()? "uninitialized": getId() + ":" + groupId); } - - - protected CompletableFuture recordTime(Timer timer, Task task) { - final Timer.Context timerContext = timer.time(); - try { - return task.run(); - } finally { - timerContext.stop(); - } - } - - protected interface Task { - CompletableFuture run(); - } - } diff --git a/ratis-server/src/test/java/org/apache/ratis/LogAppenderTests.java b/ratis-server/src/test/java/org/apache/ratis/LogAppenderTests.java index c9ef9b06b7..da0c25f9ce 100644 --- a/ratis-server/src/test/java/org/apache/ratis/LogAppenderTests.java +++ b/ratis-server/src/test/java/org/apache/ratis/LogAppenderTests.java @@ -25,6 +25,7 @@ import org.apache.ratis.client.RaftClient; import org.apache.ratis.conf.RaftProperties; import org.apache.ratis.metrics.impl.RatisMetricRegistryImpl; +import org.apache.ratis.metrics.impl.DefaultTimekeeperImpl; import org.apache.ratis.proto.RaftProtos.LogEntryProto; import org.apache.ratis.proto.RaftProtos.LogEntryProto.LogEntryBodyCase; import org.apache.ratis.protocol.RaftPeerId; @@ -163,8 +164,9 @@ public void testFollowerHeartbeatMetric() throws IOException, InterruptedExcepti final RatisMetricRegistryImpl followerMetricRegistry = (RatisMetricRegistryImpl)followerMetrics.getRegistry(); assertTrue(followerMetricRegistry.getGauges((s, m) -> s.contains("lastHeartbeatElapsedTime")).isEmpty()); for (boolean heartbeat : new boolean[] { true, false }) { - assertTrue(followerMetrics.getFollowerAppendEntryTimer(heartbeat).getMeanRate() > 0.0d); - assertTrue(followerMetrics.getFollowerAppendEntryTimer(heartbeat).getCount() > 0L); + final DefaultTimekeeperImpl t = (DefaultTimekeeperImpl) followerMetrics.getFollowerAppendEntryTimer(heartbeat); + assertTrue(t.getTimer().getMeanRate() > 0.0d); + assertTrue(t.getTimer().getCount() > 0L); } } } diff --git a/ratis-server/src/test/java/org/apache/ratis/server/impl/LeaderElectionTests.java b/ratis-server/src/test/java/org/apache/ratis/server/impl/LeaderElectionTests.java index 8929fb8625..bf9938125a 100644 --- a/ratis-server/src/test/java/org/apache/ratis/server/impl/LeaderElectionTests.java +++ b/ratis-server/src/test/java/org/apache/ratis/server/impl/LeaderElectionTests.java @@ -23,6 +23,7 @@ import org.apache.ratis.client.RaftClient; import org.apache.ratis.conf.RaftProperties; import org.apache.ratis.metrics.impl.RatisMetricRegistryImpl; +import org.apache.ratis.metrics.impl.DefaultTimekeeperImpl; import org.apache.ratis.proto.RaftProtos; import org.apache.ratis.protocol.RaftClientReply; import org.apache.ratis.protocol.RaftGroupId; @@ -417,7 +418,8 @@ public void testLeaderElectionMetrics() throws IOException, InterruptedException long numLeaderElectionTimeout = ratisMetricRegistry.counter(LEADER_ELECTION_TIMEOUT_COUNT_METRIC).getCount(); assertTrue(numLeaderElectionTimeout > 0); - Timer timer = ratisMetricRegistry.timer(LEADER_ELECTION_TIME_TAKEN); + final DefaultTimekeeperImpl timekeeper = (DefaultTimekeeperImpl) ratisMetricRegistry.timer(LEADER_ELECTION_TIME_TAKEN); + final Timer timer = timekeeper.getTimer(); double meanTimeNs = timer.getSnapshot().getMean(); long elapsedNs = timestamp.elapsedTime().toLong(TimeUnit.NANOSECONDS); assertTrue(timer.getCount() > 0 && meanTimeNs < elapsedNs); diff --git a/ratis-server/src/test/java/org/apache/ratis/statemachine/RaftSnapshotBaseTest.java b/ratis-server/src/test/java/org/apache/ratis/statemachine/RaftSnapshotBaseTest.java index ce301124f6..1e40943428 100644 --- a/ratis-server/src/test/java/org/apache/ratis/statemachine/RaftSnapshotBaseTest.java +++ b/ratis-server/src/test/java/org/apache/ratis/statemachine/RaftSnapshotBaseTest.java @@ -25,6 +25,7 @@ import org.apache.log4j.Level; import org.apache.ratis.BaseTest; import org.apache.ratis.metrics.LongCounter; +import org.apache.ratis.metrics.impl.DefaultTimekeeperImpl; import org.apache.ratis.server.impl.MiniRaftCluster; import org.apache.ratis.RaftTestUtil; import org.apache.ratis.RaftTestUtil.SimpleMessage; @@ -329,6 +330,6 @@ private static Timer getTakeSnapshotTimer(RaftServer.Division leader) { Assert.assertTrue(opt.isPresent()); RatisMetricRegistry metricRegistry = opt.get(); Assert.assertNotNull(metricRegistry); - return metricRegistry.timer(STATEMACHINE_TAKE_SNAPSHOT_TIMER); + return ((DefaultTimekeeperImpl)metricRegistry.timer(STATEMACHINE_TAKE_SNAPSHOT_TIMER)).getTimer(); } } diff --git a/ratis-test/src/test/java/org/apache/ratis/grpc/TestRaftServerWithGrpc.java b/ratis-test/src/test/java/org/apache/ratis/grpc/TestRaftServerWithGrpc.java index ccdd8474cd..d4dce4b2fc 100644 --- a/ratis-test/src/test/java/org/apache/ratis/grpc/TestRaftServerWithGrpc.java +++ b/ratis-test/src/test/java/org/apache/ratis/grpc/TestRaftServerWithGrpc.java @@ -25,6 +25,7 @@ import org.apache.ratis.server.storage.RaftStorage; import org.apache.log4j.Level; import org.apache.ratis.BaseTest; +import org.apache.ratis.metrics.impl.DefaultTimekeeperImpl; import org.apache.ratis.protocol.RaftGroup; import org.apache.ratis.server.impl.MiniRaftCluster; import org.apache.ratis.RaftTestUtil; @@ -305,24 +306,31 @@ void testRaftClientRequestMetrics(MiniRaftClusterWithGrpc cluster) throws IOExce try (final RaftClient client = cluster.createClient()) { final CompletableFuture f1 = client.async().send(new SimpleMessage("testing")); Assert.assertTrue(f1.get().isSuccess()); - Assert.assertTrue(raftServerMetrics.getTimer(RAFT_CLIENT_WRITE_REQUEST).getCount() > 0); + final DefaultTimekeeperImpl write = (DefaultTimekeeperImpl) raftServerMetrics.getTimer(RAFT_CLIENT_WRITE_REQUEST); + Assert.assertTrue(write.getTimer().getCount() > 0); final CompletableFuture f2 = client.async().sendReadOnly(new SimpleMessage("testing")); Assert.assertTrue(f2.get().isSuccess()); - Assert.assertTrue(raftServerMetrics.getTimer(RAFT_CLIENT_READ_REQUEST).getCount() > 0); + final DefaultTimekeeperImpl read = (DefaultTimekeeperImpl) raftServerMetrics.getTimer(RAFT_CLIENT_READ_REQUEST); + Assert.assertTrue(read.getTimer().getCount() > 0); final CompletableFuture f3 = client.async().sendStaleRead(new SimpleMessage("testing"), 0, leader.getId()); Assert.assertTrue(f3.get().isSuccess()); - Assert.assertTrue(raftServerMetrics.getTimer(RAFT_CLIENT_STALE_READ_REQUEST).getCount() > 0); + final DefaultTimekeeperImpl staleRead = (DefaultTimekeeperImpl) raftServerMetrics.getTimer(RAFT_CLIENT_STALE_READ_REQUEST); + Assert.assertTrue(staleRead.getTimer().getCount() > 0); final CompletableFuture f4 = client.async().watch(0, RaftProtos.ReplicationLevel.ALL); Assert.assertTrue(f4.get().isSuccess()); - Assert.assertTrue(raftServerMetrics.getTimer(String.format(RAFT_CLIENT_WATCH_REQUEST, "-ALL")).getCount() > 0); + final DefaultTimekeeperImpl watchAll = (DefaultTimekeeperImpl) raftServerMetrics.getTimer( + String.format(RAFT_CLIENT_WATCH_REQUEST, "-ALL")); + Assert.assertTrue(watchAll.getTimer().getCount() > 0); final CompletableFuture f5 = client.async().watch(0, RaftProtos.ReplicationLevel.MAJORITY); Assert.assertTrue(f5.get().isSuccess()); - Assert.assertTrue(raftServerMetrics.getTimer(String.format(RAFT_CLIENT_WATCH_REQUEST, "")).getCount() > 0); + final DefaultTimekeeperImpl watch = (DefaultTimekeeperImpl) raftServerMetrics.getTimer( + String.format(RAFT_CLIENT_WATCH_REQUEST, "")); + Assert.assertTrue(watch.getTimer().getCount() > 0); } } diff --git a/ratis-test/src/test/java/org/apache/ratis/server/raftlog/TestRaftLogMetrics.java b/ratis-test/src/test/java/org/apache/ratis/server/raftlog/TestRaftLogMetrics.java index 92a9a90ddf..d58004177e 100644 --- a/ratis-test/src/test/java/org/apache/ratis/server/raftlog/TestRaftLogMetrics.java +++ b/ratis-test/src/test/java/org/apache/ratis/server/raftlog/TestRaftLogMetrics.java @@ -17,6 +17,7 @@ */ package org.apache.ratis.server.raftlog; +import org.apache.ratis.metrics.impl.DefaultTimekeeperImpl; import org.apache.ratis.thirdparty.com.codahale.metrics.Timer; import org.apache.ratis.BaseTest; import org.apache.ratis.RaftTestUtil; @@ -166,17 +167,18 @@ static void assertRaftLogWritePathMetrics(RaftServer.Division server) throws Exc Assert.assertTrue(ratisMetricRegistry.counter(RAFT_LOG_FLUSH_COUNT).getCount() > 0); Assert.assertTrue(ratisMetricRegistry.counter(RAFT_LOG_APPEND_ENTRY_COUNT).getCount() > 0); - Timer appendLatencyTimer = ratisMetricRegistry.timer(RAFT_LOG_APPEND_ENTRY_LATENCY); - Assert.assertTrue(appendLatencyTimer.getMeanRate() > 0); + final DefaultTimekeeperImpl appendEntry = (DefaultTimekeeperImpl) ratisMetricRegistry.timer(RAFT_LOG_APPEND_ENTRY_LATENCY); + Assert.assertTrue(appendEntry.getTimer().getMeanRate() > 0); - Timer enqueuedTimer = ratisMetricRegistry.timer(RAFT_LOG_TASK_QUEUE_TIME); - Assert.assertTrue(enqueuedTimer.getMeanRate() > 0); + final DefaultTimekeeperImpl taskQueue = (DefaultTimekeeperImpl) ratisMetricRegistry.timer(RAFT_LOG_TASK_QUEUE_TIME); + Assert.assertTrue(taskQueue.getTimer().getMeanRate() > 0); - Timer queueingDelayTimer = ratisMetricRegistry.timer(RAFT_LOG_TASK_ENQUEUE_DELAY); - Assert.assertTrue(queueingDelayTimer.getMeanRate() > 0); + final DefaultTimekeeperImpl enqueueDelay = (DefaultTimekeeperImpl) ratisMetricRegistry.timer(RAFT_LOG_TASK_ENQUEUE_DELAY); + Assert.assertTrue(enqueueDelay.getTimer().getMeanRate() > 0); - Timer executionTimer = ratisMetricRegistry.timer(String.format(RAFT_LOG_TASK_EXECUTION_TIME, "writelog")); - Assert.assertTrue(executionTimer.getMeanRate() > 0); + final DefaultTimekeeperImpl write = (DefaultTimekeeperImpl) ratisMetricRegistry.timer( + String.format(RAFT_LOG_TASK_EXECUTION_TIME, "writelog")); + Assert.assertTrue(write.getTimer().getMeanRate() > 0); Assert.assertNotNull(ratisMetricRegistry.get(RAFT_LOG_DATA_QUEUE_SIZE)); Assert.assertNotNull(ratisMetricRegistry.get(RAFT_LOG_WORKER_QUEUE_SIZE)); diff --git a/ratis-test/src/test/java/org/apache/ratis/server/raftlog/segmented/TestLogSegment.java b/ratis-test/src/test/java/org/apache/ratis/server/raftlog/segmented/TestLogSegment.java index ba24f89974..63016dde1e 100644 --- a/ratis-test/src/test/java/org/apache/ratis/server/raftlog/segmented/TestLogSegment.java +++ b/ratis-test/src/test/java/org/apache/ratis/server/raftlog/segmented/TestLogSegment.java @@ -20,6 +20,7 @@ import org.apache.ratis.BaseTest; import org.apache.ratis.RaftTestUtil.SimpleOperation; import org.apache.ratis.conf.RaftProperties; +import org.apache.ratis.metrics.impl.DefaultTimekeeperImpl; import org.apache.ratis.server.RaftServerConfigKeys; import org.apache.ratis.server.impl.RaftServerTestUtil; import org.apache.ratis.server.metrics.SegmentedRaftLogMetrics; @@ -52,8 +53,6 @@ import static org.apache.ratis.server.raftlog.RaftLog.INVALID_LOG_INDEX; import static org.apache.ratis.server.raftlog.segmented.LogSegment.getEntrySize; -import org.apache.ratis.thirdparty.com.codahale.metrics.Timer; - /** * Test basic functionality of {@link LogSegment} */ @@ -219,10 +218,10 @@ public void testAppendEntryMetric() throws Exception { checkLogSegment(openSegment, 0, 98, true, openSegmentFile.length(), 0); storage.close(); - Timer readEntryTimer = raftLogMetrics.getRaftLogReadEntryTimer(); + final DefaultTimekeeperImpl readEntryTimer = (DefaultTimekeeperImpl) raftLogMetrics.getReadEntryTimer(); Assert.assertNotNull(readEntryTimer); - Assert.assertEquals(100, readEntryTimer.getCount()); - Assert.assertTrue(readEntryTimer.getMeanRate() > 0); + Assert.assertEquals(100, readEntryTimer.getTimer().getCount()); + Assert.assertTrue(readEntryTimer.getTimer().getMeanRate() > 0); } diff --git a/ratis-test/src/test/java/org/apache/ratis/server/raftlog/segmented/TestSegmentedRaftLog.java b/ratis-test/src/test/java/org/apache/ratis/server/raftlog/segmented/TestSegmentedRaftLog.java index 2600e351ee..8f7d54ea76 100644 --- a/ratis-test/src/test/java/org/apache/ratis/server/raftlog/segmented/TestSegmentedRaftLog.java +++ b/ratis-test/src/test/java/org/apache/ratis/server/raftlog/segmented/TestSegmentedRaftLog.java @@ -24,6 +24,7 @@ import org.apache.ratis.RaftTestUtil.SimpleOperation; import org.apache.ratis.conf.RaftProperties; import org.apache.ratis.metrics.RatisMetricRegistry; +import org.apache.ratis.metrics.impl.DefaultTimekeeperImpl; import org.apache.ratis.protocol.RaftGroupId; import org.apache.ratis.protocol.RaftGroupMemberId; import org.apache.ratis.protocol.RaftPeerId; @@ -68,7 +69,6 @@ import java.util.function.LongSupplier; import java.util.function.Supplier; -import org.apache.ratis.thirdparty.com.codahale.metrics.Timer; import org.junit.runner.RunWith; import org.junit.runners.Parameterized; @@ -234,11 +234,11 @@ public void testLoadLogSegments() throws Exception { final RatisMetricRegistry metricRegistryForLogWorker = RaftLogMetricsBase.getLogWorkerMetricRegistry(memberId); - Timer raftLogSegmentLoadLatencyTimer = metricRegistryForLogWorker.timer("segmentLoadLatency"); - assertTrue(raftLogSegmentLoadLatencyTimer.getMeanRate() > 0); + final DefaultTimekeeperImpl load = (DefaultTimekeeperImpl) metricRegistryForLogWorker.timer("segmentLoadLatency"); + assertTrue(load.getTimer().getMeanRate() > 0); - Timer raftLogReadLatencyTimer = metricRegistryForLogWorker.timer("readEntryLatency"); - assertTrue(raftLogReadLatencyTimer.getMeanRate() > 0); + final DefaultTimekeeperImpl read = (DefaultTimekeeperImpl) metricRegistryForLogWorker.timer("readEntryLatency"); + assertTrue(read.getTimer().getMeanRate() > 0); } } @@ -484,7 +484,8 @@ public void testPurgeLogMetric() throws Exception { long expectedIndex = segmentSize * (endTerm - startTerm - 1); final RatisMetricRegistry metricRegistryForLogWorker = RaftLogMetricsBase.getLogWorkerMetricRegistry(memberId); purgeAndVerify(startTerm, endTerm, segmentSize, 1, endIndexOfClosedSegment, expectedIndex); - assertTrue(metricRegistryForLogWorker.timer("purgeLog").getCount() > 0); + final DefaultTimekeeperImpl purge = (DefaultTimekeeperImpl) metricRegistryForLogWorker.timer("purgeLog"); + assertTrue(purge.getTimer().getCount() > 0); } @Test