diff --git a/hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/prometheus/parser/OnlineParser.java b/hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/prometheus/parser/OnlineParser.java index de48f59656..d5f9fae07f 100644 --- a/hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/prometheus/parser/OnlineParser.java +++ b/hertzbeat-collector/hertzbeat-collector-basic/src/main/java/org/apache/hertzbeat/collector/collect/prometheus/parser/OnlineParser.java @@ -60,16 +60,37 @@ public class OnlineParser { } public static Map parseMetrics(InputStream inputStream) throws IOException { - Map metricFamilyMap = new ConcurrentHashMap<>(10); + return parseMetrics(inputStream, Integer.MAX_VALUE); + } + + /** + * Parses at most {@code maxSamples} samples from the supplied stream. + * + * @param inputStream The Prometheus text stream + * @param maxSamples The maximum number of samples to materialize + * @return The parsed metric families, or {@code null} when the text format is invalid + * @throws IOException When the stream cannot be read + * @throws SampleLimitExceededException When another sample follows the configured limit + */ + public static Map parseMetrics(InputStream inputStream, int maxSamples) throws IOException { + if (maxSamples < 0) { + throw new IllegalArgumentException("maxSamples must not be negative"); + } + final Map metricFamilyMap = new ConcurrentHashMap<>(10); + int sampleCount = 0; try { int i = getChar(inputStream); while (i != -1) { if (i == '#' || i == '\n') { skipToLineEnd(inputStream).maybeEol().maybeEof().noElse(); } else { - StringBuilder stringBuilder = new StringBuilder(); + if (sampleCount >= maxSamples) { + throw new SampleLimitExceededException(maxSamples); + } + final StringBuilder stringBuilder = new StringBuilder(); stringBuilder.append((char) i); parseMetric(inputStream, metricFamilyMap, stringBuilder); + sampleCount++; } i = getChar(inputStream); // To address the `\n\r` scenario, it is necessary to skip @@ -84,6 +105,16 @@ public class OnlineParser { return metricFamilyMap; } + /** + * Signals that parsing stopped before materializing a sample beyond the configured limit. + */ + public static final class SampleLimitExceededException extends IOException { + + public SampleLimitExceededException(int limit) { + super("prometheus payload exceeds the " + limit + " sample limit"); + } + } + /** * Parses Prometheus metrics from the given {@link InputStream}, but only for the specified metric name. *

diff --git a/hertzbeat-collector/hertzbeat-collector-basic/src/test/java/org/apache/hertzbeat/collector/collect/prometheus/parser/OnlineParserTest.java b/hertzbeat-collector/hertzbeat-collector-basic/src/test/java/org/apache/hertzbeat/collector/collect/prometheus/parser/OnlineParserTest.java index 4f515b8696..370ac9a3bd 100644 --- a/hertzbeat-collector/hertzbeat-collector-basic/src/test/java/org/apache/hertzbeat/collector/collect/prometheus/parser/OnlineParserTest.java +++ b/hertzbeat-collector/hertzbeat-collector-basic/src/test/java/org/apache/hertzbeat/collector/collect/prometheus/parser/OnlineParserTest.java @@ -30,6 +30,8 @@ import java.util.stream.Collectors; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; import static org.junit.jupiter.api.Assertions.fail; class OnlineParserTest { @@ -459,4 +461,27 @@ class OnlineParserTest { assertEquals("run_as", metricFamily.getMetricList().get(0).getLabels().get(3).getName()); assertEquals("NT AUTHORITY\nLocalService", metricFamily.getMetricList().get(0).getLabels().get(3).getValue()); } + + @Test + void testParseMetricsStopsBeforeSampleBeyondLimit() { + final String metrics = "metric_a 1\nmetric_b 2\nmetric_c 3\nmetric_d 4\n"; + final ByteArrayInputStream inputStream = + new ByteArrayInputStream(metrics.getBytes(StandardCharsets.UTF_8)); + + assertThrows(OnlineParser.SampleLimitExceededException.class, + () -> OnlineParser.parseMetrics(inputStream, 2)); + + assertTrue(inputStream.available() > 0, "samples after the limit should remain unread"); + } + + @Test + void testParseMetricsAllowsExactlyTheSampleLimit() throws Exception { + final String metrics = "metric_a 1\nmetric_b 2\n"; + final InputStream inputStream = new ByteArrayInputStream(metrics.getBytes(StandardCharsets.UTF_8)); + + final Map metricFamilyMap = OnlineParser.parseMetrics(inputStream, 2); + + assertNotNull(metricFamilyMap); + assertEquals(2, metricFamilyMap.size()); + } } diff --git a/hertzbeat-push/src/main/java/org/apache/hertzbeat/push/service/impl/PushGatewayServiceImpl.java b/hertzbeat-push/src/main/java/org/apache/hertzbeat/push/service/impl/PushGatewayServiceImpl.java index 1be1a9203f..e0b706bff9 100644 --- a/hertzbeat-push/src/main/java/org/apache/hertzbeat/push/service/impl/PushGatewayServiceImpl.java +++ b/hertzbeat-push/src/main/java/org/apache/hertzbeat/push/service/impl/PushGatewayServiceImpl.java @@ -19,13 +19,19 @@ package org.apache.hertzbeat.push.service.impl; +import java.io.IOException; import java.io.InputStream; import java.time.Instant; +import java.util.HashMap; import java.util.LinkedList; import java.util.List; import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; import java.util.concurrent.ConcurrentHashMap; -import java.util.stream.Collectors; +import java.util.concurrent.atomic.AtomicInteger; + +import jakarta.annotation.Nullable; import lombok.extern.slf4j.Slf4j; import org.apache.hertzbeat.collector.collect.prometheus.parser.MetricFamily; @@ -37,6 +43,7 @@ import org.apache.hertzbeat.common.queue.CommonDataQueue; import org.apache.hertzbeat.common.util.SnowFlakeIdGenerator; import org.apache.hertzbeat.push.dao.PushMonitorDao; import org.apache.hertzbeat.push.service.PushGatewayService; +import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; /** @@ -46,26 +53,81 @@ import org.springframework.stereotype.Service; @Slf4j @Service public class PushGatewayServiceImpl implements PushGatewayService { - + + private static final byte PUSH_MONITOR_TYPE = (byte) 1; + private final CommonDataQueue commonDataQueue; private final PushMonitorDao pushMonitorDao; - private final Map jobInstanceMap; - - public PushGatewayServiceImpl(CommonDataQueue commonDataQueue, PushMonitorDao pushMonitorDao) { + private final Map jobInstanceMap; + + /** + * Cap on push monitors created automatically from unknown job/instance pairs. + * + *

The route is unauthenticated by design, and every new pair used to persist a + * monitor row and add a `jobInstanceMap` entry that is never removed, so a caller + * iterating over made up names could grow the database and the heap without bound. + * Above the cap an unknown pair is refused while the pairs already known keep working, + * which is why eviction is not used here: evicting a live entry would make the next + * push for that pair create a second monitor for the same job and instance. + */ + private final int maxAutoCreatedMonitors; + + /** + * Cap on how many bytes a single push body may carry. + * + *

The parser materializes its result in memory, and the servlet container does not bound + * a non form request body, so an independent byte limit is still needed alongside the sample + * limit to keep long names and label values from exhausting the heap. + */ + private final long maxBodyBytes; + + /** + * Cap on how many samples a single push body may carry. The parser stops before allocating + * a sample beyond this limit, so a compact body cannot create an unbounded object graph. + */ + private final int maxSamples; + + /** + * One entry per pair whose monitor is being created, so that concurrent pushes naming the + * same unknown pair wait for one creation instead of each starting their own. Persistence + * stays outside any shared lock: a slow database would otherwise hold every request for an + * unknown pair, and each of those requests is already holding its parsed samples. + */ + private final Map> monitorCreationMap; + + /** + * Successful and in flight creations together, so that the cap is claimed before the + * database write rather than counted after it. + */ + private final AtomicInteger trackedMonitorCount; + + public PushGatewayServiceImpl(CommonDataQueue commonDataQueue, PushMonitorDao pushMonitorDao, + @Value("${hertzbeat.push.max-auto-created-monitors:10000}") int maxAutoCreatedMonitors, + @Value("${hertzbeat.push.max-body-bytes:5242880}") long maxBodyBytes, + @Value("${hertzbeat.push.max-samples:10000}") int maxSamples) { + if (maxAutoCreatedMonitors < 0 || maxBodyBytes < 0 || maxSamples < 0) { + throw new IllegalArgumentException("push gateway limits must not be negative"); + } this.commonDataQueue = commonDataQueue; this.pushMonitorDao = pushMonitorDao; + this.maxAutoCreatedMonitors = maxAutoCreatedMonitors; + this.maxBodyBytes = maxBodyBytes; + this.maxSamples = maxSamples; jobInstanceMap = new ConcurrentHashMap<>(); - pushMonitorDao.findMonitorsByType((byte) 1).forEach(monitor -> - jobInstanceMap.put(monitor.getApp() + "_" + monitor.getName(), monitor.getId())); + pushMonitorDao.findMonitorsByType(PUSH_MONITOR_TYPE).forEach(monitor -> + jobInstanceMap.put(new JobInstance(monitor.getApp(), monitor.getName()), monitor.getId())); + monitorCreationMap = new ConcurrentHashMap<>(); + trackedMonitorCount = new AtomicInteger(jobInstanceMap.size()); } @Override public boolean pushPrometheusMetrics(InputStream inputStream, String job, String instance) { try { - long curTime = Instant.now().toEpochMilli(); - Map metricFamilyMap = OnlineParser.parseMetrics(inputStream); + final long curTime = Instant.now().toEpochMilli(); + final Map metricFamilyMap = OnlineParser.parseMetrics( + new BoundedInputStream(inputStream, maxBodyBytes), maxSamples); if (metricFamilyMap == null) { log.error("parse prometheus metrics is null, job: {}, instance: {}", job, instance); return false; @@ -74,20 +136,11 @@ public class PushGatewayServiceImpl implements PushGatewayService { if (job != null && instance != null) { // auto create monitor when job and instance not null // job is app, instance is the name - id = jobInstanceMap.computeIfAbsent(job + "_" + instance, key -> { - log.info("auto create monitor by prometheus push, job: {}, instance: {}", job, instance); - long monitorId = SnowFlakeIdGenerator.generateId(); - Monitor monitor = Monitor.builder() - .id(monitorId) - .app(job) - .name(instance) - .instance(instance) - .type((byte) 1) - .status(CommonConstants.MONITOR_UP_CODE) - .build(); - this.pushMonitorDao.save(monitor); - return monitorId; - }); + final Long monitorId = resolveMonitorId(new JobInstance(job, instance)); + if (monitorId == null) { + return false; + } + id = monitorId; } for (Map.Entry entry : metricFamilyMap.entrySet()) { CollectRep.MetricsData.Builder builder = CollectRep.MetricsData.newBuilder(); @@ -110,9 +163,18 @@ public class PushGatewayServiceImpl implements PushGatewayService { builder.addField(CollectRep.Field.newBuilder().setName("value") .setType(CommonConstants.TYPE_NUMBER).setLabel(false).build()); } - Map labelMap = metric.getLabels() - .stream() - .collect(Collectors.toMap(MetricFamily.Label::getName, MetricFamily.Label::getValue)); + // A repeated label name is refused rather than resolved: the exposition + // format requires the names of a label set to be unique, and keeping one + // of the values would emit a schema carrying that name twice. Built by + // hand so the refusal is a rejection this method can answer with a + // warning, not the error trace a collector's exception would produce. + Map labelMap = new HashMap<>(metric.getLabels().size()); + for (MetricFamily.Label label : metric.getLabels()) { + if (labelMap.containsKey(label.getName())) { + throw new DuplicateLabelException(label.getName()); + } + labelMap.put(label.getName(), label.getValue()); + } CollectRep.ValueRow.Builder valueRowBuilder = CollectRep.ValueRow.newBuilder(); for (String field : metricsFields) { String fieldValue = labelMap.get(field); @@ -125,9 +187,210 @@ public class PushGatewayServiceImpl implements PushGatewayService { } } return true; + } catch (BodyTooLargeException e) { + // A rejection, not a failure: caught apart from the generic handler below so that a + // caller repeating oversized bodies costs one warning line each, not a stack trace + log.warn("reject prometheus push over the {} byte body limit, job: {}, instance: {}", + maxBodyBytes, job, instance); + return false; + } catch (OnlineParser.SampleLimitExceededException e) { + log.warn("reject prometheus push over the {} sample limit, job: {}, instance: {}", + maxSamples, job, instance); + return false; + } catch (DuplicateLabelException e) { + log.warn("reject prometheus push repeating a label name, job: {}, instance: {}: {}", + job, instance, e.getMessage()); + return false; } catch (Exception e) { log.error("push prometheus metrics error", e); return false; } } + + /** + * Returns the monitor id a job/instance pair resolves to, creating the monitor on first + * sight, or null once {@link #maxAutoCreatedMonitors} is reached. + * + *

Concurrent pushes naming the same unknown pair share one creation through a future, and + * the cap is claimed by an atomic count before the database write. Nothing here holds a lock + * across that write: unrelated pairs persist in parallel, and a slow database delays only the + * requests naming the pair being created, which matters because every waiting request is + * holding the samples it already parsed. + * + * @param pair Job and instance the push named + * @return The monitor id, or null when the cap leaves no room for a new one + */ + @Nullable + private Long resolveMonitorId(JobInstance pair) { + final Long known = jobInstanceMap.get(pair); + if (known != null) { + return known; + } + + final CompletableFuture proposedCreation = new CompletableFuture<>(); + final CompletableFuture ongoingCreation = monitorCreationMap.putIfAbsent(pair, proposedCreation); + if (ongoingCreation != null) { + try { + return ongoingCreation.join(); + } catch (CompletionException e) { + // The request owning the creation reports the failure once; joining its exception + // here would multiply a single database error by every request that waited + return null; + } + } + + try { + // Looked up again now that the creation is claimed: another request may have finished + // this pair between the lookup above and this claim, and going on to create it would + // leave two monitors for one pair and spend a second slot of the cap + final Long createdMeanwhile = jobInstanceMap.get(pair); + if (createdMeanwhile != null) { + proposedCreation.complete(createdMeanwhile); + return createdMeanwhile; + } + if (!reserveMonitorSlot()) { + proposedCreation.complete(null); + log.warn("reject prometheus push for unknown job: {}, instance: {}, " + + "already tracking {} push monitors, limit is {}", + pair.job(), pair.instance(), trackedMonitorCount.get(), maxAutoCreatedMonitors); + return null; + } + boolean created = false; + try { + final long monitorId = createMonitor(pair); + jobInstanceMap.put(pair, monitorId); + created = true; + proposedCreation.complete(monitorId); + return monitorId; + } finally { + if (!created) { + trackedMonitorCount.decrementAndGet(); + } + } + } catch (RuntimeException | Error e) { + proposedCreation.completeExceptionally(e); + throw e; + } finally { + monitorCreationMap.remove(pair, proposedCreation); + } + } + + /** + * Claims one slot of the cap, or reports that none is left. Claiming before the database + * write is what keeps concurrent creations from exceeding the cap together. + */ + private boolean reserveMonitorSlot() { + int tracked = trackedMonitorCount.get(); + while (tracked < maxAutoCreatedMonitors) { + if (trackedMonitorCount.compareAndSet(tracked, tracked + 1)) { + return true; + } + tracked = trackedMonitorCount.get(); + } + return false; + } + + /** + * Persists a push monitor after its slot of the cap has been claimed. + */ + private long createMonitor(JobInstance pair) { + final String job = pair.job(); + final String instance = pair.instance(); + log.info("auto create monitor by prometheus push, job: {}, instance: {}", job, instance); + final long monitorId = SnowFlakeIdGenerator.generateId(); + final Monitor monitor = Monitor.builder() + .id(monitorId) + .app(job) + .name(instance) + .instance(instance) + .type(PUSH_MONITOR_TYPE) + .status(CommonConstants.MONITOR_UP_CODE) + .build(); + this.pushMonitorDao.save(monitor); + return monitorId; + } + + /** + * Identifies the monitor a push belongs to. + * + *

The two names are kept apart instead of being joined into one string: a separator + * carries no meaning in either name, so `job + "_" + instance` maps ("a", "b_c") and + * ("a_b", "c") onto the same key. Colliding pairs would push their samples into whichever + * monitor was created first, and at startup they would collapse into a single map entry, + * making the cap count fewer monitors than the database actually holds. + */ + private record JobInstance(String job, String instance) { + } + + /** + * Raised when a sample repeats a label name, which the exposition format does not allow. + * Kept apart from the generic handler so a malformed body costs one warning line rather + * than an error trace on a route that takes its input from anyone. + */ + static final class DuplicateLabelException extends IOException { + + DuplicateLabelException(String name) { + super("sample repeats the label name " + name); + } + } + + /** + * Raised when a body goes past {@link #maxBodyBytes}. It is kept apart from the other read + * failures so the caller can answer a body that is merely too large without an error trace. + */ + static final class BodyTooLargeException extends IOException { + + BodyTooLargeException(long limit) { + super("push body exceeds the " + limit + " byte limit"); + } + } + + /** + * Fails the read once the body has delivered more than {@code limit} bytes, instead of + * letting the parser accumulate an unbounded body in memory. Reading stops at the + * failure, so the bytes beyond the limit are never buffered. + */ + static final class BoundedInputStream extends InputStream { + + private final InputStream delegate; + + private final long limit; + + private long bytesRead; + + BoundedInputStream(InputStream delegate, long limit) { + this.delegate = delegate; + this.limit = limit; + } + + @Override + public int read() throws IOException { + final int value = delegate.read(); + if (value != -1) { + recordBytesRead(1); + } + return value; + } + + @Override + public int read(byte[] buffer, int offset, int length) throws IOException { + final int bytesReadNow = delegate.read(buffer, offset, length); + if (bytesReadNow > 0) { + recordBytesRead(bytesReadNow); + } + return bytesReadNow; + } + + private void recordBytesRead(int increment) throws IOException { + bytesRead += increment; + if (bytesRead > limit) { + throw new BodyTooLargeException(limit); + } + } + + @Override + public void close() throws IOException { + delegate.close(); + } + } } diff --git a/hertzbeat-push/src/test/java/org/apache/hertzbeat/push/service/impl/PushGatewayServiceImplTest.java b/hertzbeat-push/src/test/java/org/apache/hertzbeat/push/service/impl/PushGatewayServiceImplTest.java new file mode 100644 index 0000000000..d1991824e4 --- /dev/null +++ b/hertzbeat-push/src/test/java/org/apache/hertzbeat/push/service/impl/PushGatewayServiceImplTest.java @@ -0,0 +1,367 @@ +/* + * 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.push.service.impl; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.lenient; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.lang.reflect.Field; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CyclicBarrier; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import org.apache.hertzbeat.common.entity.manager.Monitor; +import org.apache.hertzbeat.common.entity.message.CollectRep; +import org.apache.hertzbeat.common.queue.CommonDataQueue; +import org.apache.hertzbeat.push.dao.PushMonitorDao; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.ArgumentCaptor; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +/** + * Test case for {@link PushGatewayServiceImpl}. + * + *

`/api/push/prometheus/**` is unauthenticated by design, so the resource a single + * anonymous request may consume has to be bounded: the body it may carry, the samples it + * may enqueue, and the number of push monitors it may bring into existence. + */ +@ExtendWith(MockitoExtension.class) +class PushGatewayServiceImplTest { + + private static final String BODY = "sample_metric{label=\"a\"} 1\n"; + + @Mock + private CommonDataQueue commonDataQueue; + + @Mock + private PushMonitorDao pushMonitorDao; + + @BeforeEach + void setUp() { + // The stream test below builds no service, so this default must not be strict + lenient().when(pushMonitorDao.findMonitorsByType((byte) 1)).thenReturn(List.of()); + } + + private PushGatewayServiceImpl createService(int maxMonitors, long maxBodyBytes, int maxSamples) { + return new PushGatewayServiceImpl(commonDataQueue, pushMonitorDao, maxMonitors, maxBodyBytes, maxSamples); + } + + private static ByteArrayInputStream createBody(String content) { + return new ByteArrayInputStream(content.getBytes(StandardCharsets.UTF_8)); + } + + /** + * The exposition format requires the names of a label set to be unique, so a sample that + * repeats one is refused rather than resolved. It stays a rejection though: answering it + * with an error trace would let a malformed body fill the log on an anonymous route. + */ + @Test + void testSampleRepeatingTheLabelNameIsRejected() { + final PushGatewayServiceImpl service = createService(10, 1024, 100); + + assertFalse(service.pushPrometheusMetrics( + createBody("sample_metric{label=\"a\",label=\"b\"} 1\n"), "job1", "instance1")); + + verify(commonDataQueue, never()).sendMetricsData(any()); + } + + @Test + void testPushIsAcceptedWithinTheLimits() { + final PushGatewayServiceImpl service = createService(10, 1024, 100); + + assertTrue(service.pushPrometheusMetrics(createBody(BODY), "job1", "instance1")); + + verify(pushMonitorDao).save(any(Monitor.class)); + } + + @Test + void testBodyBeyondTheByteLimitIsRejected() { + final PushGatewayServiceImpl service = createService(10, 16, 100); + + assertFalse(service.pushPrometheusMetrics(createBody(BODY.repeat(100)), "job1", "instance1")); + + verify(pushMonitorDao, never()).save(any(Monitor.class)); + } + + @Test + void testBodyBeyondTheSampleLimitIsRejected() { + final PushGatewayServiceImpl service = createService(10, 1024 * 1024, 2); + final StringBuilder many = new StringBuilder(); + for (int index = 0; index < 10; index++) { + many.append("sample_metric{label=\"value").append(index).append("\"} 1\n"); + } + final ByteArrayInputStream inputStream = createBody(many.toString()); + + assertFalse(service.pushPrometheusMetrics(inputStream, "job1", "instance1")); + assertTrue(inputStream.available() > 0, "the parser should stop before consuming the remaining samples"); + + verify(pushMonitorDao, never()).save(any(Monitor.class)); + } + + @Test + void testBodyAtTheSampleLimitIsAccepted() { + final PushGatewayServiceImpl service = createService(10, 1024, 2); + final String twoSamples = "sample_metric{label=\"a\"} 1\n" + + "sample_metric{label=\"b\"} 2\n"; + + assertTrue(service.pushPrometheusMetrics(createBody(twoSamples), "job1", "instance1")); + } + + /** + * An unknown job/instance pair persists a monitor row and adds a map entry that is + * never removed, so without a cap an anonymous caller iterating over made up names + * grows the database and the heap without bound. + */ + @Test + void testAutoCreationStopsAtTheMonitorLimit() { + final PushGatewayServiceImpl service = createService(2, 1024, 100); + + assertTrue(service.pushPrometheusMetrics(createBody(BODY), "job1", "instance1")); + assertTrue(service.pushPrometheusMetrics(createBody(BODY), "job2", "instance2")); + assertFalse(service.pushPrometheusMetrics(createBody(BODY), "job3", "instance3")); + + verify(pushMonitorDao, times(2)).save(any(Monitor.class)); + } + + /** + * The cap must not turn into eviction: a pair already known has to keep resolving to + * the monitor it created, otherwise a later push would create a second monitor for the + * same job and instance. + */ + @Test + void testKnownPairsKeepWorkingAtTheLimit() { + final PushGatewayServiceImpl service = createService(1, 1024, 100); + + assertTrue(service.pushPrometheusMetrics(createBody(BODY), "job1", "instance1")); + assertFalse(service.pushPrometheusMetrics(createBody(BODY), "other", "instance")); + assertTrue(service.pushPrometheusMetrics(createBody(BODY), "job1", "instance1")); + + verify(pushMonitorDao, times(1)).save(any(Monitor.class)); + } + + /** + * The route is anonymous, so nothing stops a caller from sending its unknown pairs all at + * once. Testing the cap and claiming the entry in two steps lets every request that already + * passed the test create a monitor of its own, which is the cap being exceeded by as many + * requests as the container serves in parallel. + */ + @Test + void testConcurrentPushesForUnknownPairsStopAtTheMonitorLimit() throws Exception { + final int callers = 16; + final PushGatewayServiceImpl service = createService(1, 1024, 100); + final CyclicBarrier startTogether = new CyclicBarrier(callers); + final ExecutorService pool = Executors.newFixedThreadPool(callers); + final AtomicInteger accepted = new AtomicInteger(); + try { + final List> pushes = new ArrayList<>(); + for (int index = 0; index < callers; index++) { + final String instance = "instance" + index; + pushes.add(pool.submit(() -> { + startTogether.await(); + if (service.pushPrometheusMetrics(createBody(BODY), "job", instance)) { + accepted.incrementAndGet(); + } + return null; + })); + } + for (final Future push : pushes) { + push.get(30, TimeUnit.SECONDS); + } + } finally { + pool.shutdownNow(); + } + + assertEquals(1, accepted.get()); + verify(pushMonitorDao, times(1)).save(any(Monitor.class)); + } + + @Test + void testFailedSaveDoesNotConsumeTheMonitorLimit() { + when(pushMonitorDao.save(any(Monitor.class))) + .thenThrow(new IllegalStateException("database unavailable")) + .thenAnswer(invocation -> invocation.getArgument(0)); + final PushGatewayServiceImpl service = createService(1, 1024, 100); + + assertFalse(service.pushPrometheusMetrics(createBody(BODY), "job1", "instance1")); + assertTrue(service.pushPrometheusMetrics(createBody(BODY), "job2", "instance2")); + + verify(pushMonitorDao, times(2)).save(any(Monitor.class)); + } + + @Test + void testNegativeLimitsAreRejectedAtConstruction() { + assertThrows(IllegalArgumentException.class, () -> createService(-1, 1024, 100)); + assertThrows(IllegalArgumentException.class, () -> createService(1, -1, 100)); + assertThrows(IllegalArgumentException.class, () -> createService(1, 1024, -1)); + } + + @Test + void testConcurrentPushesForTheSamePairCreateOneMonitor() throws Exception { + final int callers = 8; + final CyclicBarrier startTogether = new CyclicBarrier(callers); + final PushGatewayServiceImpl service = createService(1, 1024, 100); + final ExecutorService pool = Executors.newFixedThreadPool(callers); + try { + final List> pushes = new ArrayList<>(); + for (int index = 0; index < callers; index++) { + pushes.add(pool.submit(() -> { + startTogether.await(); + return service.pushPrometheusMetrics(createBody(BODY), "job", "instance"); + })); + } + for (final Future push : pushes) { + assertTrue(push.get(30, TimeUnit.SECONDS)); + } + } finally { + pool.shutdownNow(); + } + + verify(pushMonitorDao).save(any(Monitor.class)); + } + + /** + * A request may read no entry for a pair and only then claim the creation, by which time + * another request may have created that pair and cleared its claim. Without a second lookup + * once the claim is won, this request goes on to create the same pair again, leaving two + * monitors for one pair and a slot of the cap spent for good. + * + *

The interleaving is forced rather than raced: the map hands back a miss, and while it + * does, a competing push runs its whole creation. + */ + @Test + @SuppressWarnings({"rawtypes", "unchecked"}) + void testStaleMissDoesNotCreateTwoMonitorsForOnePair() throws Exception { + final PushGatewayServiceImpl service = createService(2, 1024, 100); + final Field trackedPairs = PushGatewayServiceImpl.class.getDeclaredField("jobInstanceMap"); + trackedPairs.setAccessible(true); + final AtomicBoolean competed = new AtomicBoolean(); + final Map probing = new ConcurrentHashMap() { + @Override + public Object get(Object key) { + final Object value = super.get(key); + if (value == null && competed.compareAndSet(false, true)) { + service.pushPrometheusMetrics(createBody(BODY), "job1", "instance1"); + } + return value; + } + }; + trackedPairs.set(service, probing); + + assertTrue(service.pushPrometheusMetrics(createBody(BODY), "job1", "instance1")); + + verify(pushMonitorDao, times(1)).save(any(Monitor.class)); + assertEquals(1, probing.size()); + // The slot the duplicate would have taken is still there for another pair + assertTrue(service.pushPrometheusMetrics(createBody(BODY), "job2", "instance2")); + } + + @Test + void testMonitorsLoadedAtStartupCountTowardsTheLimit() { + lenient().when(pushMonitorDao.findMonitorsByType((byte) 1)).thenReturn(List.of( + Monitor.builder().id(1L).app("job1").name("instance1").build())); + final PushGatewayServiceImpl service = createService(1, 1024, 100); + + assertFalse(service.pushPrometheusMetrics(createBody(BODY), "job2", "instance2")); + assertTrue(service.pushPrometheusMetrics(createBody(BODY), "job1", "instance1")); + + verify(pushMonitorDao, never()).save(any(Monitor.class)); + } + + /** + * A separator carries no meaning inside a job or an instance name, so the two names must not + * be joined into a single key: ("job", "a_b") and ("job_a", "b") are different monitors, and + * the second pair must not push its samples into the monitor the first one created. + */ + @Test + void testPairsSharingTheSeparatorAreDistinctMonitors() { + final PushGatewayServiceImpl service = createService(10, 1024, 100); + + assertTrue(service.pushPrometheusMetrics(createBody(BODY), "job", "a_b")); + assertTrue(service.pushPrometheusMetrics(createBody(BODY), "job_a", "b")); + + verify(pushMonitorDao, times(2)).save(any(Monitor.class)); + final ArgumentCaptor pushed = + ArgumentCaptor.forClass(CollectRep.MetricsData.class); + verify(commonDataQueue, times(2)).sendMetricsData(pushed.capture()); + assertNotEquals(pushed.getAllValues().get(0).getId(), pushed.getAllValues().get(1).getId()); + } + + /** + * Colliding pairs must not collapse into one entry while the monitors are loaded either, + * which would let the cap count fewer monitors than the database actually holds. + */ + @Test + void testPairsSharingTheSeparatorCountSeparatelyAtStartup() { + lenient().when(pushMonitorDao.findMonitorsByType((byte) 1)).thenReturn(List.of( + Monitor.builder().id(1L).app("job").name("a_b").build(), + Monitor.builder().id(2L).app("job_a").name("b").build())); + final PushGatewayServiceImpl service = createService(2, 1024, 100); + + assertFalse(service.pushPrometheusMetrics(createBody(BODY), "job3", "instance3")); + + verify(pushMonitorDao, never()).save(any(Monitor.class)); + } + + @Test + void testBoundedStreamStopsAtTheLimit() throws Exception { + final PushGatewayServiceImpl.BoundedInputStream stream = + new PushGatewayServiceImpl.BoundedInputStream(createBody("abcdef"), 3); + + assertEquals('a', stream.read()); + assertEquals('b', stream.read()); + assertEquals('c', stream.read()); + assertThrows(PushGatewayServiceImpl.BodyTooLargeException.class, stream::read); + } + + /** + * A body over the limit is a rejection rather than a failure, so it must be distinguishable + * from a read that genuinely broke: the route is anonymous, and answering every oversized + * body with an error trace lets a caller fill the log at will. + */ + @Test + void testBodyOverTheByteLimitIsRejectedNotFailed() { + final PushGatewayServiceImpl.BoundedInputStream stream = + new PushGatewayServiceImpl.BoundedInputStream(createBody(BODY.repeat(100)), 16); + + final IOException raised = assertThrows(IOException.class, stream::readAllBytes); + + assertInstanceOf(PushGatewayServiceImpl.BodyTooLargeException.class, raised); + } +} diff --git a/hertzbeat-startup/src/main/resources/application.yml b/hertzbeat-startup/src/main/resources/application.yml index 90c0ea4bb8..fa6cda6e10 100644 --- a/hertzbeat-startup/src/main/resources/application.yml +++ b/hertzbeat-startup/src/main/resources/application.yml @@ -373,3 +373,8 @@ hertzbeat: concurrency-limit: 256 reject-when-limit-reached: true task-termination-timeout: 5000 + # Bounds on what a single request to the push gateway may consume. + push: + max-auto-created-monitors: ${HERTZBEAT_PUSH_MAX_AUTO_CREATED_MONITORS:10000} + max-body-bytes: ${HERTZBEAT_PUSH_MAX_BODY_BYTES:5242880} + max-samples: ${HERTZBEAT_PUSH_MAX_SAMPLES:10000} diff --git a/script/application.yml b/script/application.yml index 90c0ea4bb8..fa6cda6e10 100644 --- a/script/application.yml +++ b/script/application.yml @@ -373,3 +373,8 @@ hertzbeat: concurrency-limit: 256 reject-when-limit-reached: true task-termination-timeout: 5000 + # Bounds on what a single request to the push gateway may consume. + push: + max-auto-created-monitors: ${HERTZBEAT_PUSH_MAX_AUTO_CREATED_MONITORS:10000} + max-body-bytes: ${HERTZBEAT_PUSH_MAX_BODY_BYTES:5242880} + max-samples: ${HERTZBEAT_PUSH_MAX_SAMPLES:10000} diff --git a/script/docker-compose/hertzbeat-mysql-iotdb/conf/application.yml b/script/docker-compose/hertzbeat-mysql-iotdb/conf/application.yml index 33b0537468..7695271d03 100644 --- a/script/docker-compose/hertzbeat-mysql-iotdb/conf/application.yml +++ b/script/docker-compose/hertzbeat-mysql-iotdb/conf/application.yml @@ -109,7 +109,7 @@ spring: hibernate: format_sql: true dialect: org.hibernate.dialect.MySQLDialect - + flyway: enabled: true clean-disabled: true @@ -117,7 +117,7 @@ spring: baseline-version: 1 locations: - classpath:db/migration/mysql - + # Not Require, Please config if you need email notify mail: # Attention: this is mail server address. @@ -138,7 +138,7 @@ common: queue: # memory or kafka type: memory - + warehouse: store: # store history metrics data, enable only one below @@ -222,7 +222,7 @@ alerter: region: AWS_REGION_FOR_END_USER_MESSAGING twilio: account-sid: YOUR_ACCOUNT_SID - auth-token: YOUR_AUTH_TOKEN + auth-token: YOUR_AUTH_TOKEN twilio-phone-number: YOUR_TWILIO_PHONE_NUMBER scheduler: server: @@ -273,3 +273,8 @@ hertzbeat: concurrency-limit: 256 reject-when-limit-reached: true task-termination-timeout: 5000 + # Bounds on what a single request to the push gateway may consume + push: + max-auto-created-monitors: ${HERTZBEAT_PUSH_MAX_AUTO_CREATED_MONITORS:10000} + max-body-bytes: ${HERTZBEAT_PUSH_MAX_BODY_BYTES:5242880} + max-samples: ${HERTZBEAT_PUSH_MAX_SAMPLES:10000} diff --git a/script/docker-compose/hertzbeat-mysql-tdengine/conf/application.yml b/script/docker-compose/hertzbeat-mysql-tdengine/conf/application.yml index 8d795ed643..90220d0669 100644 --- a/script/docker-compose/hertzbeat-mysql-tdengine/conf/application.yml +++ b/script/docker-compose/hertzbeat-mysql-tdengine/conf/application.yml @@ -109,7 +109,7 @@ spring: hibernate: format_sql: true dialect: org.hibernate.dialect.MySQLDialect - + flyway: enabled: true clean-disabled: true @@ -117,7 +117,7 @@ spring: baseline-version: 1 locations: - classpath:db/migration/mysql - + # Not Require, Please config if you need email notify mail: # Attention: this is mail server address. @@ -138,7 +138,7 @@ common: queue: # memory or kafka type: memory - + warehouse: store: # store history metrics data, enable only one below @@ -219,7 +219,7 @@ alerter: region: AWS_REGION_FOR_END_USER_MESSAGING twilio: account-sid: YOUR_ACCOUNT_SID - auth-token: YOUR_AUTH_TOKEN + auth-token: YOUR_AUTH_TOKEN twilio-phone-number: YOUR_TWILIO_PHONE_NUMBER scheduler: server: @@ -270,3 +270,8 @@ hertzbeat: concurrency-limit: 256 reject-when-limit-reached: true task-termination-timeout: 5000 + # Bounds on what a single request to the push gateway may consume. + push: + max-auto-created-monitors: ${HERTZBEAT_PUSH_MAX_AUTO_CREATED_MONITORS:10000} + max-body-bytes: ${HERTZBEAT_PUSH_MAX_BODY_BYTES:5242880} + max-samples: ${HERTZBEAT_PUSH_MAX_SAMPLES:10000} diff --git a/script/docker-compose/hertzbeat-mysql-victoria-metrics/conf/application.yml b/script/docker-compose/hertzbeat-mysql-victoria-metrics/conf/application.yml index f92abb9f98..0074a9dd3f 100644 --- a/script/docker-compose/hertzbeat-mysql-victoria-metrics/conf/application.yml +++ b/script/docker-compose/hertzbeat-mysql-victoria-metrics/conf/application.yml @@ -109,7 +109,7 @@ spring: hibernate: format_sql: true dialect: org.hibernate.dialect.MySQLDialect - + flyway: enabled: true clean-disabled: true @@ -117,7 +117,7 @@ spring: baseline-version: 1 locations: - classpath:db/migration/mysql - + # Not Require, Please config if you need email notify mail: # Attention: this is mail server address. @@ -142,7 +142,7 @@ warehouse: expire-time: 1h victoria-metrics: enabled: true - url: http://victoria-metrics:8428 + url: http://victoria-metrics:8428 username: root password: root insert: @@ -222,7 +222,7 @@ alerter: region: AWS_REGION_FOR_END_USER_MESSAGING twilio: account-sid: YOUR_ACCOUNT_SID - auth-token: YOUR_AUTH_TOKEN + auth-token: YOUR_AUTH_TOKEN twilio-phone-number: YOUR_TWILIO_PHONE_NUMBER scheduler: server: @@ -273,3 +273,8 @@ hertzbeat: concurrency-limit: 256 reject-when-limit-reached: true task-termination-timeout: 5000 + # Bounds on what a single request to the push gateway may consume. + push: + max-auto-created-monitors: ${HERTZBEAT_PUSH_MAX_AUTO_CREATED_MONITORS:10000} + max-body-bytes: ${HERTZBEAT_PUSH_MAX_BODY_BYTES:5242880} + max-samples: ${HERTZBEAT_PUSH_MAX_SAMPLES:10000} diff --git a/script/docker-compose/hertzbeat-postgresql-greptimedb/conf/application.yml b/script/docker-compose/hertzbeat-postgresql-greptimedb/conf/application.yml index 6cc1490030..c25b533a48 100644 --- a/script/docker-compose/hertzbeat-postgresql-greptimedb/conf/application.yml +++ b/script/docker-compose/hertzbeat-postgresql-greptimedb/conf/application.yml @@ -108,7 +108,7 @@ spring: hibernate: format_sql: true dialect: org.hibernate.dialect.PostgreSQLDialect - + flyway: enabled: true clean-disabled: true @@ -116,7 +116,7 @@ spring: baseline-version: 1 locations: - classpath:db/migration/postgresql - + # Not Require, Please config if you need email notify mail: # Attention: this is mail server address. @@ -219,7 +219,7 @@ alerter: region: AWS_REGION_FOR_END_USER_MESSAGING twilio: account-sid: YOUR_ACCOUNT_SID - auth-token: YOUR_AUTH_TOKEN + auth-token: YOUR_AUTH_TOKEN twilio-phone-number: YOUR_TWILIO_PHONE_NUMBER scheduler: server: @@ -270,3 +270,8 @@ hertzbeat: concurrency-limit: 256 reject-when-limit-reached: true task-termination-timeout: 5000 + # Bounds on what a single request to the push gateway may consume. + push: + max-auto-created-monitors: ${HERTZBEAT_PUSH_MAX_AUTO_CREATED_MONITORS:10000} + max-body-bytes: ${HERTZBEAT_PUSH_MAX_BODY_BYTES:5242880} + max-samples: ${HERTZBEAT_PUSH_MAX_SAMPLES:10000} diff --git a/script/docker-compose/hertzbeat-postgresql-victoria-metrics/conf/application.yml b/script/docker-compose/hertzbeat-postgresql-victoria-metrics/conf/application.yml index 3108f5b994..71d7fb2f58 100644 --- a/script/docker-compose/hertzbeat-postgresql-victoria-metrics/conf/application.yml +++ b/script/docker-compose/hertzbeat-postgresql-victoria-metrics/conf/application.yml @@ -108,7 +108,7 @@ spring: hibernate: format_sql: true dialect: org.hibernate.dialect.PostgreSQLDialect - + flyway: enabled: true clean-disabled: true @@ -116,7 +116,7 @@ spring: baseline-version: 1 locations: - classpath:db/migration/postgresql - + # Not Require, Please config if you need email notify mail: # Attention: this is mail server address. @@ -221,7 +221,7 @@ alerter: region: AWS_REGION_FOR_END_USER_MESSAGING twilio: account-sid: YOUR_ACCOUNT_SID - auth-token: YOUR_AUTH_TOKEN + auth-token: YOUR_AUTH_TOKEN twilio-phone-number: YOUR_TWILIO_PHONE_NUMBER scheduler: server: @@ -272,3 +272,8 @@ hertzbeat: concurrency-limit: 256 reject-when-limit-reached: true task-termination-timeout: 5000 + # Bounds on what a single request to the push gateway may consume. + push: + max-auto-created-monitors: ${HERTZBEAT_PUSH_MAX_AUTO_CREATED_MONITORS:10000} + max-body-bytes: ${HERTZBEAT_PUSH_MAX_BODY_BYTES:5242880} + max-samples: ${HERTZBEAT_PUSH_MAX_SAMPLES:10000}