This is an automated email from the ASF dual-hosted git repository. zqr10159 pushed a commit to branch 2.0.0 in repository https://gitbox.apache.org/repos/asf/hertzbeat.git
commit af2e534dfb5bd4fc678bb5eee196aa5ed46f201c Author: Logic <[email protected]> AuthorDate: Wed Aug 26 15:28:10 2026 +0800 feat: bound concurrent OTLP writes --- .../admission/OtlpIngestionAdmissionService.java | 164 +++++++++++++++++ .../service/impl/OtlpGrpcIngestionServiceImpl.java | 56 ++++-- .../OtlpIngestionAdmissionServiceTest.java | 203 +++++++++++++++++++++ .../src/main/resources/application.yml | 5 + 4 files changed, 415 insertions(+), 13 deletions(-) diff --git a/hertzbeat-observability/src/main/java/org/apache/hertzbeat/observability/ingestion/admission/OtlpIngestionAdmissionService.java b/hertzbeat-observability/src/main/java/org/apache/hertzbeat/observability/ingestion/admission/OtlpIngestionAdmissionService.java new file mode 100644 index 0000000000..29da67db7c --- /dev/null +++ b/hertzbeat-observability/src/main/java/org/apache/hertzbeat/observability/ingestion/admission/OtlpIngestionAdmissionService.java @@ -0,0 +1,164 @@ +/* + * 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.observability.ingestion.admission; + +import io.grpc.Context; +import io.grpc.Status; +import java.time.Duration; +import java.util.Objects; +import java.util.concurrent.Semaphore; +import java.util.concurrent.TimeUnit; +import java.util.function.Supplier; +import org.apache.hertzbeat.observability.ingestion.error.OtlpIngestionBackpressureHeaders; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.stereotype.Service; + +/** + * Bounded admission control for OTLP storage writes. + */ +@Service +public class OtlpIngestionAdmissionService { + + private final Lane metrics; + private final Lane logs; + private final Lane traces; + + @Autowired + public OtlpIngestionAdmissionService( + @Value("${hertzbeat.otlp.ingestion.admission.max-concurrent-requests:32}") int maxConcurrentRequests, + @Value("${hertzbeat.otlp.ingestion.admission.queue-capacity:32}") int queueCapacity, + @Value("${hertzbeat.otlp.ingestion.admission.max-queue-wait:100ms}") Duration maxQueueWait) { + this.metrics = new Lane("metrics", maxConcurrentRequests, queueCapacity, maxQueueWait); + this.logs = new Lane("logs", maxConcurrentRequests, queueCapacity, maxQueueWait); + this.traces = new Lane("traces", maxConcurrentRequests, queueCapacity, maxQueueWait); + } + + /** + * Execute one storage write through the lane dedicated to its signal. + * + * @param signal OTLP signal name + * @param operation storage operation + * @param <T> result type + * @return operation result + */ + public <T> T execute(String signal, Supplier<T> operation) { + return lane(signal).execute(Objects.requireNonNull(operation, "operation")); + } + + private Lane lane(String signal) { + return switch (signal) { + case "metrics" -> metrics; + case "logs" -> logs; + case "traces" -> traces; + default -> throw new IllegalArgumentException("Unsupported OTLP signal: " + signal); + }; + } + + private static final class Lane { + + private final String signal; + private final Semaphore admittedRequests; + private final Semaphore activeRequests; + private final long maxQueueWaitNanos; + + private Lane(String signal, int maxConcurrentRequests, int queueCapacity, Duration maxQueueWait) { + if (maxConcurrentRequests <= 0) { + throw new IllegalArgumentException("maxConcurrentRequests must be greater than zero"); + } + if (queueCapacity < 0) { + throw new IllegalArgumentException("queueCapacity must not be negative"); + } + long admissionCapacity = (long) maxConcurrentRequests + queueCapacity; + if (admissionCapacity > Integer.MAX_VALUE) { + throw new IllegalArgumentException("Combined admission capacity exceeds the supported limit"); + } + this.signal = signal; + this.admittedRequests = new Semaphore((int) admissionCapacity, true); + this.activeRequests = new Semaphore(maxConcurrentRequests, true); + this.maxQueueWaitNanos = durationNanos(maxQueueWait); + } + + private <T> T execute(Supplier<T> operation) { + Context requestContext = Context.current(); + Thread requestThread = Thread.currentThread(); + Context.CancellationListener cancellationListener = ignored -> requestThread.interrupt(); + requestContext.addListener(cancellationListener, Runnable::run); + if (!admittedRequests.tryAcquire()) { + requestContext.removeListener(cancellationListener); + throw requestContext.isCancelled() ? cancelled(null) : busy(); + } + boolean active = false; + try { + if (requestContext.isCancelled()) { + throw cancelled(null); + } + if (!activeRequests.tryAcquire(maxQueueWaitNanos, TimeUnit.NANOSECONDS)) { + throw busy(); + } + active = true; + T result = operation.get(); + if (requestContext.isCancelled()) { + Thread.interrupted(); + throw cancelled(null); + } + return result; + } catch (InterruptedException exception) { + if (!requestContext.isCancelled()) { + Thread.currentThread().interrupt(); + } + throw cancelled(exception); + } catch (RuntimeException exception) { + if (requestContext.isCancelled()) { + Thread.interrupted(); + throw cancelled(exception); + } + throw exception; + } finally { + if (active) { + activeRequests.release(); + } + admittedRequests.release(); + requestContext.removeListener(cancellationListener); + } + } + + private RuntimeException busy() { + return OtlpIngestionBackpressureHeaders.statusRuntimeException( + Status.RESOURCE_EXHAUSTED, "OTLP " + signal + " ingestion is busy.", null); + } + + private RuntimeException cancelled(Throwable cause) { + return Status.CANCELLED + .withDescription("OTLP " + signal + " ingestion was cancelled.") + .withCause(cause) + .asRuntimeException(); + } + + private static long durationNanos(Duration duration) { + if (duration == null || duration.isNegative() || duration.isZero()) { + return 0L; + } + try { + return duration.toNanos(); + } catch (ArithmeticException ignored) { + return Long.MAX_VALUE; + } + } + } +} diff --git a/hertzbeat-observability/src/main/java/org/apache/hertzbeat/observability/ingestion/service/impl/OtlpGrpcIngestionServiceImpl.java b/hertzbeat-observability/src/main/java/org/apache/hertzbeat/observability/ingestion/service/impl/OtlpGrpcIngestionServiceImpl.java index d5d3fbbe2a..cd20fdc90c 100644 --- a/hertzbeat-observability/src/main/java/org/apache/hertzbeat/observability/ingestion/service/impl/OtlpGrpcIngestionServiceImpl.java +++ b/hertzbeat-observability/src/main/java/org/apache/hertzbeat/observability/ingestion/service/impl/OtlpGrpcIngestionServiceImpl.java @@ -70,8 +70,8 @@ import java.util.zip.GZIPInputStream; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.StringUtils; import org.apache.hertzbeat.common.observability.gateway.ObservabilitySignalIntakeGateway; -import org.apache.hertzbeat.observability.ingestion.service.OtlpGrpcIngestionService; import org.apache.hertzbeat.observability.ingestion.adapter.OtlpLogProtocolAdapter; +import org.apache.hertzbeat.observability.ingestion.admission.OtlpIngestionAdmissionService; import org.apache.hertzbeat.observability.ingestion.audit.OtlpIngestionAuditService; import org.apache.hertzbeat.observability.ingestion.enricher.OtlpCorrelationContext; import org.apache.hertzbeat.observability.ingestion.enricher.OtlpCorrelationEnricher; @@ -85,6 +85,7 @@ import org.apache.hertzbeat.observability.ingestion.redaction.OtlpIngestionRedac import org.apache.hertzbeat.observability.ingestion.retry.OtlpIngestionRetryService; import org.apache.hertzbeat.observability.ingestion.semantic.OtlpResourceSemanticAttributes; import org.apache.hertzbeat.observability.ingestion.security.OtlpIngestionRequestContextResolver; +import org.apache.hertzbeat.observability.ingestion.service.OtlpGrpcIngestionService; import org.apache.hertzbeat.warehouse.store.history.tsdb.greptime.GreptimeProperties; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; @@ -177,6 +178,7 @@ public class OtlpGrpcIngestionServiceImpl implements OtlpGrpcIngestionService { private final OtlpIngestionAuditService auditService; private final OtlpIngestionGovernanceService governanceService; private final OtlpIngestionQuotaService quotaService; + private final OtlpIngestionAdmissionService admissionService; private final ObservabilitySignalIntakeGateway observabilitySignalIntakeGateway; private final OtlpEntityIdentityResolver otlpEntityIdentityResolver; private final OtlpIngestionRedactionService redactionService = new OtlpIngestionRedactionService(); @@ -198,10 +200,9 @@ public class OtlpGrpcIngestionServiceImpl implements OtlpGrpcIngestionService { this(restTemplate, greptimePropertiesProvider, otlpLogProtocolAdapter, greptimeOtlpForwarder, otlpCorrelationEnricher, errorResponseFactory, requestContextResolver, auditService, governanceService, quotaService, observabilitySignalIntakeGateway, otlpEntityIdentityResolver, - new OtlpIngestionRetryService()); + new OtlpIngestionRetryService(), defaultAdmissionService()); } - @Autowired public OtlpGrpcIngestionServiceImpl(RestTemplate restTemplate, ObjectProvider<GreptimeProperties> greptimePropertiesProvider, OtlpLogProtocolAdapter otlpLogProtocolAdapter, @@ -216,6 +217,28 @@ public class OtlpGrpcIngestionServiceImpl implements OtlpGrpcIngestionService { ObservabilitySignalIntakeGateway observabilitySignalIntakeGateway, OtlpEntityIdentityResolver otlpEntityIdentityResolver, OtlpIngestionRetryService retryService) { + this(restTemplate, greptimePropertiesProvider, otlpLogProtocolAdapter, greptimeOtlpForwarder, + otlpCorrelationEnricher, errorResponseFactory, requestContextResolver, auditService, governanceService, + quotaService, observabilitySignalIntakeGateway, otlpEntityIdentityResolver, + retryService, defaultAdmissionService()); + } + + @Autowired + public OtlpGrpcIngestionServiceImpl(RestTemplate restTemplate, + ObjectProvider<GreptimeProperties> greptimePropertiesProvider, + OtlpLogProtocolAdapter otlpLogProtocolAdapter, + GreptimeOtlpForwarder greptimeOtlpForwarder, + OtlpCorrelationEnricher otlpCorrelationEnricher, + OtlpIngestionErrorResponseFactory errorResponseFactory, + OtlpIngestionRequestContextResolver requestContextResolver, + OtlpIngestionAuditService auditService, + OtlpIngestionGovernanceService governanceService, + OtlpIngestionQuotaService quotaService, + @Qualifier("telemetryIntakeServiceImpl") + ObservabilitySignalIntakeGateway observabilitySignalIntakeGateway, + OtlpEntityIdentityResolver otlpEntityIdentityResolver, + OtlpIngestionRetryService retryService, + OtlpIngestionAdmissionService admissionService) { this.restTemplate = restTemplate; this.greptimePropertiesProvider = greptimePropertiesProvider; this.otlpLogProtocolAdapter = otlpLogProtocolAdapter; @@ -226,11 +249,16 @@ public class OtlpGrpcIngestionServiceImpl implements OtlpGrpcIngestionService { this.auditService = auditService; this.governanceService = governanceService; this.quotaService = quotaService; + this.admissionService = admissionService == null ? defaultAdmissionService() : admissionService; this.observabilitySignalIntakeGateway = observabilitySignalIntakeGateway; this.otlpEntityIdentityResolver = otlpEntityIdentityResolver; this.retryService = retryService == null ? new OtlpIngestionRetryService() : retryService; } + private static OtlpIngestionAdmissionService defaultAdmissionService() { + return new OtlpIngestionAdmissionService(32, 32, java.time.Duration.ofMillis(100)); + } + @Override public ResponseEntity<byte[]> ingestMetricsHttp(byte[] content, HttpHeaders requestHeaders) { long startedAtNanos = System.nanoTime(); @@ -254,8 +282,8 @@ public class OtlpGrpcIngestionServiceImpl implements OtlpGrpcIngestionService { governanceDecision.reason(), durationMillis(startedAtNanos)); return emptySignalHttpSuccess(contentType, safeRequestHeaders.getAccept(), false); } - ResponseEntity<byte[]> response = proxySignalHttp( - safeContent, safeRequestHeaders, "/v1/otlp/v1/metrics", false, correlationContext); + ResponseEntity<byte[]> response = admissionService.execute("metrics", () -> proxySignalHttp( + safeContent, safeRequestHeaders, "/v1/otlp/v1/metrics", false, correlationContext)); if (response.getStatusCode().is2xxSuccessful()) { recordMetricIntake(request); auditService.recordAccepted("metrics", "http", correlationContext, requestBytes, signalItems, @@ -298,7 +326,8 @@ public class OtlpGrpcIngestionServiceImpl implements OtlpGrpcIngestionService { governanceDecision.reason(), durationMillis(startedAtNanos)); return emptyLogsHttpSuccess(contentType, safeRequestHeaders.getAccept()); } - ResponseEntity<byte[]> forwardResponse = greptimeOtlpForwarder.forwardLogsProtobuf(redactedContent); + ResponseEntity<byte[]> forwardResponse = admissionService.execute( + "logs", () -> greptimeOtlpForwarder.forwardLogsProtobuf(redactedContent)); if (forwardResponse == null) { throw io.grpc.Status.UNAVAILABLE.withDescription("OTLP backend returned no response.") .asRuntimeException(); @@ -363,8 +392,8 @@ public class OtlpGrpcIngestionServiceImpl implements OtlpGrpcIngestionService { governanceDecision.reason(), durationMillis(startedAtNanos)); return emptySignalHttpSuccess(contentType, safeRequestHeaders.getAccept(), true); } - ResponseEntity<byte[]> response = proxySignalHttp( - safeContent, safeRequestHeaders, "/v1/otlp/v1/traces", true, correlationContext); + ResponseEntity<byte[]> response = admissionService.execute("traces", () -> proxySignalHttp( + safeContent, safeRequestHeaders, "/v1/otlp/v1/traces", true, correlationContext)); if (response.getStatusCode().is2xxSuccessful()) { recordTraceIntake(request); auditService.recordAccepted("traces", "http", correlationContext, requestBytes, signalItems, @@ -396,8 +425,8 @@ public class OtlpGrpcIngestionServiceImpl implements OtlpGrpcIngestionService { governanceDecision.reason(), durationMillis(startedAtNanos)); return ExportMetricsServiceResponse.getDefaultInstance(); } - byte[] response = proxySignalBinary(redactedRequest.toByteArray(), "/v1/otlp/v1/metrics", false, - correlationContext); + byte[] response = admissionService.execute("metrics", () -> proxySignalBinary( + redactedRequest.toByteArray(), "/v1/otlp/v1/metrics", false, correlationContext)); ExportMetricsServiceResponse parsedResponse = response.length == 0 ? ExportMetricsServiceResponse.getDefaultInstance() : ExportMetricsServiceResponse.parseFrom(response); @@ -440,7 +469,8 @@ public class OtlpGrpcIngestionServiceImpl implements OtlpGrpcIngestionService { governanceDecision.reason(), durationMillis(startedAtNanos)); return ExportLogsServiceResponse.getDefaultInstance(); } - byte[] response = greptimeOtlpForwarder.forwardLogsGrpc(redactedRequest); + byte[] response = admissionService.execute( + "logs", () -> greptimeOtlpForwarder.forwardLogsGrpc(redactedRequest)); if (response == null) { throw io.grpc.Status.UNAVAILABLE.withDescription("OTLP backend returned no response.") .asRuntimeException(); @@ -495,8 +525,8 @@ public class OtlpGrpcIngestionServiceImpl implements OtlpGrpcIngestionService { governanceDecision.reason(), durationMillis(startedAtNanos)); return ExportTraceServiceResponse.getDefaultInstance(); } - byte[] response = proxySignalBinary( - enrichedRequest.toByteArray(), "/v1/otlp/v1/traces", true, correlationContext); + byte[] response = admissionService.execute("traces", () -> proxySignalBinary( + enrichedRequest.toByteArray(), "/v1/otlp/v1/traces", true, correlationContext)); ExportTraceServiceResponse parsedResponse = response.length == 0 ? ExportTraceServiceResponse.getDefaultInstance() : ExportTraceServiceResponse.parseFrom(response); diff --git a/hertzbeat-observability/src/test/java/org/apache/hertzbeat/observability/ingestion/admission/OtlpIngestionAdmissionServiceTest.java b/hertzbeat-observability/src/test/java/org/apache/hertzbeat/observability/ingestion/admission/OtlpIngestionAdmissionServiceTest.java new file mode 100644 index 0000000000..cbd2cd0eea --- /dev/null +++ b/hertzbeat-observability/src/test/java/org/apache/hertzbeat/observability/ingestion/admission/OtlpIngestionAdmissionServiceTest.java @@ -0,0 +1,203 @@ +/* + * 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.observability.ingestion.admission; + +import static org.awaitility.Awaitility.await; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import io.grpc.Context; +import io.grpc.Status; +import io.grpc.StatusRuntimeException; +import java.time.Duration; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; +import org.apache.hertzbeat.observability.ingestion.error.OtlpIngestionBackpressureHeaders; +import org.junit.jupiter.api.Test; + +class OtlpIngestionAdmissionServiceTest { + + @Test + void rejectsImmediatelyWhenConcurrencyIsFullAndRecoversAfterCompletion() throws Exception { + OtlpIngestionAdmissionService service = new OtlpIngestionAdmissionService(1, 0, Duration.ofMillis(50)); + CountDownLatch operationStarted = new CountDownLatch(1); + CountDownLatch releaseOperation = new CountDownLatch(1); + AtomicReference<Throwable> operationFailure = new AtomicReference<>(); + Thread activeRequest = startRequest(service, "metrics", operationStarted, releaseOperation, operationFailure); + + assertTrue(operationStarted.await(1, TimeUnit.SECONDS)); + long startedAt = System.nanoTime(); + StatusRuntimeException exception = assertThrows(StatusRuntimeException.class, + () -> service.execute("metrics", () -> "rejected")); + long elapsedMillis = Duration.ofNanos(System.nanoTime() - startedAt).toMillis(); + + assertEquals(Status.Code.RESOURCE_EXHAUSTED, exception.getStatus().getCode()); + assertEquals("1", OtlpIngestionBackpressureHeaders.retryAfter(exception)); + assertTrue(elapsedMillis < 250, "A full admission lane must reject without waiting"); + + releaseOperation.countDown(); + activeRequest.join(1_000); + assertFalse(activeRequest.isAlive()); + assertNull(operationFailure.get()); + assertEquals("recovered", service.execute("metrics", () -> "recovered")); + } + + @Test + void boundsQueueWaitBeforeRejecting() throws Exception { + OtlpIngestionAdmissionService service = new OtlpIngestionAdmissionService(1, 1, Duration.ofMillis(50)); + CountDownLatch operationStarted = new CountDownLatch(1); + CountDownLatch releaseOperation = new CountDownLatch(1); + AtomicReference<Throwable> operationFailure = new AtomicReference<>(); + Thread activeRequest = startRequest(service, "logs", operationStarted, releaseOperation, operationFailure); + + assertTrue(operationStarted.await(1, TimeUnit.SECONDS)); + long startedAt = System.nanoTime(); + StatusRuntimeException exception = assertThrows(StatusRuntimeException.class, + () -> service.execute("logs", () -> "timed-out")); + long elapsedMillis = Duration.ofNanos(System.nanoTime() - startedAt).toMillis(); + + assertEquals(Status.Code.RESOURCE_EXHAUSTED, exception.getStatus().getCode()); + assertTrue(elapsedMillis >= 25, "A queued request should get a short opportunity to run"); + assertTrue(elapsedMillis < 500, "Queue waiting must remain bounded"); + + releaseOperation.countDown(); + activeRequest.join(1_000); + assertFalse(activeRequest.isAlive()); + assertNull(operationFailure.get()); + } + + @Test + void interruptionCancelsQueuedRequestWithoutLeakingCapacity() throws Exception { + OtlpIngestionAdmissionService service = new OtlpIngestionAdmissionService(1, 1, Duration.ofSeconds(5)); + CountDownLatch operationStarted = new CountDownLatch(1); + CountDownLatch releaseOperation = new CountDownLatch(1); + AtomicReference<Throwable> operationFailure = new AtomicReference<>(); + Thread activeRequest = startRequest(service, "traces", operationStarted, releaseOperation, operationFailure); + assertTrue(operationStarted.await(1, TimeUnit.SECONDS)); + + AtomicReference<Throwable> queuedFailure = new AtomicReference<>(); + Thread queuedRequest = Thread.ofVirtual().start(() -> { + try { + service.execute("traces", () -> "queued"); + } catch (Throwable throwable) { + queuedFailure.set(throwable); + } + }); + await().atMost(Duration.ofSeconds(1)).until(() -> isWaiting(queuedRequest)); + + queuedRequest.interrupt(); + queuedRequest.join(1_000); + assertFalse(queuedRequest.isAlive()); + StatusRuntimeException exception = (StatusRuntimeException) queuedFailure.get(); + assertEquals(Status.Code.CANCELLED, exception.getStatus().getCode()); + + releaseOperation.countDown(); + activeRequest.join(1_000); + assertFalse(activeRequest.isAlive()); + assertNull(operationFailure.get()); + assertEquals("recovered", service.execute("traces", () -> "recovered")); + } + + @Test + void grpcContextCancellationRemovesQueuedRequest() throws Exception { + OtlpIngestionAdmissionService service = new OtlpIngestionAdmissionService(1, 1, Duration.ofSeconds(5)); + CountDownLatch operationStarted = new CountDownLatch(1); + CountDownLatch releaseOperation = new CountDownLatch(1); + AtomicReference<Throwable> operationFailure = new AtomicReference<>(); + Thread activeRequest = startRequest(service, "logs", operationStarted, releaseOperation, operationFailure); + assertTrue(operationStarted.await(1, TimeUnit.SECONDS)); + + AtomicReference<Throwable> queuedFailure = new AtomicReference<>(); + Context.CancellableContext context = Context.current().withCancellation(); + Thread queuedRequest = Thread.ofVirtual().start(() -> context.run(() -> { + try { + service.execute("logs", () -> "queued"); + } catch (Throwable throwable) { + queuedFailure.set(throwable); + } + })); + await().atMost(Duration.ofSeconds(1)).until(() -> isWaiting(queuedRequest)); + + context.cancel(null); + queuedRequest.join(250); + boolean cancelledPromptly = !queuedRequest.isAlive(); + if (queuedRequest.isAlive()) { + queuedRequest.interrupt(); + } + releaseOperation.countDown(); + queuedRequest.join(1_000); + activeRequest.join(1_000); + + assertTrue(cancelledPromptly, "A cancelled gRPC request must leave the admission queue promptly"); + StatusRuntimeException exception = (StatusRuntimeException) queuedFailure.get(); + assertEquals(Status.Code.CANCELLED, exception.getStatus().getCode()); + assertFalse(activeRequest.isAlive()); + assertNull(operationFailure.get()); + assertEquals("recovered", service.execute("logs", () -> "recovered")); + } + + @Test + void isolatesSignalAdmissionLanes() throws Exception { + OtlpIngestionAdmissionService service = new OtlpIngestionAdmissionService(1, 0, Duration.ZERO); + CountDownLatch operationStarted = new CountDownLatch(1); + CountDownLatch releaseOperation = new CountDownLatch(1); + AtomicReference<Throwable> operationFailure = new AtomicReference<>(); + Thread activeRequest = startRequest(service, "metrics", operationStarted, releaseOperation, operationFailure); + + assertTrue(operationStarted.await(1, TimeUnit.SECONDS)); + assertEquals("logs-accepted", service.execute("logs", () -> "logs-accepted")); + + releaseOperation.countDown(); + activeRequest.join(1_000); + assertFalse(activeRequest.isAlive()); + assertNull(operationFailure.get()); + } + + private Thread startRequest(OtlpIngestionAdmissionService service, String signal, + CountDownLatch operationStarted, CountDownLatch releaseOperation, + AtomicReference<Throwable> failure) { + return Thread.ofVirtual().start(() -> { + try { + service.execute(signal, () -> { + operationStarted.countDown(); + awaitRelease(releaseOperation); + return "completed"; + }); + } catch (Throwable throwable) { + failure.set(throwable); + } + }); + } + + private void awaitRelease(CountDownLatch releaseOperation) { + try { + releaseOperation.await(); + } catch (InterruptedException exception) { + Thread.currentThread().interrupt(); + throw new IllegalStateException("Active test operation was interrupted", exception); + } + } + + private boolean isWaiting(Thread thread) { + return thread.getState() == Thread.State.WAITING || thread.getState() == Thread.State.TIMED_WAITING; + } +} diff --git a/hertzbeat-startup/src/main/resources/application.yml b/hertzbeat-startup/src/main/resources/application.yml index a10ebba43f..6a1e0b87da 100644 --- a/hertzbeat-startup/src/main/resources/application.yml +++ b/hertzbeat-startup/src/main/resources/application.yml @@ -363,6 +363,11 @@ hertzbeat: enabled: true host: 0.0.0.0 port: 4317 + ingestion: + admission: + max-concurrent-requests: ${HERTZBEAT_OTLP_MAX_CONCURRENT_REQUESTS:32} + queue-capacity: ${HERTZBEAT_OTLP_QUEUE_CAPACITY:32} + max-queue-wait: ${HERTZBEAT_OTLP_MAX_QUEUE_WAIT:100ms} # Optional virtual-thread overrides. Remove this whole block to use built-in defaults. vthreads: enabled: true --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
