Support Micrometer Observation to manage metrics and traces

README.md

@@ -423,6 +423,62 @@ Depends on amount changes in transaction there are 2 possible Commit strategies:
 3. It is safe to call `tx.rollback` after `tx.commit`.
 4. It is possible to call Query from transaction by call `tx.query().execute(); ...`. Only read-committed isolation is available. Changes made in active transaction is invisible to current and another transactions.
 
+### Observability support
+For metrics and traces, reindexer-java uses [Micrometer Observation](https://docs.micrometer.io/micrometer/reference/observation).
+To enable observation, you need to provide an `ObservationRegistry` to the `ReindexerConfiguration`.
+
+The following example shows how to configure observation for reindexer-java using Prometheus:
+
+Add `micrometer-registry-prometheus` dependency to the `pom.xml`:
+```xml
+<dependency>
+	<groupId>io.micrometer</groupId>
+	<artifactId>micrometer-registry-prometheus</artifactId>
+    <version>${micrometer.version}</version>
+</dependency>
+```
+
+Provide an `ObservationRegistry` implementation to the `ReindexerConfiguration`:
+```java
+// 1. Initialize Prometheus Meter Registry:
+PrometheusMeterRegistry prometheusRegistry = new PrometheusMeterRegistry(PrometheusConfig.DEFAULT);
+
+// 2. Initialize Observation Registry:
+ObservationRegistry observationRegistry = ObservationRegistry.create();
+
+// 3. Bridge them together using DefaultMeterObservationHandler:
+observationRegistry.observationConfig()
+        .observationHandler(new DefaultMeterObservationHandler(prometheusRegistry));
+
+// 4. Provide an ObservationRegistry to ReindexerConfiguration:
+Reindexer db = ReindexerConfiguration.builder()
+        .url("cproto://localhost:6534/testdb")
+        .connectionPoolSize(8)
+        .requestTimeout(Duration.ofSeconds(30L))
+        .observationRegistry(observationRegistry)
+        .getReindexer();
+```
+
+#### Collected metrics and traces
+All Reindexer RPC commands executed by reindexer-java are instrumented with Micrometer.
+
+The following low cardinality key values are added to observations:
+- `db.system.name` - the name of the database system, always `reindexer`
+- `db.command.name` - the name of the RPC command being executed, e.g., `selectQuery`
+- `db.namespace` - the database name e.g., `test_db`
+- `db.collection.name` - the collection name that the RPC command is executed on e.g., `items`
+- `network.transport` - the protocol used for the RPC command e.g., `cproto`, `cprotos`
+- `server.address` - the host of the Reindexer node that the RPC command is sent to e.g., `localhost`
+- `server.port` - the port of the Reindexer node that the RPC command is sent to e.g., `6534`
+- `code.execution_type` - the code execution type e.g., `SYNC`, `ASYNC`
+- `db.response.status_code` - the Reindexer response status code
+
+Additionally, the following high-cardinality key values are added to traces:
+- `thread.id` - ID of the thread executing the RPC command
+- `thread.name` - name of the thread executing the RPC command
+- `db.reindexer.tx_id` - ID of the Reindexer transaction associated with the RPC command, when applicable
+- `db.reindexer.rq_id` - ID of the Reindexer request associated with the RPC command
+
 ### Development notes
 
 To run tests locally, you need to install Reindexer using a package manager for your OS.

pom.xml

@@ -243,6 +243,11 @@
             <artifactId>commons-lang3</artifactId>
             <version>3.11</version>
         </dependency>
+        <dependency>
+            <groupId>io.micrometer</groupId>
+            <artifactId>micrometer-observation</artifactId>
+            <version>1.17.0</version>
+        </dependency>
         <dependency>
             <groupId>commons-io</groupId>
             <artifactId>commons-io</artifactId>
@@ -252,19 +257,19 @@
         <dependency>
             <groupId>org.junit.jupiter</groupId>
             <artifactId>junit-jupiter-api</artifactId>
-            <version>5.7.0</version>
+            <version>5.9.0</version>
             <scope>test</scope>
         </dependency>
         <dependency>
             <groupId>org.junit.jupiter</groupId>
             <artifactId>junit-jupiter-params</artifactId>
-            <version>5.7.0</version>
+            <version>5.9.0</version>
             <scope>test</scope>
         </dependency>
         <dependency>
             <groupId>org.junit.jupiter</groupId>
             <artifactId>junit-jupiter-engine</artifactId>
-            <version>5.7.0</version>
+            <version>5.9.0</version>
             <scope>test</scope>
         </dependency>
         <dependency>
@@ -279,6 +284,12 @@
             <version>2.2</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>io.micrometer</groupId>
+            <artifactId>micrometer-tracing-integration-test</artifactId>
+            <version>1.7.0</version>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.projectlombok</groupId>
             <artifactId>lombok</artifactId>

src/main/java/ru/rt/restream/reindexer/ReindexerConfiguration.java

@@ -15,6 +15,7 @@
  */
 package ru.rt.restream.reindexer;
 
+import io.micrometer.observation.ObservationRegistry;
 import ru.rt.restream.reindexer.binding.Binding;
 import ru.rt.restream.reindexer.binding.builtin.Builtin;
 import ru.rt.restream.reindexer.binding.builtin.server.BuiltinServer;
@@ -58,6 +59,8 @@ public final class ReindexerConfiguration {
 
     private SSLSocketFactory sslSocketFactory;
 
+    private ObservationRegistry observationRegistry = ObservationRegistry.NOOP;
+
     private ReindexerConfiguration() {
 
     }
@@ -176,6 +179,18 @@ public ReindexerConfiguration sslSocketFactory(SSLSocketFactory sslSocketFactory
         return this;
     }
 
+    /**
+     * Configure an {@link ObservationRegistry} to record connector's metrics and traces.
+     * Defaults to {@link ObservationRegistry#NOOP}.
+     *
+     * @param observationRegistry the {@link ObservationRegistry} to use
+     * @return the {@link ReindexerConfiguration} for further customizations
+     */
+    public ReindexerConfiguration observationRegistry(ObservationRegistry observationRegistry) {
+        this.observationRegistry = Objects.requireNonNull(observationRegistry, "observationRegistry cannot be null");
+        return this;
+    }
+
     /**
      * Build and return reindexer connector instance.
      *
@@ -210,6 +225,7 @@ private Binding getBinding(String protocol, List<URI> uris) {
                         .urls(urls)
                         .allowUnlistedDataSource(allowUnlistedDataSource)
                         .sslSocketFactory(sslSocketFactory)
+                        .observationRegistry(observationRegistry)
                         .build();
                 return new Cproto(dataSourceFactory, dataSourceConfig, connectionPoolSize, requestTimeout);
             case "builtin":

src/main/java/ru/rt/restream/reindexer/binding/Binding.java

@@ -83,6 +83,16 @@ public interface Binding {
 
     int RESULTS_NEED_OUTPUT_RANK = 0x400;
 
+    int ADD_TX_ITEM = 26;
+
+    int UPDATE_QUERY_TX = 31;
+
+    int DELETE_QUERY_TX = 30;
+
+    int COMMIT_TX = 27;
+
+    int ROLLBACK_TX = 28;
+
     /**
      * Open or create a new namespace and indexes based on passed definition.
      *