This is an automated email from the ASF dual-hosted git repository.
NSAmelchev pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/master by this push:
new b502bad1a30 IGNITE-28668 Keep snapshot creation metrics on each node
(#13119)
b502bad1a30 is described below
commit b502bad1a30ca103f3ee085a78a408808447f46b
Author: Vladimir Steshin <[email protected]>
AuthorDate: Fri Jul 3 10:45:16 2026 +0300
IGNITE-28668 Keep snapshot creation metrics on each node (#13119)
---
.../apache/ignite/util/GridCommandHandlerTest.java | 9 +-
.../snapshot/IgniteSnapshotManager.java | 28 +++--
.../snapshot/IgniteClusterSnapshotSelfTest.java | 124 ++++++++++++---------
.../IncrementalSnapshotWarnAtomicCachesTest.java | 57 +++++-----
4 files changed, 121 insertions(+), 97 deletions(-)
diff --git
a/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerTest.java
b/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerTest.java
index 1baad66b3c5..872fb04cce0 100644
---
a/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerTest.java
+++
b/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerTest.java
@@ -3163,11 +3163,14 @@ public class GridCommandHandlerTest extends
GridCommandHandlerClusterPerMethodAb
MetricRegistry metrics =
ig.context().metric().registry(SNAPSHOT_METRICS);
- LongMetric opEndTimeMetric = metrics.findMetric("LastSnapshotEndTime");
- BooleanSupplier endTimeMetricPredicate = () -> opEndTimeMetric.value()
> 0;
+ assertTrue(waitForCondition(() -> metrics.findMetric("LastRequestId")
!= null, getTestTimeout(), 50));
String reqId = metrics.findMetric("LastRequestId").getAsString();
- assertFalse(F.isEmpty(reqId));
+
+ assert reqId != null;
+
+ LongMetric opEndTimeMetric = metrics.findMetric("LastSnapshotEndTime");
+ BooleanSupplier endTimeMetricPredicate = () -> opEndTimeMetric.value()
> 0;
// Make sure the operation ID has been shown to the user.
assertContains(log, testOut.toString(), reqId);
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java
index 365886b366d..c6e25c2baa3 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java
@@ -773,6 +773,14 @@ public class IgniteSnapshotManager extends
GridCacheSharedManagerAdapter
"Another snapshot operation in progress [req=" + req + ",
curr=" + curSnpOp + ']'));
}
+ // Keeps the metrics on each server node.
+ if (!cctx.localNode().isClient()) {
+ if (clusterSnpFut == null)
+ clusterSnpFut = new ClusterSnapshotFuture(req.reqId,
req.snpName, req.incremental() ? req.incrementIndex() : null);
+
+ lastFuture(req.incremental(), clusterSnpFut);
+ }
+
SnapshotOperation snpOp = new SnapshotOperation(req,
new SnapshotFileTree(cctx.kernalContext(), req.snapshotName(),
req.snapshotPath()));
@@ -2050,11 +2058,6 @@ public class IgniteSnapshotManager extends
GridCacheSharedManagerAdapter
snpFut0 = new ClusterSnapshotFuture(UUID.randomUUID(), name,
incIdx);
clusterSnpFut = snpFut0;
-
- if (incremental)
- lastSeenIncSnpFut = snpFut0;
- else
- lastSeenSnpFut = snpFut0;
}
Set<String> cacheGrpNames0 = cacheGrpNames == null ? null : new
HashSet<>(cacheGrpNames);
@@ -2131,17 +2134,18 @@ public class IgniteSnapshotManager extends
GridCacheSharedManagerAdapter
U.error(log, SNAPSHOT_FAILED_MSG, e);
- ClusterSnapshotFuture errSnpFut = new ClusterSnapshotFuture(name,
e);
-
- if (incremental)
- lastSeenIncSnpFut = errSnpFut;
- else
- lastSeenSnpFut = errSnpFut;
-
return new IgniteFinishedFutureImpl<>(e);
}
}
+ /** Sets last seen snapshot future. */
+ private void lastFuture(boolean incremental, ClusterSnapshotFuture
futToSet) {
+ if (incremental)
+ lastSeenIncSnpFut = futToSet;
+ else
+ lastSeenSnpFut = futToSet;
+ }
+
/** Writes a warning message if an incremental snapshot contains atomic
caches. */
void warnAtomicCachesInIncrementalSnapshot(String snpName, int incIdx,
Collection<String> cacheGrps) {
List<String> warnCaches = new ArrayList<>();
diff --git
a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotSelfTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotSelfTest.java
index ee23e278a7d..570761dac22 100644
---
a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotSelfTest.java
+++
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotSelfTest.java
@@ -849,81 +849,99 @@ public class IgniteClusterSnapshotSelfTest extends
AbstractSnapshotSelfTest {
/** @throws Exception If fails. */
@Test
public void testClusterSnapshotMetrics() throws Exception {
- String newSnapshotName = SNAPSHOT_NAME + "_new";
CountDownLatch deltaApply = new CountDownLatch(1);
CountDownLatch deltaBlock = new CountDownLatch(1);
IgniteEx ignite = startGridsWithCache(2, dfltCacheCfg,
CACHE_KEYS_RANGE);
- MetricRegistry mreg0 =
ignite.context().metric().registry(SNAPSHOT_METRICS);
-
- LongMetric startTime = mreg0.findMetric("LastSnapshotStartTime");
- LongMetric endTime = mreg0.findMetric("LastSnapshotEndTime");
- ObjectGauge<String> snpName = mreg0.findMetric("LastSnapshotName");
- ObjectGauge<String> errMsg =
mreg0.findMetric("LastSnapshotErrorMessage");
- ObjectGauge<List<String>> snpList =
mreg0.findMetric("LocalSnapshotNames");
+ startClientGrid();
// Snapshot process will be blocked when delta partition files
processing starts.
snp(ignite).localSnapshotSenderFactory(
blockingLocalSnapshotSender(ignite, deltaApply, deltaBlock));
- assertEquals("Snapshot start time must be undefined prior to snapshot
operation started.",
- 0, startTime.value());
- assertEquals("Snapshot end time must be undefined to snapshot
operation started.",
- 0, endTime.value());
- assertTrue("Snapshot name must not exist prior to snapshot operation
started.", snpName.value().isEmpty());
- assertTrue("Snapshot error message must null prior to snapshot
operation started.", errMsg.value().isEmpty());
- assertTrue("Snapshots on local node must not exist",
snpList.value().isEmpty());
+ for (Ignite g : G.allGrids()) {
+ MetricRegistry mreg =
((IgniteEx)g).context().metric().registry(SNAPSHOT_METRICS);
+
+ LongMetric startTime = mreg.findMetric("LastSnapshotStartTime");
+ LongMetric endTime = mreg.findMetric("LastSnapshotEndTime");
+ ObjectGauge<String> snpName = mreg.findMetric("LastSnapshotName");
+ ObjectGauge<String> errMsg =
mreg.findMetric("LastSnapshotErrorMessage");
+ ObjectGauge<List<String>> snpList =
mreg.findMetric("LocalSnapshotNames");
+
+ if (g.cluster().localNode().isClient()) {
+ assertNull("Snapshot start time must not be created on a
client node.", startTime);
+ assertNull("Snapshot end time must not be created on a client
node.", endTime);
+ assertNull("Snapshot name must not be created on a client
node.", snpName);
+ assertNull("Snapshot error message must not be created on a
client node.", errMsg);
+ assertNull("Snapshot local names must not be created on a
client node.", snpList);
+
+ continue;
+ }
+
+ assertEquals("Snapshot start time must be undefined prior to
snapshot operation started.",
+ 0, startTime.value());
+ assertEquals("Snapshot end time must be undefined to snapshot
operation started.",
+ 0, endTime.value());
+ assertTrue("Snapshot name must not exist prior to snapshot
operation started.", snpName.value().isEmpty());
+ assertTrue("Snapshot error message must null prior to snapshot
operation started.", errMsg.value().isEmpty());
+ assertTrue("Snapshots on local node must not exist",
snpList.value().isEmpty());
+ }
long cutoffStartTime = U.currentTimeMillis();
- IgniteFuture<Void> fut0 = snp(ignite).createSnapshot(SNAPSHOT_NAME,
null, false, onlyPrimary);
+ IgniteFuture<Void> fut = snp(ignite).createSnapshot(SNAPSHOT_NAME,
null, false, onlyPrimary);
+
+ for (Ignite g : G.allGrids()) {
+ MetricRegistry mreg =
((IgniteEx)g).context().metric().registry(SNAPSHOT_METRICS);
- U.await(deltaApply);
+ LongMetric startTime = mreg.findMetric("LastSnapshotStartTime");
+ LongMetric endTime = mreg.findMetric("LastSnapshotEndTime");
+ ObjectGauge<String> snpName = mreg.findMetric("LastSnapshotName");
+ ObjectGauge<String> errMsg =
mreg.findMetric("LastSnapshotErrorMessage");
- assertTrue("Snapshot start time must be set prior to snapshot
operation started " +
- "[startTime=" + startTime.value() + ", cutoffTime=" +
cutoffStartTime + ']',
- startTime.value() >= cutoffStartTime);
- assertEquals("Snapshot end time must be zero prior to snapshot
operation started.",
- 0, endTime.value());
- assertEquals("Snapshot name must be set prior to snapshot operation
started.",
- SNAPSHOT_NAME, snpName.value());
- assertTrue("Snapshot error message must null prior to snapshot
operation started.",
- errMsg.value().isEmpty());
+ if (g.cluster().localNode().isClient()) {
+ assertNull("Snapshot start time must not be created on a
client node.", startTime);
+ assertNull("Snapshot end time must not be created on a client
node.", endTime);
+ assertNull("Snapshot name must not be created on a client
node.", snpName);
+ assertNull("Snapshot error message must not be created on a
client node.", errMsg);
- IgniteFuture<Void> fut1 = snp(grid(1)).createSnapshot(newSnapshotName,
null, false, onlyPrimary);
+ continue;
+ }
- assertThrowsWithCause((Callable<Object>)fut1::get,
IgniteException.class);
+ assertTrue("Snapshot start time must be set prior to snapshot
operation started " +
+ "[startTime=" + startTime.value() + ", cutoffTime=" +
cutoffStartTime + ']',
+ waitForCondition(() -> startTime.value() >= cutoffStartTime,
getTestTimeout(), 30));
- MetricRegistry mreg1 =
grid(1).context().metric().registry(SNAPSHOT_METRICS);
+ assertTrue("Snapshot end time must be zero prior to snapshot
operation started.",
+ waitForCondition(() -> 0 == endTime.value(), 30));
- LongMetric startTime1 = mreg1.findMetric("LastSnapshotStartTime");
- LongMetric endTime1 = mreg1.findMetric("LastSnapshotEndTime");
- ObjectGauge<String> snpName1 = mreg1.findMetric("LastSnapshotName");
- ObjectGauge<String> errMsg1 =
mreg1.findMetric("LastSnapshotErrorMessage");
+ assertEquals("Snapshot name must be set prior to snapshot
operation started.",
+ SNAPSHOT_NAME, snpName.value());
- assertTrue("Snapshot start time must be greater than zero for finished
snapshot.",
- startTime1.value() > 0);
- assertEquals("Snapshot end time must zero for failed on start
snapshots.",
- 0, endTime1.value());
- assertEquals("Snapshot name must be set when snapshot operation
already finished.",
- newSnapshotName, snpName1.value());
- assertNotNull("Concurrent snapshot operation must failed.",
- errMsg1.value());
+ assertTrue("Snapshot error message must null prior to snapshot
operation started.",
+ errMsg.value().isEmpty());
+ }
deltaBlock.countDown();
- fut0.get();
-
- assertTrue("Snapshot start time must be greater than zero for finished
snapshot.",
- startTime.value() > 0);
- assertTrue("Snapshot end time must be greater than zero for finished
snapshot.",
- endTime.value() > 0);
- assertEquals("Snapshot name must be set when snapshot operation
already finished.",
- SNAPSHOT_NAME, snpName.value());
- assertTrue("Concurrent snapshot operation must finished successfully.",
- errMsg.value().isEmpty());
- assertEquals("Only the first snapshot must be created and stored on
disk.",
- Collections.singletonList(SNAPSHOT_NAME), snpList.value());
+ fut.get();
+
+ for (Ignite g : G.allGrids()) {
+ MetricRegistry mreg =
((IgniteEx)g).context().metric().registry(SNAPSHOT_METRICS);
+
+ LongMetric startTime = mreg.findMetric("LastSnapshotStartTime");
+ LongMetric endTime = mreg.findMetric("LastSnapshotEndTime");
+
+ if (g.cluster().localNode().isClient()) {
+ assertNull("Snapshot start time must not be created on a
client node.", startTime);
+ assertNull("Snapshot end time must not be created on a client
node.", endTime);
+
+ continue;
+ }
+
+ assertTrue(waitForCondition(() -> endTime.value() != 0L &&
startTime.value() != 0 && endTime.value() > startTime.value(),
+ getTestTimeout()));
+ }
}
/** @throws Exception If fails. */
diff --git
a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/incremental/IncrementalSnapshotWarnAtomicCachesTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/incremental/IncrementalSnapshotWarnAtomicCachesTest.java
index 1806cee36a7..68911b39dbc 100644
---
a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/incremental/IncrementalSnapshotWarnAtomicCachesTest.java
+++
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/incremental/IncrementalSnapshotWarnAtomicCachesTest.java
@@ -19,6 +19,7 @@ package
org.apache.ignite.internal.processors.cache.persistence.snapshot.increme
import java.util.ArrayList;
import java.util.Collection;
+import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -112,8 +113,8 @@ public class IncrementalSnapshotWarnAtomicCachesTest
extends GridCommonAbstractT
/** */
public void
checkCachesSnapshotCreationAndRestore(CacheConfiguration<Integer, Integer>...
ccfgs) throws Exception {
- List<Integer> allWarnCaches = new ArrayList<>();
- Map<String, List<Integer>> warnCachesByGrps = new HashMap<>();
+ List<String> allWarnCaches = new ArrayList<>();
+ Map<String, List<String>> warnCachesByGrps = new HashMap<>();
for (CacheConfiguration<?, ?> ccfg: ccfgs) {
if (ccfg.getAtomicityMode() == CacheAtomicityMode.ATOMIC &&
ccfg.getBackups() > 0) {
@@ -122,37 +123,25 @@ public class IncrementalSnapshotWarnAtomicCachesTest
extends GridCommonAbstractT
warnCachesByGrps.compute(grpName, (grp, caches) -> {
caches = caches == null ? new ArrayList<>() : caches;
- int cacheNum =
Integer.parseInt(ccfg.getName().replace("cache", ""));
+ caches.add(ccfg.getName());
- caches.add(cacheNum);
-
- allWarnCaches.add(cacheNum);
+ allWarnCaches.add(ccfg.getName());
return caches;
});
}
}
- checkWarnMessageOnCreateSnapshot(cachesPattern(allWarnCaches), ccfgs);
- checkWarnMessageOnRestoreSnapshot(cachesPattern(allWarnCaches), null);
+ checkWarnMessageOnCreateSnapshot(allWarnCaches, ccfgs);
+ checkWarnMessageOnRestoreSnapshot(allWarnCaches, null);
for (String grp: warnCachesByGrps.keySet())
-
checkWarnMessageOnRestoreSnapshot(cachesPattern(warnCachesByGrps.get(grp)),
F.asList(grp));
- }
-
- /** Transforms cache numbers to cache pattern. For example, [0, 1] ->
cache[0,1], cache[0,1]. */
- private @Nullable String cachesPattern(List<Integer> cacheNums) {
- if (cacheNums.isEmpty())
- return null;
-
- return cacheNums.stream()
- .map(c -> "cache" + cacheNums)
- .collect(Collectors.joining(", "));
+ checkWarnMessageOnRestoreSnapshot(warnCachesByGrps.get(grp),
F.asList(grp));
}
/** */
private void checkWarnMessageOnCreateSnapshot(
- @Nullable String warnAtomicCaches,
+ Collection<String> warnAtomicCaches,
CacheConfiguration<Integer, Integer>... ccfgs
) throws Exception {
this.ccfgs = ccfgs;
@@ -172,25 +161,25 @@ public class IncrementalSnapshotWarnAtomicCachesTest
extends GridCommonAbstractT
g.snapshot().createSnapshot(SNP).get(getTestTimeout());
- assertTrue(warnAtomicCaches, lsnr.check());
+ assertTrue(lsnr.check());
for (CacheConfiguration<Integer, Integer> c: ccfgs) {
for (int i = 1_000; i < 2_000; i++)
g.cache(c.getName()).put(i, i);
}
- lsnr = warnLogListener(warnAtomicCaches, warnAtomicCaches == null ? 0
: 1);
+ lsnr = warnLogListener(warnAtomicCaches, warnAtomicCaches.isEmpty() ?
0 : 3);
lsnLogger.registerListener(lsnr);
g.snapshot().createIncrementalSnapshot(SNP).get(getTestTimeout());
- assertTrue(warnAtomicCaches, lsnr.check());
+ assertTrue(lsnr.check(getTestTimeout()));
}
/** */
private void checkWarnMessageOnRestoreSnapshot(
- @Nullable String warnAtomicCaches,
+ Collection<String> warnAtomicCaches,
@Nullable Collection<String> restoreCacheGrps
) throws Exception {
stopAllGrids();
@@ -211,26 +200,36 @@ public class IncrementalSnapshotWarnAtomicCachesTest
extends GridCommonAbstractT
g.snapshot().restoreSnapshot(SNP,
restoreCacheGrps).get(getTestTimeout());
- assertTrue(warnAtomicCaches + " " + restoreCacheGrps, lsnr.check());
+ assertTrue(lsnr.check());
g.destroyCaches(g.cacheNames());
awaitPartitionMapExchange();
- lsnr = warnLogListener(warnAtomicCaches, warnAtomicCaches == null ? 0
: 1);
+ lsnr = warnLogListener(warnAtomicCaches, warnAtomicCaches.isEmpty() ?
0 : 1);
lsnLogger.registerListener(lsnr);
g.snapshot().restoreSnapshot(SNP, restoreCacheGrps,
1).get(getTestTimeout());
- assertTrue(warnAtomicCaches + " " + restoreCacheGrps, lsnr.check());
+ assertTrue(lsnr.check());
}
/** */
- private LogListener warnLogListener(@Nullable String atomicCaches, int
times) {
+ private LogListener warnLogListener(Collection<String> atomicCaches, int
times) {
+ String cachesStr = null;
+
+ if (atomicCaches.size() == 1)
+ cachesStr = F.first(atomicCaches);
+ else if (atomicCaches.size() > 1) {
+ cachesStr = "((" + String.join(", ", atomicCaches) + ')';
+
+ cachesStr += "|(" +
atomicCaches.stream().sorted(Comparator.reverseOrder()).collect(Collectors.joining(",
")) + "))";
+ }
+
Pattern p = Pattern.compile(
"Incremental snapshot \\[snpName=" + SNP + ", incIdx=1] contains
ATOMIC caches with backups:"
- + (atomicCaches == null ? "" : " \\[" + atomicCaches) + ']');
+ + (cachesStr == null ? "" : " \\[" + cachesStr) + "]");
return LogListener.matches(p).times(times).build();
}
