You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@iotdb.apache.org by "SzyWilliam (via GitHub)" <gi...@apache.org> on 2023/10/18 08:53:21 UTC
[PR] [RatisConsensus] Improvements on periodic snapshot guardian [iotdb]
SzyWilliam opened a new pull request, #11342:
URL: https://github.com/apache/iotdb/pull/11342
This patch:
1. add a periodic snapshot checker
2. refactor
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [IOTDB-6197] Improvements on periodic snapshot guardian [iotdb]
Posted by "SzyWilliam (via GitHub)" <gi...@apache.org>.
SzyWilliam commented on code in PR #11342:
URL: https://github.com/apache/iotdb/pull/11342#discussion_r1365465394
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/SchemaRegionConsensusImpl.java:
##########
@@ -130,6 +130,7 @@ private static class SchemaRegionConsensusImplHolder {
.setImpl(
RatisConfig.Impl.newBuilder()
.setTriggerSnapshotFileSize(CONF.getSchemaRatisLogMax())
+ .setForceSnapshotInterval(60 * 60 * 2) // every 2 hrs
Review Comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [IOTDB-6197] Improvements on periodic snapshot guardian [iotdb]
Posted by "OneSizeFitsQuorum (via GitHub)" <gi...@apache.org>.
OneSizeFitsQuorum merged PR #11342:
URL: https://github.com/apache/iotdb/pull/11342
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [IOTDB-6197] Improvements on periodic snapshot guardian [iotdb]
Posted by "OneSizeFitsQuorum (via GitHub)" <gi...@apache.org>.
OneSizeFitsQuorum commented on code in PR #11342:
URL: https://github.com/apache/iotdb/pull/11342#discussion_r1364862486
##########
iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/config/RatisConfig.java:
##########
@@ -982,10 +990,16 @@ public static class Builder {
private long triggerSnapshotTime = 120;
Review Comment:
update name?
##########
iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/config/RatisConfig.java:
##########
@@ -982,10 +990,16 @@ public static class Builder {
private long triggerSnapshotTime = 120;
// 20GB
private long triggerSnapshotFileSize = 20L << 30;
Review Comment:
update name?
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/SchemaRegionConsensusImpl.java:
##########
@@ -130,6 +130,7 @@ private static class SchemaRegionConsensusImplHolder {
.setImpl(
RatisConfig.Impl.newBuilder()
.setTriggerSnapshotFileSize(CONF.getSchemaRatisLogMax())
+ .setForceSnapshotInterval(60 * 60 * 2) // every 2 hrs
Review Comment:
Please extract the corresponding three parameters, the default value is 24h
##########
iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/DiskGuardian.java:
##########
@@ -0,0 +1,246 @@
+/*
+ * 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.iotdb.consensus.ratis;
+
+import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
+import org.apache.iotdb.commons.concurrent.ThreadName;
+import org.apache.iotdb.commons.concurrent.threadpool.ScheduledExecutorUtil;
+import org.apache.iotdb.consensus.config.RatisConfig;
+import org.apache.iotdb.consensus.exception.ConsensusException;
+import org.apache.iotdb.consensus.ratis.utils.Utils;
+
+import org.apache.ratis.protocol.RaftGroupId;
+import org.apache.ratis.util.MemoizedSupplier;
+import org.apache.ratis.util.Preconditions;
+import org.apache.ratis.util.TimeDuration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Predicate;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * DiskGuardian manages to take snapshots periodically for each Raft Group in order to control disk
+ * storage consumption.
+ */
+class DiskGuardian {
+ private static final Logger logger = LoggerFactory.getLogger(DiskGuardian.class);
+
+ /**
+ * A summary of Raft Log disk file snapshot. Containing following summary statistics:
+ *
+ * <p>
+ * <li>1. Disk space consumed by this group's Raft Logs.
+ * <li>2. How many raft log segments exists.
+ */
+ static final class RaftLogSummary {
+
+ /** whether the path denotes an open segment under active writing progress */
+ private static final Predicate<Path> isOpenSegment =
+ p -> p.toFile().getName().startsWith("log_inprogress");
+
+ private final RaftGroupId gid;
+ private final File raftLogStorageRoot;
+ private long totalSize;
+ private Set<Path> logFiles;
+
+ private RaftLogSummary(RaftGroupId gid, File raftLogStorageRoot) {
+ this.gid = gid;
+ this.raftLogStorageRoot = raftLogStorageRoot;
+ this.totalSize = 0;
+ this.logFiles = Collections.emptySet();
+ }
+
+ public long getTotalSize() {
+ return totalSize;
+ }
+
+ void updateNow() {
+ try (final Stream<Path> files = Files.list(raftLogStorageRoot.toPath())) {
+ final Set<Path> latest = files.filter(isOpenSegment).collect(Collectors.toSet());
+ this.totalSize += diff(logFiles, latest);
+ this.logFiles = latest;
+ } catch (IOException e) {
+ // keep the files unchanged
+ logger.warn("{}: Error caught when listing files for {} at {}:", this, gid, e);
+ }
+ }
+
+ private static long diff(Set<Path> old, Set<Path> latest) {
+ final long incremental = totalSize(latest.stream().filter(p -> !old.contains(p)));
+ final long decremental = totalSize(old.stream().filter(p -> !latest.contains(p)));
+ return incremental - decremental;
+ }
+
+ private static long totalSize(Stream<Path> files) {
+ return files.mapToLong(p -> p.toFile().length()).sum();
+ }
+
+ @Override
+ public String toString() {
+ return String.format(
+ "[Raft Log Summary]: group=%s, total size=%d, files=%s", gid, totalSize, logFiles);
+ }
+ }
+
+ /** use a {@link MemoizedSupplier} here to avoid this reference leak */
+ private final MemoizedSupplier<RatisConsensus> serverRef;
+
+ /** whether a specific RaftGroup need to take snapshot during next check period */
+ private final Map<RaftGroupId, AtomicBoolean> snapshotFlag = new ConcurrentHashMap<>();
+
+ /** Raft Log disk usage summary statistics */
+ private final Map<RaftGroupId, RaftLogSummary> bookkeeper = new ConcurrentHashMap<>();
+
+ /** Registered checkers */
+ private final Map<TimeDuration, List<Predicate<RaftLogSummary>>> snapshotArbitrators =
+ new HashMap<>();
+
+ private final ScheduledExecutorService workerThread =
+ IoTDBThreadPoolFactory.newSingleThreadScheduledExecutor(
+ ThreadName.RATIS_BG_DISK_GUARDIAN.getName());
+ private final AtomicBoolean isStopped = new AtomicBoolean(false);
+ private final long daemonIntervalMs;
+
+ DiskGuardian(Supplier<RatisConsensus> server, RatisConfig config) {
+ this.serverRef = MemoizedSupplier.valueOf(server);
+ this.daemonIntervalMs = config.getImpl().getCheckAndTakeSnapshotInterval();
+ }
+
+ void start() {
+ // first schedule the snapshot daemon
+ ScheduledExecutorUtil.safelyScheduleWithFixedDelay(
+ workerThread, this::snapshotDaemon, 0L, daemonIntervalMs, TimeUnit.SECONDS);
+
+ // then schedule all checker daemons
+ snapshotArbitrators.forEach(
+ (interval, checkers) ->
+ ScheduledExecutorUtil.safelyScheduleWithFixedDelay(
+ workerThread,
+ () -> checkerDaemon(checkers),
+ 0L,
+ interval.toLong(TimeUnit.SECONDS),
+ TimeUnit.SECONDS));
+ }
+
+ void stop() throws InterruptedException {
+ if (isStopped.compareAndSet(true, false)) {
+ workerThread.shutdown();
+ workerThread.awaitTermination(5, TimeUnit.SECONDS);
+ }
+ }
+
+ /** call this method to register checkers before {@link #start()} */
+ void registerChecker(Predicate<RaftLogSummary> checker, TimeDuration interval) {
+ final List<Predicate<RaftLogSummary>> checkers =
+ snapshotArbitrators.computeIfAbsent(interval, i -> new CopyOnWriteArrayList<>());
+ checkers.add(checker);
+ }
+
+ /** periodically woken up. Will take a snapshot if a flag is set. */
+ private void snapshotDaemon() {
+ if (isStopped.get()) {
+ return;
+ }
+ for (RaftGroupId groupId : serverRef.get().getServer().getGroupIds()) {
+ if (getSnapshotFlag(groupId).get()) {
+ try {
+ serverRef.get().triggerSnapshot(Utils.fromRaftGroupIdToConsensusGroupId(groupId));
+ final boolean flagCleared = snapshotFlag.get(groupId).compareAndSet(true, false);
+ Preconditions.assertTrue(flagCleared);
Review Comment:
Although this shouldn't be false in theory, maybe it's nicer to log an error in case something goes wrong than to throw an exception?
##########
iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/DiskGuardian.java:
##########
@@ -0,0 +1,246 @@
+/*
+ * 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.iotdb.consensus.ratis;
+
+import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
+import org.apache.iotdb.commons.concurrent.ThreadName;
+import org.apache.iotdb.commons.concurrent.threadpool.ScheduledExecutorUtil;
+import org.apache.iotdb.consensus.config.RatisConfig;
+import org.apache.iotdb.consensus.exception.ConsensusException;
+import org.apache.iotdb.consensus.ratis.utils.Utils;
+
+import org.apache.ratis.protocol.RaftGroupId;
+import org.apache.ratis.util.MemoizedSupplier;
+import org.apache.ratis.util.Preconditions;
+import org.apache.ratis.util.TimeDuration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Predicate;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * DiskGuardian manages to take snapshots periodically for each Raft Group in order to control disk
+ * storage consumption.
+ */
+class DiskGuardian {
+ private static final Logger logger = LoggerFactory.getLogger(DiskGuardian.class);
+
+ /**
+ * A summary of Raft Log disk file snapshot. Containing following summary statistics:
+ *
+ * <p>
+ * <li>1. Disk space consumed by this group's Raft Logs.
+ * <li>2. How many raft log segments exists.
+ */
+ static final class RaftLogSummary {
+
+ /** whether the path denotes an open segment under active writing progress */
+ private static final Predicate<Path> isOpenSegment =
+ p -> p.toFile().getName().startsWith("log_inprogress");
+
+ private final RaftGroupId gid;
+ private final File raftLogStorageRoot;
+ private long totalSize;
+ private Set<Path> logFiles;
+
+ private RaftLogSummary(RaftGroupId gid, File raftLogStorageRoot) {
+ this.gid = gid;
+ this.raftLogStorageRoot = raftLogStorageRoot;
+ this.totalSize = 0;
+ this.logFiles = Collections.emptySet();
+ }
+
+ public long getTotalSize() {
+ return totalSize;
+ }
+
+ void updateNow() {
+ try (final Stream<Path> files = Files.list(raftLogStorageRoot.toPath())) {
+ final Set<Path> latest = files.filter(isOpenSegment).collect(Collectors.toSet());
+ this.totalSize += diff(logFiles, latest);
+ this.logFiles = latest;
+ } catch (IOException e) {
+ // keep the files unchanged
+ logger.warn("{}: Error caught when listing files for {} at {}:", this, gid, e);
+ }
+ }
+
+ private static long diff(Set<Path> old, Set<Path> latest) {
+ final long incremental = totalSize(latest.stream().filter(p -> !old.contains(p)));
+ final long decremental = totalSize(old.stream().filter(p -> !latest.contains(p)));
+ return incremental - decremental;
+ }
+
+ private static long totalSize(Stream<Path> files) {
+ return files.mapToLong(p -> p.toFile().length()).sum();
+ }
+
+ @Override
+ public String toString() {
+ return String.format(
+ "[Raft Log Summary]: group=%s, total size=%d, files=%s", gid, totalSize, logFiles);
+ }
+ }
+
+ /** use a {@link MemoizedSupplier} here to avoid this reference leak */
+ private final MemoizedSupplier<RatisConsensus> serverRef;
+
+ /** whether a specific RaftGroup need to take snapshot during next check period */
+ private final Map<RaftGroupId, AtomicBoolean> snapshotFlag = new ConcurrentHashMap<>();
+
+ /** Raft Log disk usage summary statistics */
+ private final Map<RaftGroupId, RaftLogSummary> bookkeeper = new ConcurrentHashMap<>();
+
+ /** Registered checkers */
+ private final Map<TimeDuration, List<Predicate<RaftLogSummary>>> snapshotArbitrators =
+ new HashMap<>();
+
+ private final ScheduledExecutorService workerThread =
+ IoTDBThreadPoolFactory.newSingleThreadScheduledExecutor(
+ ThreadName.RATIS_BG_DISK_GUARDIAN.getName());
+ private final AtomicBoolean isStopped = new AtomicBoolean(false);
+ private final long daemonIntervalMs;
+
+ DiskGuardian(Supplier<RatisConsensus> server, RatisConfig config) {
+ this.serverRef = MemoizedSupplier.valueOf(server);
+ this.daemonIntervalMs = config.getImpl().getCheckAndTakeSnapshotInterval();
+ }
+
+ void start() {
+ // first schedule the snapshot daemon
+ ScheduledExecutorUtil.safelyScheduleWithFixedDelay(
+ workerThread, this::snapshotDaemon, 0L, daemonIntervalMs, TimeUnit.SECONDS);
+
+ // then schedule all checker daemons
+ snapshotArbitrators.forEach(
+ (interval, checkers) ->
+ ScheduledExecutorUtil.safelyScheduleWithFixedDelay(
+ workerThread,
+ () -> checkerDaemon(checkers),
+ 0L,
+ interval.toLong(TimeUnit.SECONDS),
+ TimeUnit.SECONDS));
+ }
+
+ void stop() throws InterruptedException {
+ if (isStopped.compareAndSet(true, false)) {
+ workerThread.shutdown();
+ workerThread.awaitTermination(5, TimeUnit.SECONDS);
+ }
+ }
+
+ /** call this method to register checkers before {@link #start()} */
+ void registerChecker(Predicate<RaftLogSummary> checker, TimeDuration interval) {
+ final List<Predicate<RaftLogSummary>> checkers =
+ snapshotArbitrators.computeIfAbsent(interval, i -> new CopyOnWriteArrayList<>());
+ checkers.add(checker);
+ }
+
+ /** periodically woken up. Will take a snapshot if a flag is set. */
+ private void snapshotDaemon() {
+ if (isStopped.get()) {
+ return;
+ }
+ for (RaftGroupId groupId : serverRef.get().getServer().getGroupIds()) {
+ if (getSnapshotFlag(groupId).get()) {
+ try {
+ serverRef.get().triggerSnapshot(Utils.fromRaftGroupIdToConsensusGroupId(groupId));
+ final boolean flagCleared = snapshotFlag.get(groupId).compareAndSet(true, false);
+ Preconditions.assertTrue(flagCleared);
+ } catch (ConsensusException e) {
+ logger.warn(
+ "{} take snapshot failed for group {} due to {}. Disk file status {}",
+ this,
+ groupId,
+ e,
+ getLatestSummary(groupId).orElse(null));
+ }
+ }
+ }
+ }
+
+ /** periodically woken up. Will check the snapshot condition and set the snapshot flag. */
+ private void checkerDaemon(List<Predicate<RaftLogSummary>> checkerList) {
+ if (isStopped.get()) {
+ return;
+ }
+ for (RaftGroupId groupId : serverRef.get().getServer().getGroupIds()) {
+ final Optional<RaftLogSummary> summary = getLatestSummary(groupId);
+ if (summary.isPresent()) {
+ final Optional<Boolean> anyCheckerPositive =
+ checkerList.stream()
+ .map(checker -> checker.test(summary.get()))
+ .filter(Boolean::booleanValue)
+ .findAny();
+ if (anyCheckerPositive.isPresent()) {
+ getSnapshotFlag(groupId).set(true);
+ }
+ }
+ }
+ }
+
+ private AtomicBoolean getSnapshotFlag(RaftGroupId groupId) {
+ return snapshotFlag.computeIfAbsent(groupId, id -> new AtomicBoolean(false));
+ }
+
+ private Optional<RaftLogSummary> getLatestSummary(RaftGroupId groupId) {
+ // initialize the RaftLog Summary for the first time
+ if (!bookkeeper.containsKey(groupId)) {
+ synchronized (bookkeeper) {
Review Comment:
use compute API to ensure atomic? You can ask ChatGPT to optimize your code here
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [IOTDB-6197] Improvements on periodic snapshot guardian [iotdb]
Posted by "SzyWilliam (via GitHub)" <gi...@apache.org>.
SzyWilliam commented on code in PR #11342:
URL: https://github.com/apache/iotdb/pull/11342#discussion_r1365412992
##########
iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/config/RatisConfig.java:
##########
@@ -982,10 +990,16 @@ public static class Builder {
private long triggerSnapshotTime = 120;
// 20GB
private long triggerSnapshotFileSize = 20L << 30;
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [IOTDB-6197] Improvements on periodic snapshot guardian [iotdb]
Posted by "SzyWilliam (via GitHub)" <gi...@apache.org>.
SzyWilliam commented on code in PR #11342:
URL: https://github.com/apache/iotdb/pull/11342#discussion_r1366596186
##########
iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/DiskGuardian.java:
##########
@@ -0,0 +1,246 @@
+/*
+ * 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.iotdb.consensus.ratis;
+
+import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
+import org.apache.iotdb.commons.concurrent.ThreadName;
+import org.apache.iotdb.commons.concurrent.threadpool.ScheduledExecutorUtil;
+import org.apache.iotdb.consensus.config.RatisConfig;
+import org.apache.iotdb.consensus.exception.ConsensusException;
+import org.apache.iotdb.consensus.ratis.utils.Utils;
+
+import org.apache.ratis.protocol.RaftGroupId;
+import org.apache.ratis.util.MemoizedSupplier;
+import org.apache.ratis.util.Preconditions;
+import org.apache.ratis.util.TimeDuration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Predicate;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * DiskGuardian manages to take snapshots periodically for each Raft Group in order to control disk
+ * storage consumption.
+ */
+class DiskGuardian {
+ private static final Logger logger = LoggerFactory.getLogger(DiskGuardian.class);
+
+ /**
+ * A summary of Raft Log disk file snapshot. Containing following summary statistics:
+ *
+ * <p>
+ * <li>1. Disk space consumed by this group's Raft Logs.
+ * <li>2. How many raft log segments exists.
+ */
+ static final class RaftLogSummary {
+
+ /** whether the path denotes an open segment under active writing progress */
+ private static final Predicate<Path> isOpenSegment =
+ p -> p.toFile().getName().startsWith("log_inprogress");
+
+ private final RaftGroupId gid;
+ private final File raftLogStorageRoot;
+ private long totalSize;
+ private Set<Path> logFiles;
+
+ private RaftLogSummary(RaftGroupId gid, File raftLogStorageRoot) {
+ this.gid = gid;
+ this.raftLogStorageRoot = raftLogStorageRoot;
+ this.totalSize = 0;
+ this.logFiles = Collections.emptySet();
+ }
+
+ public long getTotalSize() {
+ return totalSize;
+ }
+
+ void updateNow() {
+ try (final Stream<Path> files = Files.list(raftLogStorageRoot.toPath())) {
+ final Set<Path> latest = files.filter(isOpenSegment).collect(Collectors.toSet());
+ this.totalSize += diff(logFiles, latest);
+ this.logFiles = latest;
+ } catch (IOException e) {
+ // keep the files unchanged
+ logger.warn("{}: Error caught when listing files for {} at {}:", this, gid, e);
+ }
+ }
+
+ private static long diff(Set<Path> old, Set<Path> latest) {
+ final long incremental = totalSize(latest.stream().filter(p -> !old.contains(p)));
+ final long decremental = totalSize(old.stream().filter(p -> !latest.contains(p)));
+ return incremental - decremental;
+ }
+
+ private static long totalSize(Stream<Path> files) {
+ return files.mapToLong(p -> p.toFile().length()).sum();
+ }
+
+ @Override
+ public String toString() {
+ return String.format(
+ "[Raft Log Summary]: group=%s, total size=%d, files=%s", gid, totalSize, logFiles);
+ }
+ }
+
+ /** use a {@link MemoizedSupplier} here to avoid this reference leak */
+ private final MemoizedSupplier<RatisConsensus> serverRef;
+
+ /** whether a specific RaftGroup need to take snapshot during next check period */
+ private final Map<RaftGroupId, AtomicBoolean> snapshotFlag = new ConcurrentHashMap<>();
+
+ /** Raft Log disk usage summary statistics */
+ private final Map<RaftGroupId, RaftLogSummary> bookkeeper = new ConcurrentHashMap<>();
+
+ /** Registered checkers */
+ private final Map<TimeDuration, List<Predicate<RaftLogSummary>>> snapshotArbitrators =
+ new HashMap<>();
+
+ private final ScheduledExecutorService workerThread =
+ IoTDBThreadPoolFactory.newSingleThreadScheduledExecutor(
+ ThreadName.RATIS_BG_DISK_GUARDIAN.getName());
+ private final AtomicBoolean isStopped = new AtomicBoolean(false);
+ private final long daemonIntervalMs;
+
+ DiskGuardian(Supplier<RatisConsensus> server, RatisConfig config) {
+ this.serverRef = MemoizedSupplier.valueOf(server);
+ this.daemonIntervalMs = config.getImpl().getCheckAndTakeSnapshotInterval();
+ }
+
+ void start() {
+ // first schedule the snapshot daemon
+ ScheduledExecutorUtil.safelyScheduleWithFixedDelay(
+ workerThread, this::snapshotDaemon, 0L, daemonIntervalMs, TimeUnit.SECONDS);
+
+ // then schedule all checker daemons
+ snapshotArbitrators.forEach(
+ (interval, checkers) ->
+ ScheduledExecutorUtil.safelyScheduleWithFixedDelay(
+ workerThread,
+ () -> checkerDaemon(checkers),
+ 0L,
+ interval.toLong(TimeUnit.SECONDS),
+ TimeUnit.SECONDS));
+ }
+
+ void stop() throws InterruptedException {
+ if (isStopped.compareAndSet(true, false)) {
+ workerThread.shutdown();
+ workerThread.awaitTermination(5, TimeUnit.SECONDS);
+ }
+ }
+
+ /** call this method to register checkers before {@link #start()} */
+ void registerChecker(Predicate<RaftLogSummary> checker, TimeDuration interval) {
+ final List<Predicate<RaftLogSummary>> checkers =
+ snapshotArbitrators.computeIfAbsent(interval, i -> new CopyOnWriteArrayList<>());
+ checkers.add(checker);
+ }
+
+ /** periodically woken up. Will take a snapshot if a flag is set. */
+ private void snapshotDaemon() {
+ if (isStopped.get()) {
+ return;
+ }
+ for (RaftGroupId groupId : serverRef.get().getServer().getGroupIds()) {
+ if (getSnapshotFlag(groupId).get()) {
+ try {
+ serverRef.get().triggerSnapshot(Utils.fromRaftGroupIdToConsensusGroupId(groupId));
+ final boolean flagCleared = snapshotFlag.get(groupId).compareAndSet(true, false);
+ Preconditions.assertTrue(flagCleared);
+ } catch (ConsensusException e) {
+ logger.warn(
+ "{} take snapshot failed for group {} due to {}. Disk file status {}",
+ this,
+ groupId,
+ e,
+ getLatestSummary(groupId).orElse(null));
+ }
+ }
+ }
+ }
+
+ /** periodically woken up. Will check the snapshot condition and set the snapshot flag. */
+ private void checkerDaemon(List<Predicate<RaftLogSummary>> checkerList) {
+ if (isStopped.get()) {
+ return;
+ }
+ for (RaftGroupId groupId : serverRef.get().getServer().getGroupIds()) {
+ final Optional<RaftLogSummary> summary = getLatestSummary(groupId);
+ if (summary.isPresent()) {
+ final Optional<Boolean> anyCheckerPositive =
+ checkerList.stream()
+ .map(checker -> checker.test(summary.get()))
+ .filter(Boolean::booleanValue)
+ .findAny();
+ if (anyCheckerPositive.isPresent()) {
+ getSnapshotFlag(groupId).set(true);
+ }
+ }
+ }
+ }
+
+ private AtomicBoolean getSnapshotFlag(RaftGroupId groupId) {
+ return snapshotFlag.computeIfAbsent(groupId, id -> new AtomicBoolean(false));
+ }
+
+ private Optional<RaftLogSummary> getLatestSummary(RaftGroupId groupId) {
+ // initialize the RaftLog Summary for the first time
+ if (!bookkeeper.containsKey(groupId)) {
+ synchronized (bookkeeper) {
Review Comment:
Good to know.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [IOTDB-6197] Improvements on periodic snapshot guardian [iotdb]
Posted by "SzyWilliam (via GitHub)" <gi...@apache.org>.
SzyWilliam commented on code in PR #11342:
URL: https://github.com/apache/iotdb/pull/11342#discussion_r1365468721
##########
iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/DiskGuardian.java:
##########
@@ -0,0 +1,246 @@
+/*
+ * 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.iotdb.consensus.ratis;
+
+import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
+import org.apache.iotdb.commons.concurrent.ThreadName;
+import org.apache.iotdb.commons.concurrent.threadpool.ScheduledExecutorUtil;
+import org.apache.iotdb.consensus.config.RatisConfig;
+import org.apache.iotdb.consensus.exception.ConsensusException;
+import org.apache.iotdb.consensus.ratis.utils.Utils;
+
+import org.apache.ratis.protocol.RaftGroupId;
+import org.apache.ratis.util.MemoizedSupplier;
+import org.apache.ratis.util.Preconditions;
+import org.apache.ratis.util.TimeDuration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Predicate;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * DiskGuardian manages to take snapshots periodically for each Raft Group in order to control disk
+ * storage consumption.
+ */
+class DiskGuardian {
+ private static final Logger logger = LoggerFactory.getLogger(DiskGuardian.class);
+
+ /**
+ * A summary of Raft Log disk file snapshot. Containing following summary statistics:
+ *
+ * <p>
+ * <li>1. Disk space consumed by this group's Raft Logs.
+ * <li>2. How many raft log segments exists.
+ */
+ static final class RaftLogSummary {
+
+ /** whether the path denotes an open segment under active writing progress */
+ private static final Predicate<Path> isOpenSegment =
+ p -> p.toFile().getName().startsWith("log_inprogress");
+
+ private final RaftGroupId gid;
+ private final File raftLogStorageRoot;
+ private long totalSize;
+ private Set<Path> logFiles;
+
+ private RaftLogSummary(RaftGroupId gid, File raftLogStorageRoot) {
+ this.gid = gid;
+ this.raftLogStorageRoot = raftLogStorageRoot;
+ this.totalSize = 0;
+ this.logFiles = Collections.emptySet();
+ }
+
+ public long getTotalSize() {
+ return totalSize;
+ }
+
+ void updateNow() {
+ try (final Stream<Path> files = Files.list(raftLogStorageRoot.toPath())) {
+ final Set<Path> latest = files.filter(isOpenSegment).collect(Collectors.toSet());
+ this.totalSize += diff(logFiles, latest);
+ this.logFiles = latest;
+ } catch (IOException e) {
+ // keep the files unchanged
+ logger.warn("{}: Error caught when listing files for {} at {}:", this, gid, e);
+ }
+ }
+
+ private static long diff(Set<Path> old, Set<Path> latest) {
+ final long incremental = totalSize(latest.stream().filter(p -> !old.contains(p)));
+ final long decremental = totalSize(old.stream().filter(p -> !latest.contains(p)));
+ return incremental - decremental;
+ }
+
+ private static long totalSize(Stream<Path> files) {
+ return files.mapToLong(p -> p.toFile().length()).sum();
+ }
+
+ @Override
+ public String toString() {
+ return String.format(
+ "[Raft Log Summary]: group=%s, total size=%d, files=%s", gid, totalSize, logFiles);
+ }
+ }
+
+ /** use a {@link MemoizedSupplier} here to avoid this reference leak */
+ private final MemoizedSupplier<RatisConsensus> serverRef;
+
+ /** whether a specific RaftGroup need to take snapshot during next check period */
+ private final Map<RaftGroupId, AtomicBoolean> snapshotFlag = new ConcurrentHashMap<>();
+
+ /** Raft Log disk usage summary statistics */
+ private final Map<RaftGroupId, RaftLogSummary> bookkeeper = new ConcurrentHashMap<>();
+
+ /** Registered checkers */
+ private final Map<TimeDuration, List<Predicate<RaftLogSummary>>> snapshotArbitrators =
+ new HashMap<>();
+
+ private final ScheduledExecutorService workerThread =
+ IoTDBThreadPoolFactory.newSingleThreadScheduledExecutor(
+ ThreadName.RATIS_BG_DISK_GUARDIAN.getName());
+ private final AtomicBoolean isStopped = new AtomicBoolean(false);
+ private final long daemonIntervalMs;
+
+ DiskGuardian(Supplier<RatisConsensus> server, RatisConfig config) {
+ this.serverRef = MemoizedSupplier.valueOf(server);
+ this.daemonIntervalMs = config.getImpl().getCheckAndTakeSnapshotInterval();
+ }
+
+ void start() {
+ // first schedule the snapshot daemon
+ ScheduledExecutorUtil.safelyScheduleWithFixedDelay(
+ workerThread, this::snapshotDaemon, 0L, daemonIntervalMs, TimeUnit.SECONDS);
+
+ // then schedule all checker daemons
+ snapshotArbitrators.forEach(
+ (interval, checkers) ->
+ ScheduledExecutorUtil.safelyScheduleWithFixedDelay(
+ workerThread,
+ () -> checkerDaemon(checkers),
+ 0L,
+ interval.toLong(TimeUnit.SECONDS),
+ TimeUnit.SECONDS));
+ }
+
+ void stop() throws InterruptedException {
+ if (isStopped.compareAndSet(true, false)) {
+ workerThread.shutdown();
+ workerThread.awaitTermination(5, TimeUnit.SECONDS);
+ }
+ }
+
+ /** call this method to register checkers before {@link #start()} */
+ void registerChecker(Predicate<RaftLogSummary> checker, TimeDuration interval) {
+ final List<Predicate<RaftLogSummary>> checkers =
+ snapshotArbitrators.computeIfAbsent(interval, i -> new CopyOnWriteArrayList<>());
+ checkers.add(checker);
+ }
+
+ /** periodically woken up. Will take a snapshot if a flag is set. */
+ private void snapshotDaemon() {
+ if (isStopped.get()) {
+ return;
+ }
+ for (RaftGroupId groupId : serverRef.get().getServer().getGroupIds()) {
+ if (getSnapshotFlag(groupId).get()) {
+ try {
+ serverRef.get().triggerSnapshot(Utils.fromRaftGroupIdToConsensusGroupId(groupId));
+ final boolean flagCleared = snapshotFlag.get(groupId).compareAndSet(true, false);
+ Preconditions.assertTrue(flagCleared);
Review Comment:
It's your call, captain.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org