SOLR-18060: CrossDC Consumer - add Prometheus metrics.

changelog/unreleased/solr-18060.yml

@@ -0,0 +1,9 @@
+# See https://github.com/apache/solr/blob/main/dev-docs/changelog.adoc
+title: SOLR-18060 CrossDC Consumer - add Prometheus metrics
+type: changed
+authors:
+  - name: Andrzej Bialecki
+  - nick: ab
+links:
+  - name: SOLR-18060
+    url: https://issues.apache.org/jira/browse/SOLR-18060

gradle/libs.versions.toml

@@ -189,6 +189,7 @@ owasp-dependencycheck = "12.1.3"
 # @keep for version alignment
 perfmark = "0.27.0"
 prometheus-metrics = "1.1.0"
+prometheus-metrics14 = "1.4.3"
 prometheus-simpleclient = "0.16.0"
 quicktheories = "0.26"
 semver4j = "6.0.0"
@@ -500,6 +501,11 @@ ow2-asm-tree = { module = "org.ow2.asm:asm-tree", version.ref = "ow2-asm" }
 perfmark-api = { module = "io.perfmark:perfmark-api", version.ref = "perfmark" }
 prometheus-metrics-expositionformats = { module = "io.prometheus:prometheus-metrics-exposition-formats", version.ref = "prometheus-metrics" }
 prometheus-metrics-model = { module = "io.prometheus:prometheus-metrics-model", version.ref = "prometheus-metrics" }
+prometheus-metrics14-core = { module = "io.prometheus:prometheus-metrics-core", version.ref = "prometheus-metrics14" }
+prometheus-metrics14-exporter-servlet-jakarta = { module = "io.prometheus:prometheus-metrics-exporter-servlet-jakarta", version.ref = "prometheus-metrics14" }
+prometheus-metrics14-instrumentation-dropwizard = { module = "io.prometheus:prometheus-metrics-instrumentation-dropwizard", version.ref = "prometheus-metrics14" }
+prometheus-metrics14-instrumentation-dropwizard5 = { module = "io.prometheus:prometheus-metrics-instrumentation-dropwizard5", version.ref = "prometheus-metrics14" }
+prometheus-metrics14-model = { module = "io.prometheus:prometheus-metrics-model", version.ref = "prometheus-metrics14" }
 prometheus-simpleclient = { module = "io.prometheus:simpleclient", version.ref = "prometheus-simpleclient" }
 prometheus-simpleclient-httpserver = { module = "io.prometheus:simpleclient_httpserver", version.ref = "prometheus-simpleclient" }
 quicktheories-quicktheories = { module = "org.quicktheories:quicktheories", version.ref = "quicktheories" }

solr/api/gradle.lockfile

@@ -69,8 +69,9 @@ io.opentelemetry:opentelemetry-sdk-common:1.56.0=jarValidation,testRuntimeClassp
 io.opentelemetry:opentelemetry-sdk-metrics:1.56.0=jarValidation,testRuntimeClasspath
 io.opentelemetry:opentelemetry-sdk-trace:1.56.0=jarValidation,testRuntimeClasspath
 io.opentelemetry:opentelemetry-sdk:1.56.0=jarValidation,testRuntimeClasspath
+io.prometheus:prometheus-metrics-config:1.4.3=jarValidation,testRuntimeClasspath
 io.prometheus:prometheus-metrics-exposition-formats:1.1.0=jarValidation,testRuntimeClasspath
-io.prometheus:prometheus-metrics-model:1.1.0=jarValidation,testRuntimeClasspath
+io.prometheus:prometheus-metrics-model:1.4.3=jarValidation,testRuntimeClasspath
 io.sgr:s2-geometry-library-java:1.0.0=jarValidation,testRuntimeClasspath
 io.swagger.core.v3:swagger-annotations-jakarta:2.2.22=apiHelper,compileClasspath,jarValidation,runtimeClasspath,swaggerBuild,testCompileClasspath,testRuntimeClasspath
 io.swagger.core.v3:swagger-core-jakarta:2.2.22=swaggerBuild

solr/benchmark/gradle.lockfile

@@ -64,8 +64,9 @@ io.opentelemetry:opentelemetry-sdk-common:1.56.0=jarValidation,runtimeClasspath,
 io.opentelemetry:opentelemetry-sdk-metrics:1.56.0=jarValidation,runtimeClasspath,testRuntimeClasspath
 io.opentelemetry:opentelemetry-sdk-trace:1.56.0=jarValidation,runtimeClasspath,testRuntimeClasspath
 io.opentelemetry:opentelemetry-sdk:1.56.0=jarValidation,runtimeClasspath,testRuntimeClasspath
+io.prometheus:prometheus-metrics-config:1.4.3=jarValidation,runtimeClasspath,testRuntimeClasspath
 io.prometheus:prometheus-metrics-exposition-formats:1.1.0=jarValidation,runtimeClasspath,testRuntimeClasspath
-io.prometheus:prometheus-metrics-model:1.1.0=jarValidation,runtimeClasspath,testRuntimeClasspath
+io.prometheus:prometheus-metrics-model:1.4.3=jarValidation,runtimeClasspath,testRuntimeClasspath
 io.sgr:s2-geometry-library-java:1.0.0=jarValidation,runtimeClasspath,testRuntimeClasspath
 io.swagger.core.v3:swagger-annotations-jakarta:2.2.22=compileClasspath,jarValidation,runtimeClasspath,testCompileClasspath,testRuntimeClasspath
 jakarta.annotation:jakarta.annotation-api:2.1.1=jarValidation,runtimeClasspath,testRuntimeClasspath

solr/core/gradle.lockfile

@@ -71,8 +71,9 @@ io.opentelemetry:opentelemetry-sdk-metrics:1.56.0=compileClasspath,jarValidation
 io.opentelemetry:opentelemetry-sdk-testing:1.56.0=jarValidation,testCompileClasspath,testRuntimeClasspath
 io.opentelemetry:opentelemetry-sdk-trace:1.56.0=compileClasspath,jarValidation,runtimeClasspath,runtimeLibs,testCompileClasspath,testRuntimeClasspath
 io.opentelemetry:opentelemetry-sdk:1.56.0=compileClasspath,jarValidation,runtimeClasspath,runtimeLibs,testCompileClasspath,testRuntimeClasspath
+io.prometheus:prometheus-metrics-config:1.4.3=compileClasspath,jarValidation,runtimeClasspath,runtimeLibs,testCompileClasspath,testRuntimeClasspath
 io.prometheus:prometheus-metrics-exposition-formats:1.1.0=compileClasspath,jarValidation,runtimeClasspath,runtimeLibs,testCompileClasspath,testRuntimeClasspath
-io.prometheus:prometheus-metrics-model:1.1.0=compileClasspath,jarValidation,runtimeClasspath,runtimeLibs,testCompileClasspath,testRuntimeClasspath
+io.prometheus:prometheus-metrics-model:1.4.3=compileClasspath,jarValidation,runtimeClasspath,runtimeLibs,testCompileClasspath,testRuntimeClasspath
 io.sgr:s2-geometry-library-java:1.0.0=compileClasspath,jarValidation,runtimeClasspath,runtimeLibs,testCompileClasspath,testRuntimeClasspath
 io.swagger.core.v3:swagger-annotations-jakarta:2.2.22=apiHelper,compileClasspath,jarValidation,runtimeClasspath,runtimeLibs,testCompileClasspath,testRuntimeClasspath
 jakarta.activation:jakarta.activation-api:2.1.3=permitUnusedDeclared

solr/cross-dc-manager/build.gradle

@@ -30,6 +30,20 @@ dependencies {
 
   implementation libs.dropwizard.metrics.core
   implementation libs.dropwizard.metrics.servlets
+  implementation libs.prometheus.metrics14.model
+  implementation libs.prometheus.metrics14.exporter.servlet.jakarta
+  implementation (libs.prometheus.metrics14.instrumentation.dropwizard, {
+    exclude group: "io.prometheus", module: "prometheus-metrics-tracer-common"
+    exclude group: "io.prometheus", module: "prometheus-metrics-tracer-initializer"
+    exclude group: "io.prometheus", module: "prometheus-metrics-tracer-otel"
+    exclude group: "io.prometheus", module: "prometheus-metrics-tracer-otel-agent"
+  })
+  implementation (libs.prometheus.metrics14.instrumentation.dropwizard5, {
+    exclude group: "io.prometheus", module: "prometheus-metrics-tracer-common"
+    exclude group: "io.prometheus", module: "prometheus-metrics-tracer-initializer"
+    exclude group: "io.prometheus", module: "prometheus-metrics-tracer-otel"
+    exclude group: "io.prometheus", module: "prometheus-metrics-tracer-otel-agent"
+  })
   implementation libs.eclipse.jetty.server
   implementation libs.eclipse.jetty.ee10.servlet
   implementation libs.slf4j.api
@@ -48,6 +62,7 @@ dependencies {
   testImplementation project(':solr:test-framework')
   testImplementation libs.apache.lucene.testframework
   testImplementation libs.carrotsearch.randomizedtesting.runner
+  testImplementation libs.commonsio.commonsio
   testImplementation libs.junit.junit
   // The explicit dependency on bytebuddy is required for Java 25 support
   // Once Mockito upgrades its dependency on ByteBuddy to 1.16.1, we should

solr/cross-dc-manager/gradle.lockfile

@@ -82,8 +82,16 @@ io.opentelemetry:opentelemetry-sdk-common:1.56.0=jarValidation,runtimeClasspath,
 io.opentelemetry:opentelemetry-sdk-metrics:1.56.0=jarValidation,runtimeClasspath,runtimeLibs,solrPlatformLibs,testRuntimeClasspath
 io.opentelemetry:opentelemetry-sdk-trace:1.56.0=jarValidation,runtimeClasspath,runtimeLibs,solrPlatformLibs,testRuntimeClasspath
 io.opentelemetry:opentelemetry-sdk:1.56.0=jarValidation,runtimeClasspath,runtimeLibs,solrPlatformLibs,testRuntimeClasspath
-io.prometheus:prometheus-metrics-exposition-formats:1.1.0=jarValidation,runtimeClasspath,runtimeLibs,solrPlatformLibs,testRuntimeClasspath
-io.prometheus:prometheus-metrics-model:1.1.0=jarValidation,runtimeClasspath,runtimeLibs,solrPlatformLibs,testRuntimeClasspath
+io.prometheus:prometheus-metrics-config:1.4.3=compileClasspath,jarValidation,runtimeClasspath,runtimeLibs,solrPlatformLibs,testCompileClasspath,testRuntimeClasspath
+io.prometheus:prometheus-metrics-core:1.4.3=compileClasspath,jarValidation,runtimeClasspath,runtimeLibs,testCompileClasspath,testRuntimeClasspath
+io.prometheus:prometheus-metrics-exporter-common:1.4.3=compileClasspath,jarValidation,runtimeClasspath,runtimeLibs,testCompileClasspath,testRuntimeClasspath
+io.prometheus:prometheus-metrics-exporter-servlet-jakarta:1.4.3=compileClasspath,jarValidation,runtimeClasspath,runtimeLibs,testCompileClasspath,testRuntimeClasspath
+io.prometheus:prometheus-metrics-exposition-formats:1.1.0=solrPlatformLibs
+io.prometheus:prometheus-metrics-exposition-formats:1.4.3=jarValidation,runtimeClasspath,runtimeLibs,testRuntimeClasspath
+io.prometheus:prometheus-metrics-exposition-textformats:1.4.3=compileClasspath,jarValidation,runtimeClasspath,runtimeLibs,testCompileClasspath,testRuntimeClasspath
+io.prometheus:prometheus-metrics-instrumentation-dropwizard5:1.4.3=compileClasspath,jarValidation,runtimeClasspath,runtimeLibs,testCompileClasspath,testRuntimeClasspath
+io.prometheus:prometheus-metrics-instrumentation-dropwizard:1.4.3=compileClasspath,jarValidation,runtimeClasspath,runtimeLibs,testCompileClasspath,testRuntimeClasspath
+io.prometheus:prometheus-metrics-model:1.4.3=compileClasspath,jarValidation,runtimeClasspath,runtimeLibs,solrPlatformLibs,testCompileClasspath,testRuntimeClasspath
 io.sgr:s2-geometry-library-java:1.0.0=jarValidation,runtimeClasspath,runtimeLibs,solrPlatformLibs,testRuntimeClasspath
 io.swagger.core.v3:swagger-annotations-jakarta:2.2.22=compileClasspath,jarValidation,runtimeClasspath,runtimeLibs,solrPlatformLibs,testCompileClasspath,testRuntimeClasspath
 jakarta.annotation:jakarta.annotation-api:2.1.1=jarValidation,runtimeClasspath,runtimeLibs,solrPlatformLibs,testRuntimeClasspath

solr/cross-dc-manager/src/java/org/apache/solr/crossdc/manager/consumer/Consumer.java

@@ -20,11 +20,18 @@
 import static org.apache.solr.crossdc.common.KafkaCrossDcConf.TOPIC_NAME;
 import static org.apache.solr.crossdc.common.KafkaCrossDcConf.ZK_CONNECT_STRING;
 
+import com.codahale.metrics.MetricFilter;
 import com.codahale.metrics.SharedMetricRegistries;
 import io.dropwizard.metrics.servlets.MetricsServlet;
 import io.dropwizard.metrics.servlets.ThreadDumpServlet;
+import io.prometheus.metrics.exporter.servlet.jakarta.PrometheusMetricsServlet;
+import io.prometheus.metrics.instrumentation.dropwizard.DropwizardExports;
+import io.prometheus.metrics.instrumentation.dropwizard5.labels.CustomLabelMapper;
+import io.prometheus.metrics.instrumentation.dropwizard5.labels.MapperConfig;
+import io.prometheus.metrics.model.registry.PrometheusRegistry;
 import java.lang.invoke.MethodHandles;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
@@ -98,11 +105,23 @@ public void start(Map<String, Object> properties) {
       ServletContextHandler context = new ServletContextHandler(ServletContextHandler.NO_SESSIONS);
       context.setContextPath("/");
       server.setHandler(context);
+
       context.addServlet(ThreadDumpServlet.class, "/threads/*");
-      context.addServlet(MetricsServlet.class, "/metrics/*");
+
+      MapperConfig mapperConfig = new MapperConfig("*.*", "crossdc_consumer_${0}_${1}", Map.of());
+      CustomLabelMapper customLabelMapper = new CustomLabelMapper(List.of(mapperConfig));
+      DropwizardExports dropwizardExports =
+          new DropwizardExports(
+              SharedMetricRegistries.getOrCreate(METRICS_REGISTRY),
+              MetricFilter.ALL,
+              customLabelMapper);
+      PrometheusRegistry.defaultRegistry.register(dropwizardExports);
+      context.addServlet(PrometheusMetricsServlet.class, "/metrics/prometheus/*");
+
+      context.addServlet(MetricsServlet.class, "/metrics/json/*");
       context.setAttribute(
-          "com.codahale.metrics.servlets.MetricsServlet.registry",
-          SharedMetricRegistries.getOrCreate(METRICS_REGISTRY));
+          MetricsServlet.METRICS_REGISTRY, SharedMetricRegistries.getOrCreate(METRICS_REGISTRY));
+
       for (ServletMapping mapping : context.getServletHandler().getServletMappings()) {
         if (log.isInfoEnabled()) {
           log.info(" - {}", mapping.getPathSpecs()[0]);

solr/cross-dc-manager/src/test/org/apache/solr/crossdc/manager/SolrAndKafkaIntegrationTest.java

@@ -18,20 +18,22 @@
 
 import static org.apache.solr.crossdc.common.KafkaCrossDcConf.DEFAULT_MAX_REQUEST_SIZE;
 import static org.apache.solr.crossdc.common.KafkaCrossDcConf.INDEX_UNMIRRORABLE_DOCS;
-import static org.apache.solr.crossdc.common.KafkaCrossDcConf.PORT;
 
 import com.carrotsearch.randomizedtesting.annotations.ThreadLeakFilters;
 import com.carrotsearch.randomizedtesting.annotations.ThreadLeakLingering;
+import java.io.InputStream;
 import java.lang.invoke.MethodHandles;
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
+import java.util.Date;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 import java.util.Properties;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
+import org.apache.commons.io.IOUtils;
 import org.apache.kafka.clients.producer.KafkaProducer;
 import org.apache.kafka.clients.producer.Producer;
 import org.apache.kafka.clients.producer.ProducerRecord;
@@ -40,15 +42,20 @@
 import org.apache.lucene.tests.util.QuickPatchThreadsFilter;
 import org.apache.solr.SolrIgnoredThreadsFilter;
 import org.apache.solr.client.solrj.SolrClient;
+import org.apache.solr.client.solrj.SolrRequest;
 import org.apache.solr.client.solrj.impl.CloudSolrClient;
+import org.apache.solr.client.solrj.jetty.HttpJettySolrClient;
 import org.apache.solr.client.solrj.request.CollectionAdminRequest;
+import org.apache.solr.client.solrj.request.GenericSolrRequest;
 import org.apache.solr.client.solrj.request.SolrQuery;
 import org.apache.solr.client.solrj.request.UpdateRequest;
+import org.apache.solr.client.solrj.response.InputStreamResponseParser;
 import org.apache.solr.client.solrj.response.QueryResponse;
 import org.apache.solr.cloud.MiniSolrCloudCluster;
 import org.apache.solr.cloud.SolrCloudTestCase;
 import org.apache.solr.common.SolrInputDocument;
 import org.apache.solr.common.util.ExecutorUtil;
+import org.apache.solr.common.util.NamedList;
 import org.apache.solr.common.util.ObjectReleaseTracker;
 import org.apache.solr.common.util.SolrNamedThreadFactory;
 import org.apache.solr.crossdc.common.KafkaCrossDcConf;
@@ -61,6 +68,7 @@
 import org.junit.BeforeClass;
 import org.junit.Ignore;
 import org.junit.Test;
+import org.noggit.ObjectBuilder;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -101,7 +109,7 @@ public void beforeSolrAndKafkaIntegrationTest() throws Exception {
     uceh = Thread.getDefaultUncaughtExceptionHandler();
     Thread.setDefaultUncaughtExceptionHandler(
         (t, e) -> log.error("Uncaught exception in thread {}", t, e));
-    System.setProperty(PORT, "-1");
+    //    System.setProperty(PORT, "-1");
     consumer = new Consumer();
     Properties config = new Properties();
 
@@ -175,6 +183,51 @@ public void afterSolrAndKafkaIntegrationTest() throws Exception {
     Thread.setDefaultUncaughtExceptionHandler(uceh);
   }
 
+  @Test
+  @SuppressWarnings({"unchecked"})
+  public void testMetrics() throws Exception {
+    CloudSolrClient client = solrCluster1.getSolrClient();
+    SolrInputDocument doc = new SolrInputDocument();
+    doc.addField("id", String.valueOf(new Date().getTime()));
+    doc.addField("text", "some test");
+
+    client.add(COLLECTION, doc);
+
+    client.commit(COLLECTION);
+
+    System.out.println("Sent producer record");
+
+    assertCluster2EventuallyHasDocs(COLLECTION, "*:*", 1);
+
+    String baseUrl = "http://localhost:" + KafkaCrossDcConf.DEFAULT_PORT;
+    HttpJettySolrClient httpJettySolrClient =
+        new HttpJettySolrClient.Builder(baseUrl).useHttp1_1(true).build();
+    try {
+      // first try the JSON format
+      GenericSolrRequest req = new GenericSolrRequest(SolrRequest.METHOD.GET, "/metrics/json");
+      req.setResponseParser(new InputStreamResponseParser("test/plain"));
+      NamedList<Object> rsp = httpJettySolrClient.request(req);
+      String content =
+          IOUtils.toString(
+              (InputStream) rsp.get(InputStreamResponseParser.STREAM_KEY), StandardCharsets.UTF_8);
+      Map<String, Object> map = (Map<String, Object>) ObjectBuilder.fromJSON(content);
+      assertTrue(map.containsKey("counters"));
+      assertFalse(((Map<String, Object>) map.get("counters")).isEmpty());
+
+      // then try the prometheus format
+      req = new GenericSolrRequest(SolrRequest.METHOD.GET, "/metrics/prometheus");
+      req.setResponseParser(new InputStreamResponseParser("test/plain"));
+      rsp = httpJettySolrClient.request(req);
+      content =
+          IOUtils.toString(
+              (InputStream) rsp.get(InputStreamResponseParser.STREAM_KEY), StandardCharsets.UTF_8);
+      assertTrue(content.contains("UPDATE"));
+    } finally {
+      httpJettySolrClient.close();
+      client.close();
+    }
+  }
+
   public void testFullCloudToCloud() throws Exception {
     CloudSolrClient client = solrCluster1.getSolrClient(COLLECTION);
     SolrInputDocument doc = new SolrInputDocument();