diff --git a/src/main/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3HandlerInstaller.java b/src/main/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3HandlerInstaller.java index b975262e04..48ed92a819 100644 --- a/src/main/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3HandlerInstaller.java +++ b/src/main/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3HandlerInstaller.java @@ -1,14 +1,17 @@ package io.stargate.sgv2.jsonapi.service.billing; +import com.google.common.annotations.VisibleForTesting; import io.micrometer.core.instrument.MeterRegistry; import io.quarkus.runtime.ShutdownEvent; import io.quarkus.runtime.StartupEvent; +import io.smallrye.mutiny.infrastructure.Infrastructure; import io.stargate.sgv2.jsonapi.config.BillingS3ExportConfig; import io.stargate.sgv2.jsonapi.metrics.BatchedLogBufferMetrics; import io.stargate.sgv2.jsonapi.metrics.BatchedLogUploaderMetrics; import jakarta.enterprise.context.ApplicationScoped; import jakarta.enterprise.event.Observes; import jakarta.inject.Inject; +import java.util.function.BiFunction; import java.util.logging.Logger; import org.slf4j.LoggerFactory; @@ -28,13 +31,24 @@ public class BillingS3HandlerInstaller { private final BillingS3ExportConfig config; private final MeterRegistry meterRegistry; + private final BiFunction + handlerFactory; private volatile BillingS3LogHandler handler; @Inject public BillingS3HandlerInstaller(BillingS3ExportConfig config, MeterRegistry meterRegistry) { + this(config, meterRegistry, BillingS3LogHandler::new); + } + + @VisibleForTesting + BillingS3HandlerInstaller( + BillingS3ExportConfig config, + MeterRegistry meterRegistry, + BiFunction handlerFactory) { this.config = config; this.meterRegistry = meterRegistry; + this.handlerFactory = handlerFactory; } void onStart(@Observes StartupEvent event) { @@ -62,14 +76,14 @@ void onStart(@Observes StartupEvent event) { config.queueCapacity(), new BatchedLogBufferMetrics(meterRegistry, METRICS_PREFIX)); LOGGER.info("Billing is using log buffer: {}", buffer); - this.handler = new BillingS3LogHandler(buffer, uploader); + this.handler = handlerFactory.apply(buffer, uploader); // TODO: LOGGER NAME SHOULD BE IN CONFIG Logger.getLogger(BILLING_LOGGER_NAME).addHandler(this.handler); LOGGER.info( "Billing has attached BillingS3LogHandler to the logger named: {}", BILLING_LOGGER_NAME); - // TODO: XXXX call start on the thread. + Infrastructure.getDefaultWorkerPool().execute(this.handler::startUploading); } void onStop(@Observes ShutdownEvent event) { diff --git a/src/main/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3LogHandler.java b/src/main/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3LogHandler.java index 76633fe105..4e5a239356 100644 --- a/src/main/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3LogHandler.java +++ b/src/main/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3LogHandler.java @@ -56,7 +56,7 @@ public final class BillingS3LogHandler extends Handler { * Started at 1 and then decremented in {@link #startUploading()} when it exists so we know we * have finished uploading. */ - private final CountDownLatch uploadingFinished = new CountDownLatch(0); + private final CountDownLatch uploadingFinished = new CountDownLatch(1); private final AsyncBatchedLogUploader uploader; private final BatchedLogBuffer batchedLogBuffer; diff --git a/src/test/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3HandlerInstallerTest.java b/src/test/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3HandlerInstallerTest.java index 741f783864..56a1597799 100644 --- a/src/test/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3HandlerInstallerTest.java +++ b/src/test/java/io/stargate/sgv2/jsonapi/service/billing/BillingS3HandlerInstallerTest.java @@ -1,11 +1,60 @@ package io.stargate.sgv2.jsonapi.service.billing; +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.timeout; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import io.micrometer.core.instrument.simple.SimpleMeterRegistry; +import io.quarkus.runtime.ShutdownEvent; +import io.quarkus.runtime.StartupEvent; +import io.stargate.sgv2.jsonapi.config.BillingS3ExportConfig; +import java.time.Duration; +import java.util.Optional; +import java.util.concurrent.atomic.AtomicBoolean; +import org.junit.jupiter.api.Test; + /** * Unit tests for {@link BillingS3HandlerInstaller}: install/uninstall symmetry on the {@code * billing.events} JUL logger, the disabled path, and fail-loud startup on bad config. Delivery * through an installed handler is covered by {@code BillingS3ExportIntegrationTest}. */ class BillingS3HandlerInstallerTest { + + @Test + void startupCreatesAndStartsHandler() { + var config = mock(BillingS3ExportConfig.class); + when(config.enabled()).thenReturn(true); + when(config.region()).thenReturn("us-east-2"); + when(config.bucket()).thenReturn("test-bucket"); + when(config.endpointOverride()).thenReturn(Optional.empty()); + when(config.maxEventsPerBatch()).thenReturn(1); + when(config.maxBytesPerBatch()).thenReturn(1_000_000L); + when(config.maxAge()).thenReturn(Duration.ofMinutes(1)); + when(config.queueCapacity()).thenReturn(10); + var handler = mock(BillingS3LogHandler.class); + var handlerCreated = new AtomicBoolean(); + var installer = + new BillingS3HandlerInstaller( + config, + new SimpleMeterRegistry(), + (buffer, uploader) -> { + assertThat(buffer).isNotNull(); + assertThat(uploader).isNotNull(); + handlerCreated.set(true); + return handler; + }); + + installer.onStart(new StartupEvent()); + try { + assertThat(handlerCreated).isTrue(); + verify(handler, timeout(10_000)).startUploading(); + } finally { + installer.onStop(new ShutdownEvent()); + } + } + // // private static BillingS3ExportConfig config(boolean enabled, String bucket, String region) { // BillingS3ExportConfig config = mock(BillingS3ExportConfig.class);