Introduce new module reactor-core-micrometer (#3015)

The new module introduces an alternative more explicit way of bringing
metrics in reactor-core, both for Schedulers and for Flux and Mono. The
entry point is the `Micrometer` class.

The following metrics-related features in core are superseded:
 - global configuration previously done in core's `Metrics`
 - scheduler `ExecutorService` instrumentation previously done in
 `Scheduler#enableMetrics`
 - Flux/Mono `metrics()` operators replaced by `tap` + the factory from
  the new `Micrometer.metrics()` method

As a consequence, these metrics-related classes and methods in core are
deprecated, to be removed in 3.6.0 at the earliest. Note that most
logic needed to be duplicated or reproduced in the new module.

The new module lives in the core repository and shares the same groupId
as reactor-core (io.projectreactor). It is dedicated to Micrometer
instrumentation for reactor-core.

The new module brings an explicit dependency to Micrometer 1.10+, which
means that the old `Metrics.isInstrumentationAvailable` will always be
`true` when the module is on the classpath. Extra care should be taken
to avoid activating metrics via both the core features and their module
equivalents.

build.gradle

@@ -49,30 +49,36 @@ repositories { //needed at root for asciidoctor and nohttp-checkstyle
 	mavenCentral()
 }
 
-ext {
-	jdk = JavaVersion.current().majorVersion
-	jdkJavadoc = "https://docs.oracle.com/javase/$jdk/docs/api/"
-	if (JavaVersion.current().isJava11Compatible()) {
-		jdkJavadoc = "https://docs.oracle.com/en/java/javase/$jdk/docs/api/"
-	}
-	println "JDK Javadoc link for this build is ${rootProject.jdkJavadoc}"
-
-	versionNumber = VersionNumber.parse(version.toString())
+def osgiVersion(String v) {
+	def versionNumber = VersionNumber.parse(v)
+	def result
 	if (versionNumber.qualifier == null || versionNumber.qualifier.size() == 0) {
-		osgiVersion = "${version}.RELEASE"
-		println "$version is a release, will use $osgiVersion for bnd"
+		result = "${v}.RELEASE"
+		println "$v is a release, will use $result for bnd"
 	}
 	else if (versionNumber.qualifier.equalsIgnoreCase("SNAPSHOT")) {
 		def sdf = new SimpleDateFormat("yyyyMMddHHmm");
 		sdf.setTimeZone(TimeZone.getTimeZone("UTC"));
 		def buildTimestamp = sdf.format(new Date())
-		osgiVersion = "${versionNumber.major}.${versionNumber.minor}.${versionNumber.micro}.BUILD-$buildTimestamp"
-		println "$version is a snapshot, will use $osgiVersion for bnd"
+		result = "${versionNumber.major}.${versionNumber.minor}.${versionNumber.micro}.BUILD-$buildTimestamp"
+		println "$v is a snapshot, will use $result for bnd"
 	}
 	else {
-		osgiVersion = "${versionNumber.major}.${versionNumber.minor}.${versionNumber.micro}.${versionNumber.qualifier}"
-		println "$version is neither release nor snapshot, will use $osgiVersion for bnd"
+		result = "${versionNumber.major}.${versionNumber.minor}.${versionNumber.micro}.${versionNumber.qualifier}"
+		println "$v is neither release nor snapshot, will use $result for bnd"
+	}
+	return result
+}
+
+ext {
+	jdk = JavaVersion.current().majorVersion
+	jdkJavadoc = "https://docs.oracle.com/javase/$jdk/docs/api/"
+	if (JavaVersion.current().isJava11Compatible()) {
+		jdkJavadoc = "https://docs.oracle.com/en/java/javase/$jdk/docs/api/"
 	}
+	println "JDK Javadoc link for this build is ${rootProject.jdkJavadoc}"
+
+	osgiVersion = osgiVersion(version.toString())
 
 	/*
 	 * Note that all dependencies and their versions are now defined in

gradle.properties

@@ -1,2 +1,3 @@
 version=3.5.0-SNAPSHOT
 bomVersion=2022.0.0-M1
+metricsMicrometerVersion=1.0.0-SNAPSHOT

reactor-core-micrometer/build.gradle

@@ -0,0 +1,85 @@
+/*
+ * Copyright (c) 2022 VMware Inc. or its affiliates, All Rights Reserved.
+ *
+ * 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.
+ */
+
+apply plugin: 'biz.aQute.bnd.builder'
+apply plugin: 'java-library'
+
+description = 'Reactor-Core Micrometer Metrics support'
+
+version = "$metricsMicrometerVersion"
+group = "io.projectreactor"
+
+ext {
+	def osgiVersion = osgiVersion("$metricsMicrometerVersion")
+
+	bndOptions = [
+		"Export-Package": [
+			"!*internal*",
+			"reactor.core.observability.micrometer*;version=$osgiVersion;-noimport:=true"
+		].join(","),
+		"Import-Package": [
+			"!javax.annotation",
+			"*"
+		].join(","),
+		"Bundle-Name" : "reactor-core-micrometer",
+		"Bundle-SymbolicName" : "io.projectreactor.reactor-core-micrometer",
+		"Bundle-Version" : "$osgiVersion"
+	]
+}
+
+dependencies {
+	api project(":reactor-core")
+	compileOnly libs.jsr305
+
+	implementation platform(libs.micrometer.bom)
+	api libs.micrometer.core
+
+	testImplementation platform(libs.junit.bom)
+	testImplementation "org.junit.jupiter:junit-jupiter-api"
+	testImplementation "org.junit.platform:junit-platform-launcher"
+	testImplementation "org.junit.jupiter:junit-jupiter-params"
+	testRuntimeOnly "org.junit.jupiter:junit-jupiter-engine"
+
+	testImplementation platform(libs.micrometer.bom)
+	testImplementation libs.micrometer.core
+
+	testImplementation(project(":reactor-test")) {
+		exclude module: 'reactor-core'
+	}
+	// Needs sourceSets.test.output because tests there use helpers like AutoDisposingRule etc.
+	testImplementation project(":reactor-core").sourceSets.test.output
+
+	testRuntimeOnly libs.logback
+	testImplementation libs.assertj
+	testImplementation libs.mockito
+}
+
+tasks.withType(Test).all {
+  useJUnitPlatform()
+}
+
+// javadoc is configured in gradle/javadoc.gradle
+
+jar {
+	manifest {
+		attributes 'Implementation-Title': 'reactor-core-micrometer',
+			'Implementation-Version': project.version,
+			'Automatic-Module-Name': 'reactor.core.micrometer'
+	}
+	bnd(bndOptions)
+}
+
+//TODO once 1.0.0 is released, introduce JAPICMP checks

reactor-core-micrometer/src/main/java/reactor/core/observability/micrometer/Micrometer.java

@@ -0,0 +1,133 @@
+/*
+ * Copyright (c) 2022 VMware Inc. or its affiliates, All Rights Reserved.
+ *
+ * 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 reactor.core.observability.micrometer;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ScheduledExecutorService;
+
+import io.micrometer.core.instrument.Clock;
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.Metrics;
+
+import reactor.core.observability.SignalListener;
+import reactor.core.scheduler.Scheduler;
+import reactor.core.scheduler.Schedulers;
+import reactor.core.observability.SignalListenerFactory;
+
+public final class Micrometer {
+
+	private static final String SCHEDULERS_DECORATOR_KEY = "reactor.core.observability.micrometer.schedulerDecorator";
+	private static MeterRegistry registry = Metrics.globalRegistry;
+
+	/**
+	 * The default "name" to use as a prefix for meter IDs if the instrumented sequence doesn't
+	 * define a {@link reactor.core.publisher.Flux#name(String) name}.
+	 */
+	public static final String DEFAULT_METER_PREFIX = "reactor";
+
+	/**
+	 * Set the registry to use in reactor for metrics related purposes.
+	 * @return the previously configured registry.
+	 */
+	public static MeterRegistry useRegistry(MeterRegistry newRegistry) {
+		MeterRegistry previous = registry;
+		registry = newRegistry;
+		return previous;
+	}
+
+	/**
+	 * Get the registry used in reactor for metrics related purposes.
+	 */
+	public static MeterRegistry getRegistry() {
+		return registry;
+	}
+
+	/**
+	 * A {@link SignalListener} factory that will ultimately produce Micrometer metrics
+	 * to the configured default {@link #getRegistry() registry}.
+	 * To be used with either the {@link reactor.core.publisher.Flux#tap(SignalListenerFactory)} or
+	 * {@link reactor.core.publisher.Mono#tap(SignalListenerFactory)} operator.
+	 * <p>
+	 * When used in a {@link reactor.core.publisher.Flux#tap(SignalListenerFactory)} operator, meter names use
+	 * the {@link reactor.core.publisher.Flux#name(String)} set upstream of the tap as id prefix if applicable
+	 * or default to {@link #DEFAULT_METER_PREFIX}. Similarly, upstream tags are gathered and added
+	 * to the default set of tags for meters.
+	 * <p>
+	 * Note that some monitoring systems like Prometheus require to have the exact same set of
+	 * tags for each meter bearing the same name.
+	 *
+	 * @param <T> the type of onNext in the target publisher
+	 * @return a {@link SignalListenerFactory} to record metrics
+	 */
+	public static <T> SignalListenerFactory<T, ?> metrics() {
+		return new MicrometerListenerFactory<>();
+	}
+
+	/**
+	 * A {@link SignalListener} factory that will ultimately produce Micrometer metrics
+	 * to the provided {@link MeterRegistry} using the provided {@link Clock} for timings.
+	 * To be used with either the {@link reactor.core.publisher.Flux#tap(SignalListenerFactory)} or
+	 * {@link reactor.core.publisher.Mono#tap(SignalListenerFactory)} operator.
+	 * <p>
+	 * When used in a {@link reactor.core.publisher.Flux#tap(SignalListenerFactory)} operator, meter names use
+	 * the {@link reactor.core.publisher.Flux#name(String)} set upstream of the tap as id prefix if applicable
+	 * or default to {@link #DEFAULT_METER_PREFIX}. Similarly, upstream tags are gathered and added
+	 * to the default set of tags for meters.
+	 * <p>
+	 * Note that some monitoring systems like Prometheus require to have the exact same set of
+	 * tags for each meter bearing the same name.
+	 *
+	 * @param <T> the type of onNext in the target publisher
+	 * @return a {@link SignalListenerFactory} to record metrics
+	 */
+	public static <T> SignalListenerFactory<T, ?> metrics(MeterRegistry registry, Clock clock) {
+		return new MicrometerListenerFactory<T>() {
+			@Override
+			protected Clock useClock() {
+				return clock;
+			}
+
+			@Override
+			protected MeterRegistry useRegistry() {
+				return registry;
+			}
+		};
+	}
+
+	/**
+	 * Set-up a decorator that will instrument any {@link ExecutorService} that backs a reactor-core {@link Scheduler}
+	 * (or scheduler implementations which use {@link Schedulers#decorateExecutorService(Scheduler, ScheduledExecutorService)}).
+	 * <p>
+	 * The {@link MeterRegistry} to use can be configured via {@link #useRegistry(MeterRegistry)}
+	 * prior to using this method, the default being {@link io.micrometer.core.instrument.Metrics#globalRegistry}.
+	 *
+	 * @implNote Note that this is added as a decorator via Schedulers when enabling metrics for schedulers,
+	 * which doesn't change the Factory.
+	 */
+	public static void enableSchedulersMetricsDecorator() {
+		Schedulers.addExecutorServiceDecorator(SCHEDULERS_DECORATOR_KEY,
+			new MicrometerSchedulerMetricsDecorator(getRegistry()));
+	}
+
+	/**
+	 * If {@link #enableSchedulersMetricsDecorator()} has been previously called, removes the decorator.
+	 * No-op if {@link #enableSchedulersMetricsDecorator()} hasn't been called.
+	 */
+	public static void disableSchedulersMetricsDecorator() {
+		Schedulers.removeExecutorServiceDecorator(SCHEDULERS_DECORATOR_KEY);
+	}
+}