This is an automated email from the ASF dual-hosted git repository.
Duansg pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hertzbeat.git
The following commit(s) were added to refs/heads/master by this push:
new e03e0e02e1 Generate metric scheduling intervals lazily (#4290)
e03e0e02e1 is described below
commit e03e0e02e19c37e5e179457eeb8de26d65e2d3c5
Author: Logic <[email protected]>
AuthorDate: Mon Aug 3 13:54:28 2026 +0800
Generate metric scheduling intervals lazily (#4290)
Co-authored-by: Duansg <[email protected]>
---
.../apache/hertzbeat/common/entity/job/Job.java | 76 +++++++++++-----
.../hertzbeat/common/entity/job/JobTest.java | 100 +++++++++++++++++++++
2 files changed, 155 insertions(+), 21 deletions(-)
diff --git
a/hertzbeat-common-core/src/main/java/org/apache/hertzbeat/common/entity/job/Job.java
b/hertzbeat-common-core/src/main/java/org/apache/hertzbeat/common/entity/job/Job.java
index e7fb6e52c1..9ac93ca227 100644
---
a/hertzbeat-common-core/src/main/java/org/apache/hertzbeat/common/entity/job/Job.java
+++
b/hertzbeat-common-core/src/main/java/org/apache/hertzbeat/common/entity/job/Job.java
@@ -25,6 +25,7 @@ import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedDeque;
@@ -110,9 +111,21 @@ public class Job {
*/
private long defaultInterval = 600L;
/**
- * Refresh time list for one cycle of the job.
+ * Explicit refresh delays used for one-time overrides and fallback
schedules.
*/
private ConcurrentLinkedDeque<Long> intervals;
+
+ /**
+ * Collector-only periods used to generate the next refresh delay lazily.
+ */
+ @JsonIgnore
+ private transient long[] metricsSchedulePeriods;
+
+ /**
+ * Collector-only time remaining until each metric period is due.
+ */
+ @JsonIgnore
+ private transient long[] metricsScheduleRemaining;
/**
* Whether it is a recurring periodic task true is yes, false is no.
*/
@@ -328,7 +341,7 @@ public class Job {
if (array != null && !array.isEmpty()) {
long result = array.get(0);
for (int i = 1; i < array.size(); i++) {
- result = result / gcd(result, array.get(i)) * array.get(i);
+ result = Math.multiplyExact(result / gcd(result,
array.get(i)), array.get(i));
}
return result;
}
@@ -340,29 +353,50 @@ public class Job {
* Generate a list of refresh intervals for metric collection.
*/
public synchronized void generateMetricsIntervals(List<Long>
metricsIntervals) {
- // 1. To find the least common multiple (LCM) of all metric refresh
intervals
- long lcm = lcm(metricsIntervals);
- List<Long> refreshTimes = new LinkedList<>();
- // 2. Calculate all possible refresh intervals in one round
- for (long interval : metricsIntervals) {
- for (long t = interval; t <= lcm; t += interval) {
- if (!refreshTimes.contains(t)) {
- refreshTimes.add(t);
- }
- }
- }
- // 3. Sort from smallest to largest
- Collections.sort(refreshTimes);
- // 4. Calculate the refresh interval list for Job's cycle
- LinkedList<Long> intervals = new LinkedList<>();
- intervals.add(refreshTimes.get(0));
- for (int i = 1; i < refreshTimes.size(); i++) {
- intervals.add(refreshTimes.get(i) - refreshTimes.get(i - 1));
+ long[] periods = metricsIntervals == null
+ ? new long[0]
+ : metricsIntervals.stream()
+ .filter(Objects::nonNull)
+ .mapToLong(Long::longValue)
+ .filter(interval -> interval > 0)
+ .distinct()
+ .sorted()
+ .toArray();
+ if (periods.length == 0) {
+ metricsSchedulePeriods = null;
+ metricsScheduleRemaining = null;
+ intervals = new ConcurrentLinkedDeque<>(List.of(Math.max(1L,
defaultInterval)));
+ return;
}
- setIntervals(new ConcurrentLinkedDeque<>(intervals));
+ metricsSchedulePeriods = periods;
+ metricsScheduleRemaining = periods.clone();
+ intervals = null;
+ }
+
+ public synchronized void setIntervals(ConcurrentLinkedDeque<Long>
intervals) {
+ this.intervals = intervals;
+ metricsSchedulePeriods = null;
+ metricsScheduleRemaining = null;
}
public synchronized long getInterval() {
+ if (metricsScheduleRemaining != null && metricsSchedulePeriods != null
+ && metricsScheduleRemaining.length ==
metricsSchedulePeriods.length
+ && metricsScheduleRemaining.length > 0) {
+ long nextInterval = Long.MAX_VALUE;
+ for (long remaining : metricsScheduleRemaining) {
+ nextInterval = Math.min(nextInterval, remaining);
+ }
+ if (nextInterval > 0) {
+ for (int i = 0; i < metricsScheduleRemaining.length; i++) {
+ metricsScheduleRemaining[i] -= nextInterval;
+ if (metricsScheduleRemaining[i] == 0) {
+ metricsScheduleRemaining[i] =
metricsSchedulePeriods[i];
+ }
+ }
+ return nextInterval;
+ }
+ }
if (this.intervals != null && !this.intervals.isEmpty()) {
Long interval = this.intervals.removeFirst();
if (interval != null) {
diff --git
a/hertzbeat-common-core/src/test/java/org/apache/hertzbeat/common/entity/job/JobTest.java
b/hertzbeat-common-core/src/test/java/org/apache/hertzbeat/common/entity/job/JobTest.java
new file mode 100644
index 0000000000..9a9e1a1b20
--- /dev/null
+++
b/hertzbeat-common-core/src/test/java/org/apache/hertzbeat/common/entity/job/JobTest.java
@@ -0,0 +1,100 @@
+/*
+ * 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.hertzbeat.common.entity.job;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.List;
+import java.util.concurrent.ConcurrentLinkedDeque;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Test case for {@link Job}.
+ */
+class JobTest {
+
+ @Test
+ void schedulesCoprimeIntervalsWithoutMaterializingTheirWholeCycle() {
+ Job job = new Job();
+
+ job.generateMetricsIntervals(List.of(1009L, 1013L));
+
+ assertTrue(job.getIntervals() == null || job.getIntervals().size() <=
2);
+ assertEquals(
+ List.of(1009L, 4L, 1005L, 8L, 1001L, 12L, 997L, 16L),
+ List.of(
+ job.getInterval(),
+ job.getInterval(),
+ job.getInterval(),
+ job.getInterval(),
+ job.getInterval(),
+ job.getInterval(),
+ job.getInterval(),
+ job.getInterval()));
+ }
+
+ @Test
+ void repeatsTheSameScheduleForOrdinaryIntervals() {
+ Job job = new Job();
+
+ job.generateMetricsIntervals(List.of(4L, 6L));
+
+ assertEquals(
+ List.of(4L, 2L, 2L, 4L, 4L, 2L, 2L, 4L),
+ List.of(
+ job.getInterval(),
+ job.getInterval(),
+ job.getInterval(),
+ job.getInterval(),
+ job.getInterval(),
+ job.getInterval(),
+ job.getInterval(),
+ job.getInterval()));
+ }
+
+ @Test
+ void explicitIntervalsReplaceGeneratedSchedule() {
+ Job job = new Job();
+ job.generateMetricsIntervals(List.of(4L, 6L));
+
+ job.setIntervals(new ConcurrentLinkedDeque<>(List.of(0L)));
+
+ assertEquals(0L, job.getInterval());
+ assertEquals(0L, job.getInterval());
+ }
+
+ @Test
+ void usesDefaultIntervalWhenNoValidMetricIntervalExists() {
+ Job job = new Job();
+ job.setDefaultInterval(15L);
+
+ job.generateMetricsIntervals(List.of(0L, -1L));
+
+ assertEquals(15L, job.getInterval());
+ assertEquals(15L, job.getInterval());
+ }
+
+ @Test
+ void rejectsLeastCommonMultipleOverflow() {
+ assertThrows(
+ ArithmeticException.class,
+ () -> Job.lcm(List.of(Long.MAX_VALUE, Long.MAX_VALUE - 1)));
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
