diff --git a/micrometer-core/src/main/java/io/micronaut/configuration/metrics/binder/logging/LogbackMeterRegistryBinderFactory.java b/micrometer-core/src/main/java/io/micronaut/configuration/metrics/binder/logging/LogbackMeterRegistryBinderFactory.java index df1d940a1..f590780c4 100644 --- a/micrometer-core/src/main/java/io/micronaut/configuration/metrics/binder/logging/LogbackMeterRegistryBinderFactory.java +++ b/micrometer-core/src/main/java/io/micronaut/configuration/metrics/binder/logging/LogbackMeterRegistryBinderFactory.java @@ -48,6 +48,6 @@ public class LogbackMeterRegistryBinderFactory { @Singleton @Primary public LogbackMetrics logbackMetrics() { - return new LogbackMetrics(); + return new MicronautLogbackMetrics(); } } diff --git a/micrometer-core/src/main/java/io/micronaut/configuration/metrics/binder/logging/MicronautLogbackMetrics.java b/micrometer-core/src/main/java/io/micronaut/configuration/metrics/binder/logging/MicronautLogbackMetrics.java new file mode 100644 index 000000000..56e09bd6f --- /dev/null +++ b/micrometer-core/src/main/java/io/micronaut/configuration/metrics/binder/logging/MicronautLogbackMetrics.java @@ -0,0 +1,201 @@ +/* + * Copyright 2017-2019 original authors + * + * Licensed 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 + * + * https://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 io.micronaut.configuration.metrics.binder.logging; + +import ch.qos.logback.classic.Level; +import ch.qos.logback.classic.Logger; +import ch.qos.logback.classic.LoggerContext; +import ch.qos.logback.classic.spi.LoggerContextListener; +import ch.qos.logback.classic.turbo.TurboFilter; +import ch.qos.logback.core.spi.FilterReply; +import io.micrometer.core.instrument.FunctionCounter; +import io.micrometer.core.instrument.MeterRegistry; +import io.micrometer.core.instrument.Tag; +import io.micrometer.core.instrument.binder.BaseUnits; +import io.micrometer.core.instrument.binder.logging.LogbackMetrics; +import org.slf4j.LoggerFactory; +import org.slf4j.Marker; + +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.atomic.LongAdder; + +import static java.util.Collections.emptyList; + +/** + * Async-safe variant of {@link LogbackMetrics}. + */ +final class MicronautLogbackMetrics extends LogbackMetrics { + + private final Iterable tags; + private final LoggerContext loggerContext; + private final Map metricsTurboFilters = new HashMap<>(); + private final LoggerContextListener resetListener; + + MicronautLogbackMetrics() { + this(emptyList()); + } + + MicronautLogbackMetrics(Iterable tags) { + this(tags, (LoggerContext) LoggerFactory.getILoggerFactory()); + } + + MicronautLogbackMetrics(Iterable tags, LoggerContext loggerContext) { + super(tags, loggerContext); + this.tags = tags; + this.loggerContext = loggerContext; + + this.resetListener = new LoggerContextListener() { + @Override + public boolean isResetResistant() { + return true; + } + + @Override + public void onReset(LoggerContext context) { + synchronized (metricsTurboFilters) { + for (MetricsTurboFilter metricsTurboFilter : metricsTurboFilters.values()) { + loggerContext.addTurboFilter(metricsTurboFilter); + } + } + } + + @Override + public void onStart(LoggerContext context) { + // no-op + } + + @Override + public void onStop(LoggerContext context) { + // no-op + } + + @Override + public void onLevelChange(Logger logger, Level level) { + // no-op + } + }; + loggerContext.addListener(resetListener); + } + + @Override + public void bindTo(MeterRegistry registry) { + synchronized (metricsTurboFilters) { + MetricsTurboFilter existingFilter = metricsTurboFilters.get(registry); + if (existingFilter != null) { + loggerContext.getTurboFilterList().remove(existingFilter); + loggerContext.addTurboFilter(existingFilter); + return; + } + MetricsTurboFilter filter = new MetricsTurboFilter(registry, tags); + metricsTurboFilters.put(registry, filter); + loggerContext.addTurboFilter(filter); + } + } + + @Override + public void close() { + synchronized (metricsTurboFilters) { + for (MetricsTurboFilter metricsTurboFilter : metricsTurboFilters.values()) { + loggerContext.getTurboFilterList().remove(metricsTurboFilter); + } + metricsTurboFilters.clear(); + } + loggerContext.removeListener(resetListener); + } + + static final class MetricsTurboFilter extends TurboFilter { + + private static final String METER_NAME = "logback.events"; + private static final String METER_DESCRIPTION = "Number of log events that were enabled by the effective log level"; + private static final String LEVEL_TAG = "level"; + + private final LongAdder errorCount = new LongAdder(); + private final LongAdder warnCount = new LongAdder(); + private final LongAdder infoCount = new LongAdder(); + private final LongAdder debugCount = new LongAdder(); + private final LongAdder traceCount = new LongAdder(); + + MetricsTurboFilter(MeterRegistry registry, Iterable tags) { + FunctionCounter.builder(METER_NAME, errorCount, LongAdder::doubleValue) + .tags(tags) + .tags(LEVEL_TAG, "error") + .description(METER_DESCRIPTION) + .baseUnit(BaseUnits.EVENTS) + .register(registry); + + FunctionCounter.builder(METER_NAME, warnCount, LongAdder::doubleValue) + .tags(tags) + .tags(LEVEL_TAG, "warn") + .description(METER_DESCRIPTION) + .baseUnit(BaseUnits.EVENTS) + .register(registry); + + FunctionCounter.builder(METER_NAME, infoCount, LongAdder::doubleValue) + .tags(tags) + .tags(LEVEL_TAG, "info") + .description(METER_DESCRIPTION) + .baseUnit(BaseUnits.EVENTS) + .register(registry); + + FunctionCounter.builder(METER_NAME, debugCount, LongAdder::doubleValue) + .tags(tags) + .tags(LEVEL_TAG, "debug") + .description(METER_DESCRIPTION) + .baseUnit(BaseUnits.EVENTS) + .register(registry); + + FunctionCounter.builder(METER_NAME, traceCount, LongAdder::doubleValue) + .tags(tags) + .tags(LEVEL_TAG, "trace") + .description(METER_DESCRIPTION) + .baseUnit(BaseUnits.EVENTS) + .register(registry); + } + + @Override + public FilterReply decide(Marker marker, Logger logger, Level level, String format, Object[] params, Throwable t) { + if (format == null || !level.isGreaterOrEqual(logger.getEffectiveLevel())) { + return FilterReply.NEUTRAL; + } + + recordMetrics(level); + return FilterReply.NEUTRAL; + } + + private void recordMetrics(Level level) { + switch (level.toInt()) { + case Level.ERROR_INT: + errorCount.increment(); + break; + case Level.WARN_INT: + warnCount.increment(); + break; + case Level.INFO_INT: + infoCount.increment(); + break; + case Level.DEBUG_INT: + debugCount.increment(); + break; + case Level.TRACE_INT: + traceCount.increment(); + break; + default: + break; + } + } + } +} diff --git a/micrometer-core/src/test/groovy/io/micronaut/configuration/metrics/binder/logging/LogbackMeterRegistryBinderFactorySpec.groovy b/micrometer-core/src/test/groovy/io/micronaut/configuration/metrics/binder/logging/LogbackMeterRegistryBinderFactorySpec.groovy index ae8e93c88..4e3d1c713 100644 --- a/micrometer-core/src/test/groovy/io/micronaut/configuration/metrics/binder/logging/LogbackMeterRegistryBinderFactorySpec.groovy +++ b/micrometer-core/src/test/groovy/io/micronaut/configuration/metrics/binder/logging/LogbackMeterRegistryBinderFactorySpec.groovy @@ -1,10 +1,20 @@ package io.micronaut.configuration.metrics.binder.logging +import io.micrometer.core.instrument.Counter +import io.micrometer.core.instrument.Meter import io.micrometer.core.instrument.binder.logging.LogbackMetrics +import io.micrometer.core.instrument.cumulative.CumulativeCounter +import io.micrometer.core.instrument.simple.SimpleConfig +import io.micrometer.core.instrument.simple.SimpleMeterRegistry import io.micronaut.context.ApplicationContext +import org.slf4j.LoggerFactory import spock.lang.Specification import spock.lang.Unroll +import java.util.concurrent.CountDownLatch +import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicBoolean + import static io.micronaut.configuration.metrics.micrometer.MeterRegistryFactory.MICRONAUT_METRICS_BINDERS import static io.micronaut.configuration.metrics.micrometer.MeterRegistryFactory.MICRONAUT_METRICS_ENABLED @@ -18,6 +28,25 @@ class LogbackMeterRegistryBinderFactorySpec extends Specification { binder.logbackMetrics() } + void "logback metrics do not recurse when counter emits an async log event"() { + given: + def logger = LoggerFactory.getLogger("logback-binder-recursion-test") + LogbackMetrics binder = new LogbackMeterRegistryBinderFactory().logbackMetrics() + def registry = new AsyncLoggingCounterRegistry() + binder.bindTo(registry) + + when: + logger.info("primary event") + + then: + !registry.awaitAsyncLog() + registry.get("logback.events").tags("level", "info").functionCounter().count() == 1d + + cleanup: + binder.close() + registry.close() + } + void "test getting the beans"() { when: ApplicationContext context = ApplicationContext.run() @@ -49,4 +78,88 @@ class LogbackMeterRegistryBinderFactorySpec extends Specification { MICRONAUT_METRICS_BINDERS + ".logback.enabled" | true MICRONAUT_METRICS_BINDERS + ".logback.enabled" | false } + + void "logback metrics close removes filters and reset listener"() { + given: + def loggerContext = newLoggerContext() + def binder = new MicronautLogbackMetrics([], loggerContext) + def registry = new SimpleMeterRegistry() + + when: + binder.bindTo(registry) + + then: + binder.@metricsTurboFilters.size() == 1 + loggerContext.getTurboFilterList().size() == 1 + loggerContext.getCopyOfListenerList().contains(binder.@resetListener) + + when: + binder.close() + loggerContext.reset() + + then: + binder.@metricsTurboFilters.isEmpty() + loggerContext.getTurboFilterList().isEmpty() + !loggerContext.getCopyOfListenerList().contains(binder.@resetListener) + + cleanup: + registry.close() + loggerContext.stop() + } + + void "logback metrics bind same registry once"() { + given: + def loggerContext = newLoggerContext() + def logger = loggerContext.getLogger("logback-binder-repeat-bind-test") + def binder = new MicronautLogbackMetrics([], loggerContext) + def registry = new SimpleMeterRegistry() + + when: + binder.bindTo(registry) + binder.bindTo(registry) + logger.info("primary event") + + then: + binder.@metricsTurboFilters.size() == 1 + loggerContext.getTurboFilterList().findAll { it instanceof MicronautLogbackMetrics.MetricsTurboFilter }.size() == 1 + registry.get("logback.events").tags("level", "info").functionCounter().count() == 1d + + cleanup: + binder.close() + registry.close() + loggerContext.stop() + } + + private static newLoggerContext() { + LoggerFactory.getILoggerFactory().class.getDeclaredConstructor().newInstance() + } + + private static final class AsyncLoggingCounterRegistry extends SimpleMeterRegistry { + private final AtomicBoolean asyncLogged = new AtomicBoolean() + private final CountDownLatch asyncLogLatch = new CountDownLatch(1) + + AsyncLoggingCounterRegistry() { + super(SimpleConfig.DEFAULT, io.micrometer.core.instrument.Clock.SYSTEM) + } + + @Override + protected Counter newCounter(Meter.Id id) { + return new CumulativeCounter(id) { + @Override + void increment(double amount) { + super.increment(amount) + if (asyncLogged.compareAndSet(false, true)) { + Thread.start { + LoggerFactory.getLogger("logback-binder-recursion-test").info("counter emitted event") + asyncLogLatch.countDown() + } + } + } + } + } + + boolean awaitAsyncLog() { + asyncLogLatch.await(5, TimeUnit.SECONDS) + } + } } diff --git a/micrometer-registry-statsd/src/main/java/io/micronaut/configuration/metrics/micrometer/statsd/NativeImageStatsdCondition.java b/micrometer-registry-statsd/src/main/java/io/micronaut/configuration/metrics/micrometer/statsd/NativeImageStatsdCondition.java new file mode 100644 index 000000000..688f89f86 --- /dev/null +++ b/micrometer-registry-statsd/src/main/java/io/micronaut/configuration/metrics/micrometer/statsd/NativeImageStatsdCondition.java @@ -0,0 +1,40 @@ +/* + * Copyright 2017-2026 original authors + * + * Licensed 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 + * + * https://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 io.micronaut.configuration.metrics.micrometer.statsd; + +import io.micronaut.context.condition.Condition; +import io.micronaut.context.condition.ConditionContext; + +/** + * Enables the native-image StatsD workaround only for UDP-based native executables. + */ +final class NativeImageStatsdCondition implements Condition { + + static final String NATIVE_IMAGE_CODE_PROPERTY = "org.graalvm.nativeimage.imagecode"; + private static final String RUNTIME = "runtime"; + private static final String UDP = "UDP"; + + @Override + public boolean matches(ConditionContext context) { + String imageCode = System.getProperty(NATIVE_IMAGE_CODE_PROPERTY); + if (!RUNTIME.equalsIgnoreCase(imageCode)) { + return false; + } + String protocol = context.getProperty(StatsdMeterRegistryFactory.STATSD_CONFIG + ".protocol", String.class) + .orElse(UDP); + return UDP.equalsIgnoreCase(protocol); + } +} diff --git a/micrometer-registry-statsd/src/main/java/io/micronaut/configuration/metrics/micrometer/statsd/NativeImageUdpStatsdLineSink.java b/micrometer-registry-statsd/src/main/java/io/micronaut/configuration/metrics/micrometer/statsd/NativeImageUdpStatsdLineSink.java new file mode 100644 index 000000000..813ede7dc --- /dev/null +++ b/micrometer-registry-statsd/src/main/java/io/micronaut/configuration/metrics/micrometer/statsd/NativeImageUdpStatsdLineSink.java @@ -0,0 +1,230 @@ +/* + * Copyright 2017-2026 original authors + * + * Licensed 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 + * + * https://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 io.micronaut.configuration.metrics.micrometer.statsd; + +import io.micrometer.statsd.StatsdConfig; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.Closeable; +import java.io.IOException; +import java.io.UncheckedIOException; +import java.net.DatagramPacket; +import java.net.DatagramSocket; +import java.net.InetAddress; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.TimeUnit; +import java.util.function.Consumer; + +/** + * Native-image friendly UDP line sink for StatsD. + */ +final class NativeImageUdpStatsdLineSink implements Consumer, Closeable { + + private static final Logger LOG = LoggerFactory.getLogger(NativeImageUdpStatsdLineSink.class); + + private final Object lock = new Object(); + private final String host; + private final int port; + private final int maxPacketLength; + private final boolean buffered; + private final ScheduledExecutorService scheduler; + private final Consumer sender; + + private StringBuilder buffer; + private int bufferBytes; + + NativeImageUdpStatsdLineSink(StatsdConfig config) { + this(config, new UdpSender(config.host(), config.port())); + } + + NativeImageUdpStatsdLineSink(StatsdConfig config, Consumer sender) { + this.host = config.host(); + this.port = config.port(); + this.maxPacketLength = config.maxPacketLength(); + this.buffered = config.buffered(); + this.sender = sender; + this.buffer = new StringBuilder(Math.max(128, maxPacketLength)); + Duration pollingFrequency = config.pollingFrequency(); + long pollingFrequencyMillis = pollingFrequency.toMillis(); + if (buffered && pollingFrequencyMillis > 0) { + this.scheduler = Executors.newSingleThreadScheduledExecutor(new StatsdThreadFactory()); + this.scheduler.scheduleAtFixedRate(this::flushSafely, + pollingFrequencyMillis, + pollingFrequencyMillis, + TimeUnit.MILLISECONDS); + } else { + this.scheduler = null; + } + } + + @Override + public void accept(String line) { + if (line == null || line.isEmpty()) { + return; + } + String payloadToSend = null; + synchronized (lock) { + if (!buffered) { + payloadToSend = line; + } else { + int lineBytes = line.getBytes(StandardCharsets.UTF_8).length; + int additionalLength = lineBytes; + if (!buffer.isEmpty()) { + additionalLength += 1; + } + if (bufferBytes + additionalLength > maxPacketLength && !buffer.isEmpty()) { + payloadToSend = clearBuffer(); + } + if (!buffer.isEmpty()) { + buffer.append('\n'); + bufferBytes++; + } + buffer.append(line); + bufferBytes += lineBytes; + } + } + if (payloadToSend != null) { + send(payloadToSend); + } + } + + private void flushSafely() { + String payloadToSend; + synchronized (lock) { + payloadToSend = clearBuffer(); + } + if (payloadToSend != null) { + send(payloadToSend); + } + } + + private void send(String payload) { + try { + sender.accept(payload); + } catch (RuntimeException e) { + LOG.debug("Error sending StatsD metrics to {}:{}", host, port, e); + } + } + + @Override + public void close() { + if (scheduler != null) { + scheduler.shutdownNow(); + } + String payloadToSend; + synchronized (lock) { + payloadToSend = clearBuffer(); + } + if (payloadToSend != null) { + send(payloadToSend); + } + closeSender(); + } + + private String clearBuffer() { + if (buffer.isEmpty()) { + return null; + } + String payload = buffer.toString(); + buffer = new StringBuilder(Math.max(128, maxPacketLength)); + bufferBytes = 0; + return payload; + } + + private void closeSender() { + if (sender instanceof Closeable closeableSender) { + try { + closeableSender.close(); + } catch (IOException e) { + LOG.debug("Error closing StatsD sender to {}:{}", host, port, e); + } + } + } + + private static final class UdpSender implements Consumer, Closeable { + private final Object lock = new Object(); + private final String host; + private final int port; + + private DatagramSocket socket; + private InetAddress address; + + private UdpSender(String host, int port) { + this.host = host; + this.port = port; + } + + @Override + public void accept(String payload) { + byte[] bytes = payload.getBytes(StandardCharsets.UTF_8); + synchronized (lock) { + try { + DatagramSocket currentSocket = getSocket(); + InetAddress currentAddress = getAddress(); + currentSocket.send(new DatagramPacket(bytes, bytes.length, currentAddress, port)); + } catch (IOException e) { + closeSocket(); + address = null; + throw new UncheckedIOException(e); + } + } + } + + @Override + public void close() { + synchronized (lock) { + closeSocket(); + address = null; + } + } + + private DatagramSocket getSocket() throws IOException { + if (socket == null || socket.isClosed()) { + socket = new DatagramSocket(); + } + return socket; + } + + private InetAddress getAddress() throws IOException { + if (address == null) { + address = InetAddress.getByName(host); + } + return address; + } + + private void closeSocket() { + if (socket != null) { + socket.close(); + socket = null; + } + } + } + + private static final class StatsdThreadFactory implements ThreadFactory { + @Override + public Thread newThread(Runnable runnable) { + Thread thread = Executors.defaultThreadFactory().newThread(runnable); + thread.setName("micronaut-statsd-native-sink"); + thread.setDaemon(true); + return thread; + } + } +} diff --git a/micrometer-registry-statsd/src/main/java/io/micronaut/configuration/metrics/micrometer/statsd/StatsdMeterRegistryFactory.java b/micrometer-registry-statsd/src/main/java/io/micronaut/configuration/metrics/micrometer/statsd/StatsdMeterRegistryFactory.java index b0f8286bc..75ee4f66d 100644 --- a/micrometer-registry-statsd/src/main/java/io/micronaut/configuration/metrics/micrometer/statsd/StatsdMeterRegistryFactory.java +++ b/micrometer-registry-statsd/src/main/java/io/micronaut/configuration/metrics/micrometer/statsd/StatsdMeterRegistryFactory.java @@ -16,14 +16,21 @@ package io.micronaut.configuration.metrics.micrometer.statsd; import io.micrometer.core.instrument.composite.CompositeMeterRegistry; +import io.micrometer.statsd.StatsdConfig; import io.micrometer.statsd.StatsdMeterRegistry; import io.micronaut.configuration.metrics.micrometer.ExportConfigurationProperties; +import io.micronaut.context.annotation.Bean; import io.micronaut.context.annotation.Factory; import io.micronaut.context.annotation.Requires; +import io.micronaut.core.annotation.TypeHint; +import jakarta.annotation.Nullable; import jakarta.inject.Singleton; import java.util.Properties; +import static io.micronaut.core.annotation.TypeHint.AccessType.ALL_DECLARED_FIELDS; +import static io.micronaut.core.annotation.TypeHint.AccessType.ALL_DECLARED_METHODS; +import static io.micronaut.core.annotation.TypeHint.AccessType.ALL_PUBLIC_CONSTRUCTORS; import static io.micrometer.core.instrument.Clock.SYSTEM; import static io.micronaut.configuration.metrics.micrometer.MeterRegistryFactory.MICRONAUT_METRICS_ENABLED; import static io.micronaut.configuration.metrics.micrometer.MeterRegistryFactory.MICRONAUT_METRICS_EXPORT; @@ -33,7 +40,24 @@ * Creates a StatsD meter registry. */ @Factory -public class StatsdMeterRegistryFactory { +@TypeHint( + typeNames = { + "io.micrometer.shaded.io.netty.buffer.AbstractByteBufAllocator", + "io.micrometer.shaded.io.netty.util.ReferenceCountUtil", + "io.micrometer.shaded.io.netty.util.internal.shaded.org.jctools.queues.BaseMpscLinkedArrayQueueColdProducerFields", + "io.micrometer.shaded.io.netty.util.internal.shaded.org.jctools.queues.BaseMpscLinkedArrayQueueConsumerFields", + "io.micrometer.shaded.io.netty.util.internal.shaded.org.jctools.queues.BaseMpscLinkedArrayQueueProducerFields", + "io.micrometer.shaded.io.netty.util.internal.shaded.org.jctools.queues.MpscArrayQueueConsumerIndexField", + "io.micrometer.shaded.io.netty.util.internal.shaded.org.jctools.queues.MpscArrayQueueProducerIndexField", + "io.micrometer.shaded.io.netty.util.internal.shaded.org.jctools.queues.MpscArrayQueueProducerLimitField" + }, + accessType = { + ALL_PUBLIC_CONSTRUCTORS, + ALL_DECLARED_METHODS, + ALL_DECLARED_FIELDS + } +) +public final class StatsdMeterRegistryFactory { public static final String STATSD_CONFIG = MICRONAUT_METRICS_EXPORT + ".statsd"; public static final String STATSD_ENABLED = STATSD_CONFIG + ".enabled"; @@ -46,10 +70,29 @@ public class StatsdMeterRegistryFactory { * @return StatsdMeterRegistry */ @Singleton + @Bean(preDestroy = "close") + @Requires(property = MICRONAUT_METRICS_ENABLED, notEquals = FALSE) + @Requires(beans = CompositeMeterRegistry.class) + StatsdMeterRegistry statsdMeterRegistry(ExportConfigurationProperties exportConfigurationProperties, + @Nullable NativeImageUdpStatsdLineSink nativeImageUdpStatsdLineSink) { + Properties exportConfig = exportConfigurationProperties.getExport(); + StatsdConfig statsdConfig = exportConfig::getProperty; + if (nativeImageUdpStatsdLineSink != null) { + return StatsdMeterRegistry.builder(statsdConfig) + .lineSink(nativeImageUdpStatsdLineSink) + .clock(SYSTEM) + .build(); + } + return new StatsdMeterRegistry(statsdConfig, SYSTEM); + } + + @Singleton + @Bean(preDestroy = "close") @Requires(property = MICRONAUT_METRICS_ENABLED, notEquals = FALSE) @Requires(beans = CompositeMeterRegistry.class) - StatsdMeterRegistry statsdMeterRegistry(ExportConfigurationProperties exportConfigurationProperties) { + @Requires(condition = NativeImageStatsdCondition.class) + NativeImageUdpStatsdLineSink nativeImageUdpStatsdLineSink(ExportConfigurationProperties exportConfigurationProperties) { Properties exportConfig = exportConfigurationProperties.getExport(); - return new StatsdMeterRegistry(exportConfig::getProperty, SYSTEM); + return new NativeImageUdpStatsdLineSink(exportConfig::getProperty); } } diff --git a/micrometer-registry-statsd/src/main/resources/META-INF/native-image/io.micronaut.micrometer/micronaut-micrometer-registry-statsd/native-image.properties b/micrometer-registry-statsd/src/main/resources/META-INF/native-image/io.micronaut.micrometer/micronaut-micrometer-registry-statsd/native-image.properties new file mode 100644 index 000000000..f550de9e1 --- /dev/null +++ b/micrometer-registry-statsd/src/main/resources/META-INF/native-image/io.micronaut.micrometer/micronaut-micrometer-registry-statsd/native-image.properties @@ -0,0 +1,14 @@ +Args = --initialize-at-run-time=io.micrometer.shaded.io.netty.buffer.PooledByteBufAllocator,io.micrometer.shaded.io.netty.buffer.ByteBufAllocator,io.micrometer.shaded.io.netty.buffer.ByteBufUtil,io.micrometer.shaded.io.netty.buffer.AbstractReferenceCountedByteBuf \ + --initialize-at-run-time=io.micrometer.shaded.io.netty.util.AbstractReferenceCounted,io.micrometer.shaded.io.netty.util.concurrent.GlobalEventExecutor,io.micrometer.shaded.io.netty.util.concurrent.ImmediateEventExecutor,io.micrometer.shaded.io.netty.util.concurrent.ScheduledFutureTask,io.micrometer.shaded.io.netty.util.internal.ThreadLocalRandom \ + --initialize-at-run-time=io.micrometer.shaded.io.netty.util.NetUtilSubstitutions$NetUtilLocalhost4LazyHolder \ + --initialize-at-run-time=io.micrometer.shaded.io.netty.util.NetUtilSubstitutions$NetUtilLocalhost6LazyHolder \ + --initialize-at-run-time=io.micrometer.shaded.io.netty.util.NetUtilSubstitutions$NetUtilLocalhostLazyHolder \ + --initialize-at-run-time=io.micrometer.shaded.io.netty.util.NetUtilSubstitutions$NetUtilNetworkInterfacesLazyHolder \ + --initialize-at-run-time=io.micrometer.shaded.io.netty.handler.ssl.util.ThreadLocalInsecureRandom \ + --initialize-at-run-time=io.micrometer.shaded.io.netty.resolver.dns.DefaultDnsServerAddressStreamProvider \ + --initialize-at-run-time=io.micrometer.shaded.io.netty.resolver.dns.DnsServerAddressStreamProviders$DefaultProviderHolder \ + --initialize-at-run-time=io.micrometer.shaded.io.netty.resolver.dns.DnsNameResolver \ + --initialize-at-run-time=io.micrometer.shaded.io.netty.resolver.dns.DnsNameResolverBuilder \ + --initialize-at-run-time=io.micrometer.shaded.io.netty.resolver.HostsFileEntriesResolver \ + --initialize-at-run-time=io.micrometer.shaded.io.netty.resolver.dns.ResolvConf$ResolvConfLazy \ + --initialize-at-run-time=io.micrometer.shaded.io.netty.channel.epoll,io.micrometer.shaded.io.netty.channel.unix.Limits,io.micrometer.shaded.io.netty.channel.unix.IovArray,io.micrometer.shaded.io.netty.channel.unix.Errors diff --git a/micrometer-registry-statsd/src/test/groovy/io/micronaut/configuration/metrics/micrometer/statsd/NativeImageUdpStatsdLineSinkSpec.groovy b/micrometer-registry-statsd/src/test/groovy/io/micronaut/configuration/metrics/micrometer/statsd/NativeImageUdpStatsdLineSinkSpec.groovy new file mode 100644 index 000000000..0706c2098 --- /dev/null +++ b/micrometer-registry-statsd/src/test/groovy/io/micronaut/configuration/metrics/micrometer/statsd/NativeImageUdpStatsdLineSinkSpec.groovy @@ -0,0 +1,250 @@ +package io.micronaut.configuration.metrics.micrometer.statsd + +import io.micrometer.statsd.StatsdConfig +import io.micronaut.context.ApplicationContext +import spock.lang.Specification +import spock.lang.Unroll + +import java.util.concurrent.CountDownLatch +import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicInteger + +import static io.micronaut.configuration.metrics.micrometer.MeterRegistryFactory.MICRONAUT_METRICS_ENABLED + +class NativeImageUdpStatsdLineSinkSpec extends Specification { + + private static final String IMAGE_CODE_PROPERTY = NativeImageStatsdCondition.NATIVE_IMAGE_CODE_PROPERTY + + @Unroll + void "verify native image line sink bean presence = #present for imageCode=#imageCode protocol=#protocol"() { + given: + if (imageCode != null) { + System.setProperty(IMAGE_CODE_PROPERTY, imageCode) + } else { + System.clearProperty(IMAGE_CODE_PROPERTY) + } + ApplicationContext context = ApplicationContext.run([ + (StatsdMeterRegistryFactory.STATSD_ENABLED): true, + (StatsdMeterRegistryFactory.STATSD_CONFIG + ".protocol"): protocol + ]) + + expect: + context.findBean(NativeImageUdpStatsdLineSink).present == present + + cleanup: + context.close() + System.clearProperty(IMAGE_CODE_PROPERTY) + + where: + imageCode | protocol | present + "runtime" | "udp" | true + "runtime" | "tcp" | false + "buildtime" | "udp" | false + null | "udp" | false + } + + void "verify native image line sink sends buffered payloads over udp"() { + given: + List payloads = [] + NativeImageUdpStatsdLineSink sink = new NativeImageUdpStatsdLineSink(config( + host: "127.0.0.1", + port: "8125", + buffered: "true", + pollingFrequency: "PT10S", + maxPacketLength: "256" + ), payloads.&add) + + when: + sink.accept("first:1|c") + sink.accept("second:2|c") + sink.close() + + then: + payloads == ["first:1|c\nsecond:2|c"] + + cleanup: + sink?.close() + } + + void "verify native image line sink buffers according to utf8 payload bytes"() { + given: + List payloads = [] + NativeImageUdpStatsdLineSink sink = new NativeImageUdpStatsdLineSink(config( + host: "127.0.0.1", + port: "8125", + buffered: "true", + pollingFrequency: "PT10S", + maxPacketLength: "5" + ), payloads.&add) + + when: + sink.accept("aé") + sink.accept("bé") + sink.close() + + then: + payloads == ["aé", "bé"] + + cleanup: + sink?.close() + } + + void "verify native image line sink does not hold lock while sending"() { + given: + CountDownLatch firstSendStarted = new CountDownLatch(1) + CountDownLatch allowFirstSendToFinish = new CountDownLatch(1) + CountDownLatch secondSendStarted = new CountDownLatch(1) + AtomicInteger sendCount = new AtomicInteger() + Thread first = null + Thread second = null + NativeImageUdpStatsdLineSink sink = new NativeImageUdpStatsdLineSink(config( + host: "127.0.0.1", + port: "8125", + buffered: "false", + pollingFrequency: "PT10S", + maxPacketLength: "256" + ), { String payload -> + int currentSend = sendCount.incrementAndGet() + if (currentSend == 1) { + firstSendStarted.countDown() + allowFirstSendToFinish.await(5, TimeUnit.SECONDS) + } else if (currentSend == 2) { + secondSendStarted.countDown() + } + }) + first = Thread.start { + sink.accept("first:1|c") + } + + expect: + firstSendStarted.await(5, TimeUnit.SECONDS) + + when: + second = Thread.start { + sink.accept("second:2|c") + } + + then: + secondSendStarted.await(5, TimeUnit.SECONDS) + + cleanup: + allowFirstSendToFinish?.countDown() + first?.join(5_000) + second?.join(5_000) + sink?.close() + } + + void "verify native image line sink tolerates non positive polling frequency"() { + given: + List payloads = [] + NativeImageUdpStatsdLineSink sink = new NativeImageUdpStatsdLineSink(config( + host: "127.0.0.1", + port: "8125", + buffered: "true", + pollingFrequency: "PT0S", + maxPacketLength: "256" + ), payloads.&add) + + when: + sink.accept("first:1|c") + sink.close() + + then: + payloads == ["first:1|c"] + + cleanup: + sink?.close() + } + + void "verify native image line sink keeps sending after listener starts later"() { + given: + DatagramSocket reservation = new DatagramSocket(0) + int port = reservation.localPort + reservation.close() + NativeImageUdpStatsdLineSink sink = new NativeImageUdpStatsdLineSink(config( + host: "127.0.0.1", + port: port.toString(), + buffered: "false", + pollingFrequency: "PT0.05S", + maxPacketLength: "256" + )) + + when: + sink.accept("dropped:1|c") + DatagramSocket server = new DatagramSocket(port) + server.soTimeout = 5_000 + sink.accept("recovered:2|c") + String packet = receive(server) + + then: + packet == "recovered:2|c" + + cleanup: + sink?.close() + server?.close() + } + + void "verify native image line sink reuses udp sender socket"() { + given: + DatagramSocket server = new DatagramSocket(0) + server.soTimeout = 5_000 + NativeImageUdpStatsdLineSink sink = new NativeImageUdpStatsdLineSink(config( + host: "127.0.0.1", + port: server.localPort.toString(), + buffered: "false", + pollingFrequency: "PT0S", + maxPacketLength: "256" + )) + + when: + sink.accept("first:1|c") + DatagramPacket first = receivePacket(server) + sink.accept("second:2|c") + DatagramPacket second = receivePacket(server) + + then: + new String(first.data, first.offset, first.length) == "first:1|c" + new String(second.data, second.offset, second.length) == "second:2|c" + first.port == second.port + + cleanup: + sink?.close() + server?.close() + } + + void "verify native image line sink bean absent when metrics are disabled"() { + given: + System.setProperty(IMAGE_CODE_PROPERTY, "runtime") + ApplicationContext context = ApplicationContext.run([ + (MICRONAUT_METRICS_ENABLED) : false, + (StatsdMeterRegistryFactory.STATSD_ENABLED): true, + (StatsdMeterRegistryFactory.STATSD_CONFIG + ".protocol"): "udp" + ]) + + expect: + !context.findBean(NativeImageUdpStatsdLineSink).present + + cleanup: + context.close() + System.clearProperty(IMAGE_CODE_PROPERTY) + } + + private static StatsdConfig config(Map properties) { + Map prefixedProperties = properties.collectEntries { String key, String value -> + [((key.startsWith('statsd.') ? key : "statsd.${key}").toString()): value] + } + return { String key -> properties.get(key) ?: prefixedProperties.get(key) } as StatsdConfig + } + + private static String receive(DatagramSocket socket) { + DatagramPacket packet = receivePacket(socket) + return new String(packet.data, packet.offset, packet.length) + } + + private static DatagramPacket receivePacket(DatagramSocket socket) { + byte[] buffer = new byte[512] + DatagramPacket packet = new DatagramPacket(buffer, buffer.length) + socket.receive(packet) + return packet + } +} diff --git a/micrometer-registry-statsd/src/test/groovy/io/micronaut/configuration/metrics/micrometer/statsd/StatsdNativeImageMetadataSpec.groovy b/micrometer-registry-statsd/src/test/groovy/io/micronaut/configuration/metrics/micrometer/statsd/StatsdNativeImageMetadataSpec.groovy new file mode 100644 index 000000000..4923e9a1a --- /dev/null +++ b/micrometer-registry-statsd/src/test/groovy/io/micronaut/configuration/metrics/micrometer/statsd/StatsdNativeImageMetadataSpec.groovy @@ -0,0 +1,41 @@ +package io.micronaut.configuration.metrics.micrometer.statsd + +import io.micronaut.core.annotation.TypeHint +import spock.lang.Specification + +class StatsdNativeImageMetadataSpec extends Specification { + + void "verify shaded native-image properties are packaged for statsd"() { + given: + String path = 'META-INF/native-image/io.micronaut.micrometer/micronaut-micrometer-registry-statsd/native-image.properties' + + when: + InputStream stream = StatsdMeterRegistryFactory.classLoader.getResourceAsStream(path) + + then: + stream != null + + when: + String contents = stream.getText('UTF-8') + + then: + contents.contains('io.micrometer.shaded.io.netty.buffer.PooledByteBufAllocator') + contents.contains('io.micrometer.shaded.io.netty.resolver.dns.DnsNameResolver') + contents.contains('io.micrometer.shaded.io.netty.channel.epoll') + !contents.contains('--initialize-at-run-time=io.netty.') + + cleanup: + stream?.close() + } + + void "verify statsd factory declares shaded netty reflection hints"() { + when: + TypeHint hint = StatsdMeterRegistryFactory.getAnnotation(TypeHint) + + then: + hint != null + hint.typeNames().contains('io.micrometer.shaded.io.netty.buffer.AbstractByteBufAllocator') + hint.typeNames().contains('io.micrometer.shaded.io.netty.util.ReferenceCountUtil') + hint.typeNames().contains('io.micrometer.shaded.io.netty.util.internal.shaded.org.jctools.queues.MpscArrayQueueProducerLimitField') + } +}