[fix] bound what one anonymous push request can consume (#4273)

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<String, MetricFamily> parseMetrics(InputStream inputStream) throws IOException {
-        Map<String, MetricFamily> 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<String, MetricFamily> parseMetrics(InputStream inputStream, int maxSamples) throws IOException {
+        if (maxSamples < 0) {
+            throw new IllegalArgumentException("maxSamples must not be negative");
+        }
+        final Map<String, MetricFamily> 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.
      * <p>

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<String, MetricFamily> metricFamilyMap = OnlineParser.parseMetrics(inputStream, 2);
+
+        assertNotNull(metricFamilyMap);
+        assertEquals(2, metricFamilyMap.size());
+    }
 }

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<String, Long> jobInstanceMap;
-    
-    public PushGatewayServiceImpl(CommonDataQueue commonDataQueue, PushMonitorDao pushMonitorDao) {
+    private final Map<JobInstance, Long> jobInstanceMap;
+
+    /**
+     * Cap on push monitors created automatically from unknown job/instance pairs.
+     *
+     * <p>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.
+     *
+     * <p>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<JobInstance, CompletableFuture<Long>> 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<String, MetricFamily> metricFamilyMap = OnlineParser.parseMetrics(inputStream);
+            final long curTime = Instant.now().toEpochMilli();
+            final Map<String, MetricFamily> 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<String, MetricFamily> 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<String, String> 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<String, String> 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.
+     *
+     * <p>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<Long> proposedCreation = new CompletableFuture<>();
+        final CompletableFuture<Long> 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.
+     *
+     * <p>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();
+        }
+    }
 }

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}.
+ *
+ * <p>`/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<Future<?>> 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<Future<Boolean>> 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<Boolean> 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.
+     *
+     * <p>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<CollectRep.MetricsData> 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);
+    }
+}

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}

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}

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}

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}

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}

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}

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}