This is an automated email from the ASF dual-hosted git repository. cschneider pushed a commit to branch SLING-12690 in repository https://gitbox.apache.org/repos/asf/sling-org-apache-sling-distribution-journal.git
commit 842f5feab4e1f53cd04cefd9b0dedfc37aa8c57d Author: Christian Schneider <[email protected]> AuthorDate: Mon Mar 10 15:49:00 2025 +0100 SLING-12690 - Refactor current import observability --- .../journal/bookkeeper/BookKeeper.java | 29 ++-------- .../journal/bookkeeper/CurrentImportInfo.java | 61 ++++++++++++++++++++++ .../journal/bookkeeper/SubscriberMetrics.java | 22 ++++---- .../journal/bookkeeper/BookKeeperTest.java | 12 ++--- 4 files changed, 82 insertions(+), 42 deletions(-) diff --git a/src/main/java/org/apache/sling/distribution/journal/bookkeeper/BookKeeper.java b/src/main/java/org/apache/sling/distribution/journal/bookkeeper/BookKeeper.java index cb63164..346e5e2 100644 --- a/src/main/java/org/apache/sling/distribution/journal/bookkeeper/BookKeeper.java +++ b/src/main/java/org/apache/sling/distribution/journal/bookkeeper/BookKeeper.java @@ -27,13 +27,10 @@ import static org.apache.sling.distribution.event.DistributionEventProperties.*; import java.io.IOException; import java.io.PrintWriter; import java.io.StringWriter; -import java.time.Duration; import java.util.Date; import java.util.HashMap; import java.util.Map; import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicLong; -import java.util.concurrent.atomic.AtomicReference; import java.util.function.Consumer; import org.apache.sling.api.resource.LoginException; @@ -87,7 +84,7 @@ public class BookKeeper { private static final String SUBSERVICE_BOOKKEEPER = "bookkeeper"; private static final int RETRY_SEND_DELAY = 1000; public static final int NUM_ERRORS_BLOCKING = 4; - public static final Duration IMPORT_TIME_WARN_LEVEL = Duration.ofMinutes(5); + private final Logger log = LoggerFactory.getLogger(this.getClass()); private final ResourceResolverFactory resolverFactory; @@ -105,9 +102,6 @@ public class BookKeeper { private final ImportPreProcessor importPreProcessor; private final ImportPostProcessor importPostProcessor; private final InvalidationProcessor invalidationProcessor; - private final AtomicLong currentImportStartTime; - private final AtomicReference<PackageMessage> currentImportPackage; - private final AtomicLong currentImportOffset; private int skippedCounter = 0; public BookKeeper(ResourceResolverFactory resolverFactory, SubscriberMetrics subscriberMetrics, @@ -130,10 +124,6 @@ public class BookKeeper { this.importPreProcessor = importPreProcessor; this.importPostProcessor = importPostProcessor; this.invalidationProcessor = invalidationProcessor; - this.subscriberMetrics.currentImportDuration(this::getCurrentImportDuration); - this.currentImportStartTime = new AtomicLong(); - this.currentImportPackage = new AtomicReference<>(); - this.currentImportOffset = new AtomicLong(); log.info("Started bookkeeper {}.", config); } @@ -159,9 +149,7 @@ public class BookKeeper { ResourceResolver importerResolver = getServiceResolver(SUBSERVICE_IMPORTER)) { // Execute the pre-processor preProcess(pkgMsg); - this.currentImportStartTime.set(importStartTime); - this.currentImportPackage.set(pkgMsg); - this.currentImportOffset.set(offset); + subscriberMetrics.setCurrentImport(new CurrentImportInfo(pkgMsg, offset, importStartTime)); packageHandler.apply(importerResolver, pkgMsg); if (config.isEditable()) { storeStatus(importerResolver, new PackageStatus(Status.IMPORTED, offset, pkgMsg.getPubAgentName())); @@ -185,7 +173,7 @@ public class BookKeeper { } catch (DistributionException | LoginException | IOException | RuntimeException | ImportPreProcessException |ImportPostProcessException e) { failure(pkgMsg, offset, e); } finally { - this.currentImportStartTime.set(0L); + subscriberMetrics.clearCurrentImport(); } } @@ -513,15 +501,4 @@ public class BookKeeper { } } - private Long getCurrentImportDuration() { - long importStartTime = this.currentImportStartTime.get(); - if (importStartTime == 0L) { - return 0L; // No import running - } - long currentImportDurationMs = System.currentTimeMillis() - importStartTime; - if (currentImportDurationMs > IMPORT_TIME_WARN_LEVEL.toMillis()) { - log.warn("Import of package={}, offset={} takes currentImportTimeSeconds={} which is longer than warnLevelSeconds={}", currentImportPackage.get(), currentImportOffset.get(), currentImportDurationMs / 1000, IMPORT_TIME_WARN_LEVEL.toSeconds()); - } - return currentImportDurationMs; - } } diff --git a/src/main/java/org/apache/sling/distribution/journal/bookkeeper/CurrentImportInfo.java b/src/main/java/org/apache/sling/distribution/journal/bookkeeper/CurrentImportInfo.java new file mode 100644 index 0000000..f864fb2 --- /dev/null +++ b/src/main/java/org/apache/sling/distribution/journal/bookkeeper/CurrentImportInfo.java @@ -0,0 +1,61 @@ +/* + * 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.sling.distribution.journal.bookkeeper; + +import java.time.Duration; + +import org.apache.sling.distribution.journal.messages.PackageMessage; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class CurrentImportInfo { + private final Logger log = LoggerFactory.getLogger(BookKeeper.class); + public static final Duration IMPORT_TIME_WARN_LEVEL = Duration.ofMinutes(5); + + private final PackageMessage pkgMsg; + private final long offset; + private final long startTime; + + public CurrentImportInfo(PackageMessage pkgMsg, long offset, long importStartTime) { + this.pkgMsg = pkgMsg; + this.offset = offset; + this.startTime = importStartTime; + } + + public PackageMessage getPkgMsg() { + return pkgMsg; + } + + public long getOffset() { + return offset; + } + + public long getImportStartTime() { + return startTime; + } + + Long getCurrentImportDuration() { + long currentImportDurationMs = System.currentTimeMillis() - startTime; + if (currentImportDurationMs > IMPORT_TIME_WARN_LEVEL.toMillis()) { + log.warn("Import of package={}, offset={} takes currentImportTimeSeconds={} which is longer than warnLevelSeconds={}", + pkgMsg, offset, currentImportDurationMs / 1000, IMPORT_TIME_WARN_LEVEL.toSeconds()); + } + return currentImportDurationMs; + } +} diff --git a/src/main/java/org/apache/sling/distribution/journal/bookkeeper/SubscriberMetrics.java b/src/main/java/org/apache/sling/distribution/journal/bookkeeper/SubscriberMetrics.java index 1a07ab8..8011cbb 100644 --- a/src/main/java/org/apache/sling/distribution/journal/bookkeeper/SubscriberMetrics.java +++ b/src/main/java/org/apache/sling/distribution/journal/bookkeeper/SubscriberMetrics.java @@ -22,6 +22,7 @@ import static org.apache.sling.distribution.journal.metrics.TaggedMetrics.getMet import java.util.Arrays; import java.util.List; +import java.util.concurrent.atomic.AtomicReference; import java.util.function.Supplier; import org.apache.sling.commons.metrics.Counter; @@ -97,7 +98,7 @@ public class SubscriberMetrics { private final Tag tagEditable; private final List<Tag> tags; - private Supplier<Long> currentImportDurationCallback; + private final AtomicReference<CurrentImportInfo> currentImportInfo = new AtomicReference<CurrentImportInfo>(); public SubscriberMetrics(MetricsService metricsService, String subAgentName, String pubAgentName, boolean editable) { this.metricsService = metricsService; @@ -108,6 +109,7 @@ public class SubscriberMetrics { tagSubName, tagPubName, tagEditable); + metricsService.gauge(getMetricName(CURRENT_IMPORT_DURATION, tags), this::getCurrentImportDuration); } /** @@ -250,17 +252,17 @@ public class SubscriberMetrics { metricsService.gauge(getMetricName(CURRENT_RETRIES, tags), retriesCallback); } - public void currentImportDuration(Supplier<Long> importDurationCallback) { - currentImportDurationCallback = importDurationCallback; - metricsService.gauge(getMetricName(CURRENT_IMPORT_DURATION, tags), importDurationCallback); + public void setCurrentImport(CurrentImportInfo currentImport) { + this.currentImportInfo.set(currentImport); } - /** - * For testing - * @return callback - */ - public Supplier<Long> getCurrentImportDurationCallback() { - return currentImportDurationCallback; + public void clearCurrentImport() { + this.currentImportInfo.set(null); } + + public long getCurrentImportDuration() { + CurrentImportInfo importInfo = currentImportInfo.get(); + return importInfo == null ? 0L : importInfo.getCurrentImportDuration(); + } } \ No newline at end of file diff --git a/src/test/java/org/apache/sling/distribution/journal/bookkeeper/BookKeeperTest.java b/src/test/java/org/apache/sling/distribution/journal/bookkeeper/BookKeeperTest.java index 19c1f9a..eb2998b 100644 --- a/src/test/java/org/apache/sling/distribution/journal/bookkeeper/BookKeeperTest.java +++ b/src/test/java/org/apache/sling/distribution/journal/bookkeeper/BookKeeperTest.java @@ -152,12 +152,12 @@ public class BookKeeperTest { @Test public void testPackageImportFailCurrentDuration() throws DistributionException, PersistenceException { - assertThat(subscriberMetrics.getCurrentImportDurationCallback().get(), equalTo(0L)); + assertThat(subscriberMetrics.getCurrentImportDuration(), equalTo(0L)); doAnswer(new Answer<Void>() { @Override public Void answer(InvocationOnMock invocation) throws Throwable { - Long duration = subscriberMetrics.getCurrentImportDurationCallback().get(); + long duration = subscriberMetrics.getCurrentImportDuration(); if (duration < Duration.ofMinutes(6).toMillis()) { throw new IllegalStateException("Should get valid duration"); } @@ -169,17 +169,17 @@ public class BookKeeperTest { long simulatedStartTime = currentTimeMillis() - Duration.ofMinutes(6).toMillis(); bookKeeper.importPackage(buildPackageMessage(PackageMessage.ReqType.ADD), 10, simulatedStartTime, simulatedStartTime); - assertThat(subscriberMetrics.getCurrentImportDurationCallback().get(), equalTo(0L)); + assertThat(subscriberMetrics.getCurrentImportDuration(), equalTo(0L)); } @Test public void testPackageImportCurrentDuration() throws DistributionException, PersistenceException { - assertThat(subscriberMetrics.getCurrentImportDurationCallback().get(), equalTo(0L)); + assertThat(subscriberMetrics.getCurrentImportDuration(), equalTo(0L)); doAnswer(new Answer<Void>() { @Override public Void answer(InvocationOnMock invocation) throws Throwable { - Long duration = subscriberMetrics.getCurrentImportDurationCallback().get(); + Long duration = subscriberMetrics.getCurrentImportDuration(); if (duration < Duration.ofMinutes(1).toMillis()) { throw new IllegalStateException("Should get valid duration"); } @@ -191,7 +191,7 @@ public class BookKeeperTest { long simulatedStartTime = currentTimeMillis() - Duration.ofMinutes(1).toMillis(); bookKeeper.importPackage(buildPackageMessage(PackageMessage.ReqType.ADD), 10, currentTimeMillis(), simulatedStartTime); - assertThat(subscriberMetrics.getCurrentImportDurationCallback().get(), equalTo(0L)); + assertThat(subscriberMetrics.getCurrentImportDuration(), equalTo(0L)); } @Test
