diff --git a/README.md b/README.md index e2377f5d..2ea18668 100644 --- a/README.md +++ b/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 + + io.micrometer + micrometer-registry-prometheus + ${micrometer.version} + +``` + +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. diff --git a/pom.xml b/pom.xml index 5ea787b3..ba105036 100644 --- a/pom.xml +++ b/pom.xml @@ -243,6 +243,11 @@ commons-lang3 3.11 + + io.micrometer + micrometer-observation + 1.17.0 + commons-io commons-io @@ -252,19 +257,19 @@ org.junit.jupiter junit-jupiter-api - 5.7.0 + 5.9.0 test org.junit.jupiter junit-jupiter-params - 5.7.0 + 5.9.0 test org.junit.jupiter junit-jupiter-engine - 5.7.0 + 5.9.0 test @@ -279,6 +284,12 @@ 2.2 test + + io.micrometer + micrometer-tracing-integration-test + 1.7.0 + test + org.projectlombok lombok diff --git a/src/main/java/ru/rt/restream/reindexer/ReindexerConfiguration.java b/src/main/java/ru/rt/restream/reindexer/ReindexerConfiguration.java index f5d90685..4e544670 100644 --- a/src/main/java/ru/rt/restream/reindexer/ReindexerConfiguration.java +++ b/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 uris) { .urls(urls) .allowUnlistedDataSource(allowUnlistedDataSource) .sslSocketFactory(sslSocketFactory) + .observationRegistry(observationRegistry) .build(); return new Cproto(dataSourceFactory, dataSourceConfig, connectionPoolSize, requestTimeout); case "builtin": diff --git a/src/main/java/ru/rt/restream/reindexer/binding/Binding.java b/src/main/java/ru/rt/restream/reindexer/binding/Binding.java index 97848473..430609cb 100644 --- a/src/main/java/ru/rt/restream/reindexer/binding/Binding.java +++ b/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. * diff --git a/src/main/java/ru/rt/restream/reindexer/binding/cproto/CommandObservationContext.java b/src/main/java/ru/rt/restream/reindexer/binding/cproto/CommandObservationContext.java new file mode 100644 index 00000000..1fd7e9dd --- /dev/null +++ b/src/main/java/ru/rt/restream/reindexer/binding/cproto/CommandObservationContext.java @@ -0,0 +1,241 @@ +/* + * Copyright 2020-present Restream + * + * 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 + * + * http://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 ru.rt.restream.reindexer.binding.cproto; + +import com.google.gson.FieldNamingPolicy; +import com.google.gson.Gson; +import com.google.gson.GsonBuilder; +import com.google.gson.JsonSyntaxException; +import io.micrometer.observation.transport.Kind; +import io.micrometer.observation.transport.RequestReplySenderContext; +import lombok.Getter; +import lombok.RequiredArgsConstructor; +import org.apache.commons.lang3.ArrayUtils; +import ru.rt.restream.reindexer.ReindexerResponse; +import ru.rt.restream.reindexer.binding.Binding; +import ru.rt.restream.reindexer.binding.definition.NamespaceDefinition; + +import java.net.URI; + +/** + * A context for command observation. + */ +final class CommandObservationContext extends RequestReplySenderContext { + + private static final Gson GSON = new GsonBuilder() + .setFieldNamingPolicy(FieldNamingPolicy.LOWER_CASE_WITH_UNDERSCORES) + .create(); + + private final int command; + + private final Object[] arguments; + + private String collectionName; + + private RemoteServerAddress remoteServerAddress; + + CommandObservationContext(int command, Object[] arguments) { + super((carrier, key, value) -> {}, Kind.CLIENT); + this.command = command; + this.arguments = arguments; + } + + String getCommandName() { + switch (command) { + case Binding.OPEN_NAMESPACE: + return "openNamespace"; + case Binding.CLOSE_NAMESPACE: + return "closeNamespace"; + case Binding.DROP_NAMESPACE: + return "dropNamespace"; + case Binding.ADD_INDEX: + return "addIndex"; + case Binding.UPDATE_INDEX: + return "updateIndex"; + case Binding.DROP_INDEX: + return "dropIndex"; + case Binding.MODIFY_ITEM: + return "modifyItem"; + case Binding.SELECT: + return "selectQuery"; + case Binding.UPDATE_QUERY: + return "updateQuery"; + case Binding.UPDATE_QUERY_TX: + return "updateQueryTx"; + case Binding.DELETE_QUERY: + return "deleteQuery"; + case Binding.DELETE_QUERY_TX: + return "deleteQueryTx"; + case Binding.SELECT_SQL: + return "selectSql"; + case Binding.FETCH_RESULTS: + return "fetchResults"; + case Binding.CLOSE_RESULTS: + return "closeResults"; + case Binding.START_TRANSACTION: + return "startTransaction"; + case Binding.ADD_TX_ITEM: + return "addTxItem"; + case Binding.COMMIT_TX: + return "commitTx"; + case Binding.ROLLBACK_TX: + return "rollbackTx"; + case Binding.PING: + return "ping"; + case Binding.GET_META: + return "getMeta"; + case Binding.PUT_META: + return "putMeta"; + default: + // Fallback to command code. + return String.valueOf(command); + } + } + + String getCollectionName() { + if (collectionName == null) { + collectionName = extractCollectionName(); + } + return collectionName; + } + + private String extractCollectionName() { + switch (command) { + case Binding.OPEN_NAMESPACE: { + // For openNamespace command, the [0] argument is a JSON string representing the namespace definition. + String json = ArrayUtils.get(arguments, 0, "").toString(); + try { + NamespaceDefinition namespace = GSON.fromJson(json, NamespaceDefinition.class); + return namespace.getName() != null ? namespace.getName() : ""; + } catch (JsonSyntaxException ignored) { + // Return an empty string if the JSON string is invalid. + return ""; + } + } + case Binding.UPDATE_QUERY: + case Binding.UPDATE_QUERY_TX: + case Binding.DELETE_QUERY: + case Binding.DELETE_QUERY_TX: + case Binding.SELECT: { + // Command arguments[0] is a byte array of query data. + Object queryData = ArrayUtils.get(arguments, 0, null); + // Read the variable length string from the buffer which is a namespace name. + return queryData instanceof byte[] + ? new ByteBuffer((byte[]) queryData).rewind().getVString() : ""; + } + case Binding.DROP_NAMESPACE: + case Binding.CLOSE_NAMESPACE: + case Binding.ADD_INDEX: + case Binding.UPDATE_INDEX: + case Binding.DROP_INDEX: + case Binding.MODIFY_ITEM: + case Binding.PUT_META: + case Binding.GET_META: + case Binding.START_TRANSACTION: + // Command arguments[0] is the namespace name. + return ArrayUtils.get(arguments, 0, "").toString(); + default: + return ""; + } + } + + String getTransactionId() { + switch (command) { + case Binding.ADD_TX_ITEM: + // Command arguments[5] is the transaction id. + return ArrayUtils.get(arguments, 5, "").toString(); + case Binding.UPDATE_QUERY_TX: + case Binding.DELETE_QUERY_TX: + // Command arguments[1] is the transaction id. + return ArrayUtils.get(arguments, 1, "").toString(); + case Binding.START_TRANSACTION: + // Response arguments[0] is the transaction id. + return getResponse() != null + ? ArrayUtils.get(getResponse().getArguments(), 0, "").toString() : ""; + case Binding.COMMIT_TX: + case Binding.ROLLBACK_TX: + // Command arguments[0] is the transaction id. + return ArrayUtils.get(arguments, 0, "").toString(); + default: + return ""; + } + } + + String getRequestId() { + switch (command) { + case Binding.FETCH_RESULTS: + case Binding.CLOSE_RESULTS: + // Command arguments[0] is the request id. + return ArrayUtils.get(arguments, 0, "").toString(); + case Binding.SELECT: + case Binding.SELECT_SQL: + // Response arguments[1] is the request id. + return getResponse() != null + ? ArrayUtils.get(getResponse().getArguments(), 1, "").toString() : ""; + default: + return ""; + } + } + + ExecutionType getExecutionType() { + return Thread.currentThread().getName().startsWith(ConnectionPool.ConnectionThreadFactory.POOL_NAME_PREFIX) + ? ExecutionType.ASYNC : ExecutionType.SYNC; + } + + RemoteServerAddress getRemoteServerAddress() { + if (remoteServerAddress == null && getRemoteServiceAddress() != null) { + URI uri = URI.create(getRemoteServiceAddress()); + String path = uri.getPath(); + String database = path != null && path.startsWith("/") ? path.substring(1) : path; + remoteServerAddress = new RemoteServerAddress( + uri.getScheme() != null ? uri.getScheme() : "", + uri.getHost() != null ? uri.getHost() : "", + database != null ? database : "", + uri.getPort() + ); + } + return remoteServerAddress; + } + + /** + * Code execution type for a command being run. + */ + enum ExecutionType { + + /** + * Synchronous execution. Usually called from the user's thread. + */ + SYNC, + + /** + * Asynchronous execution. Always called from the {@link ConnectionPool.ConnectionThreadFactory#POOL_NAME_PREFIX} thread. + */ + ASYNC + } + + /** + * Represents a remote server address i.e., protocol, host, database, and port. + */ + @Getter + @RequiredArgsConstructor + static final class RemoteServerAddress { + private final String protocol; + private final String host; + private final String database; + private final int port; + } + +} diff --git a/src/main/java/ru/rt/restream/reindexer/binding/cproto/CommandObservationConvention.java b/src/main/java/ru/rt/restream/reindexer/binding/cproto/CommandObservationConvention.java new file mode 100644 index 00000000..8e4a2e8f --- /dev/null +++ b/src/main/java/ru/rt/restream/reindexer/binding/cproto/CommandObservationConvention.java @@ -0,0 +1,88 @@ +/* + * Copyright 2020-present Restream + * + * 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 + * + * http://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 ru.rt.restream.reindexer.binding.cproto; + +import io.micrometer.common.KeyValues; +import io.micrometer.observation.Observation; +import io.micrometer.observation.ObservationConvention; + +/** + * An {@link ObservationConvention} to handle {@link CommandObservationContext} observations. + */ +final class CommandObservationConvention implements ObservationConvention { + + private static final String OBSERVATION_NAME = "reindexer.rpc"; + + @Override + public String getName() { + return OBSERVATION_NAME; + } + + @Override + public String getContextualName(CommandObservationContext context) { + return OBSERVATION_NAME + "." + context.getCommandName(); + } + + @Override + public KeyValues getLowCardinalityKeyValues(CommandObservationContext context) { + String commandName = context.getCommandName(); + String collectionName = context.getCollectionName(); + String responseStatusCode = context.getResponse() != null ? String.valueOf(context.getResponse().getCode()) : ""; + String executionType = context.getExecutionType().name(); + String networkTransport = ""; + String namespace = ""; + String serverAddress = ""; + String serverPort = ""; + CommandObservationContext.RemoteServerAddress remoteServerAddress = context.getRemoteServerAddress(); + if (remoteServerAddress != null) { + networkTransport = remoteServerAddress.getProtocol(); + namespace = remoteServerAddress.getDatabase(); + serverAddress = remoteServerAddress.getHost(); + serverPort = String.valueOf(remoteServerAddress.getPort()); + } + return KeyValues.of( + "db.system.name", "reindexer", + "db.command.name", commandName, + "db.namespace", namespace, + "db.collection.name", collectionName, + "network.transport", networkTransport, + "server.address", serverAddress, + "server.port", serverPort, + "code.execution_type", executionType, + "db.response.status_code", responseStatusCode + ); + } + + @Override + public KeyValues getHighCardinalityKeyValues(CommandObservationContext context) { + String threadId = String.valueOf(Thread.currentThread().getId()); + String threadName = Thread.currentThread().getName(); + String transactionId = context.getTransactionId(); + String requestId = context.getRequestId(); + return KeyValues.of( + "thread.id", threadId, + "thread.name", threadName, + "db.reindexer.tx_id", transactionId, + "db.reindexer.rq_id", requestId + ); + } + + @Override + public boolean supportsContext(Observation.Context context) { + return context instanceof CommandObservationContext; + } + +} diff --git a/src/main/java/ru/rt/restream/reindexer/binding/cproto/ConnectionPool.java b/src/main/java/ru/rt/restream/reindexer/binding/cproto/ConnectionPool.java index 9c904054..7f57745c 100644 --- a/src/main/java/ru/rt/restream/reindexer/binding/cproto/ConnectionPool.java +++ b/src/main/java/ru/rt/restream/reindexer/binding/cproto/ConnectionPool.java @@ -26,7 +26,9 @@ import java.time.Instant; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.Executors; import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.ThreadFactory; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; @@ -97,7 +99,7 @@ public ConnectionPool(DataSourceFactory dataSourceFactory, DataSourceConfigurati int connectionPoolSize, Duration requestTimeout) { this.dataSourceFactory = dataSourceFactory; this.dataSourceConfiguration = dataSourceConfiguration; - scheduler = new ScheduledThreadPoolExecutor(connectionPoolSize * 2 + 1); + scheduler = new ScheduledThreadPoolExecutor(connectionPoolSize * 2 + 1, new ConnectionThreadFactory()); scheduler.setRemoveOnCancelPolicy(true); connections = new ArrayList<>(connectionPoolSize); timeout = requestTimeout; @@ -222,4 +224,32 @@ public void run() { } + /** + * A {@link ThreadFactory} that creates threads with a rx specific prefix. + */ + static final class ConnectionThreadFactory implements ThreadFactory { + + static final String POOL_NAME_PREFIX = "rx-pool-"; + + private static final AtomicInteger poolNumber = new AtomicInteger(1); + + private final AtomicInteger threadNumber = new AtomicInteger(1); + + private final ThreadFactory defaultFactory = Executors.defaultThreadFactory(); + + private final String namePrefix; + + private ConnectionThreadFactory() { + namePrefix = POOL_NAME_PREFIX + poolNumber.getAndIncrement() + "-thread-"; + } + + @Override + public Thread newThread(Runnable runnable) { + Thread thread = defaultFactory.newThread(runnable); + thread.setName(namePrefix + threadNumber.getAndIncrement()); + return thread; + } + + } + } diff --git a/src/main/java/ru/rt/restream/reindexer/binding/cproto/CprotoRequestContext.java b/src/main/java/ru/rt/restream/reindexer/binding/cproto/CprotoRequestContext.java index 2deb15fd..08e0c364 100644 --- a/src/main/java/ru/rt/restream/reindexer/binding/cproto/CprotoRequestContext.java +++ b/src/main/java/ru/rt/restream/reindexer/binding/cproto/CprotoRequestContext.java @@ -19,6 +19,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import ru.rt.restream.reindexer.ReindexerResponse; +import ru.rt.restream.reindexer.binding.Binding; import ru.rt.restream.reindexer.binding.Consts; import ru.rt.restream.reindexer.binding.QueryResult; import ru.rt.restream.reindexer.binding.QueryResultReader; @@ -32,10 +33,6 @@ public class CprotoRequestContext implements RequestContext { private static final Logger LOGGER = LoggerFactory.getLogger(CprotoRequestContext.class); - private static final int FETCH_RESULTS = 50; - - private static final int CLOSE_RESULTS = 51; - private final QueryResultReader reader = new QueryResultReader(); private static final int RESULTS_WITH_JOINED = 0x100; @@ -74,7 +71,7 @@ public void fetchResults(int offset, int limit) { ? Consts.RESULTS_JSON : Consts.RESULTS_C_JSON | Consts.RESULTS_WITH_PAYLOAD_TYPES; int fetchCount = limit <= 0 ? Integer.MAX_VALUE : limit; - ReindexerResponse rpcResponse = ConnectionUtils.rpcCall(connection, FETCH_RESULTS, requestId, flags, offset, fetchCount); + ReindexerResponse rpcResponse = ConnectionUtils.rpcCall(connection, Binding.FETCH_RESULTS, requestId, flags, offset, fetchCount); queryResult = getQueryResult(rpcResponse); } @@ -84,7 +81,7 @@ public void fetchResults(int offset, int limit) { @Override public void closeResults() { if (requestId != -1) { - ReindexerResponse rpcResponse = connection.rpcCall(CLOSE_RESULTS, requestId); + ReindexerResponse rpcResponse = connection.rpcCall(Binding.CLOSE_RESULTS, requestId); if (rpcResponse.hasError()) { LOGGER.error("rx: query close error {}", rpcResponse.getErrorMessage()); } diff --git a/src/main/java/ru/rt/restream/reindexer/binding/cproto/CprotoTransactionContext.java b/src/main/java/ru/rt/restream/reindexer/binding/cproto/CprotoTransactionContext.java index ec404990..2fc1f7d4 100644 --- a/src/main/java/ru/rt/restream/reindexer/binding/cproto/CprotoTransactionContext.java +++ b/src/main/java/ru/rt/restream/reindexer/binding/cproto/CprotoTransactionContext.java @@ -18,6 +18,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import ru.rt.restream.reindexer.ReindexerResponse; +import ru.rt.restream.reindexer.binding.Binding; import ru.rt.restream.reindexer.binding.Consts; import ru.rt.restream.reindexer.binding.RequestContext; import ru.rt.restream.reindexer.binding.TransactionContext; @@ -26,7 +27,6 @@ import java.util.concurrent.CompletableFuture; import static ru.rt.restream.reindexer.binding.Binding.SELECT; -import static ru.rt.restream.reindexer.binding.Consts.FORMAT_C_JSON; /** * A transaction context which establish a connection to the Reindexer instance via RPC. @@ -35,16 +35,6 @@ public class CprotoTransactionContext implements TransactionContext { private static final Logger LOGGER = LoggerFactory.getLogger(CprotoTransactionContext.class); - private static final int ADD_TX_ITEM = 26; - - private static final int UPDATE_QUERY_TX = 31; - - private static final int DELETE_QUERY_TX = 30; - - private static final int COMMIT_TX = 27; - - private static final int ROLLBACK_TX = 28; - private final long transactionId; private final Connection connection; @@ -63,14 +53,14 @@ public CprotoTransactionContext(long transactionId, Connection connection) { @Override public void modifyItem(byte[] data, int format, int mode, String[] precepts, int stateToken) { byte[] packedPrecepts = packPrecepts(precepts); - ConnectionUtils.rpcCallNoResults(connection, ADD_TX_ITEM, format, data, mode, packedPrecepts, stateToken, + ConnectionUtils.rpcCallNoResults(connection, Binding.ADD_TX_ITEM, format, data, mode, packedPrecepts, stateToken, transactionId); } @Override public CompletableFuture modifyItemAsync(byte[] data, int format, int mode, String[] precepts, int stateToken) { byte[] packedPrecepts = packPrecepts(precepts); - return connection.rpcCallAsync(ADD_TX_ITEM, format, data, mode, packedPrecepts, stateToken, transactionId); + return connection.rpcCallAsync(Binding.ADD_TX_ITEM, format, data, mode, packedPrecepts, stateToken, transactionId); } private byte[] packPrecepts(String[] precepts) { @@ -88,12 +78,12 @@ private byte[] packPrecepts(String[] precepts) { @Override public void updateQuery(byte[] queryData) { - ConnectionUtils.rpcCallNoResults(connection, UPDATE_QUERY_TX, queryData, transactionId); + ConnectionUtils.rpcCallNoResults(connection, Binding.UPDATE_QUERY_TX, queryData, transactionId); } @Override public void deleteQuery(byte[] queryData) { - ConnectionUtils.rpcCallNoResults(connection, DELETE_QUERY_TX, queryData, transactionId); + ConnectionUtils.rpcCallNoResults(connection, Binding.DELETE_QUERY_TX, queryData, transactionId); } @Override @@ -109,7 +99,7 @@ public RequestContext selectQuery(byte[] queryData, int fetchCount, long[] ptVer @Override public void commit() { try { - ConnectionUtils.rpcCallNoResults(connection, COMMIT_TX, transactionId); + ConnectionUtils.rpcCallNoResults(connection, Binding.COMMIT_TX, transactionId); } catch (Exception e) { LOGGER.error("rx: commit error", e); } @@ -118,7 +108,7 @@ public void commit() { @Override public void rollback() { try { - ConnectionUtils.rpcCallNoResults(connection, ROLLBACK_TX, transactionId); + ConnectionUtils.rpcCallNoResults(connection, Binding.ROLLBACK_TX, transactionId); } catch (Exception e) { LOGGER.error("rx: rollback error", e); } diff --git a/src/main/java/ru/rt/restream/reindexer/binding/cproto/DataSourceConfiguration.java b/src/main/java/ru/rt/restream/reindexer/binding/cproto/DataSourceConfiguration.java index 7391c59e..1218c704 100644 --- a/src/main/java/ru/rt/restream/reindexer/binding/cproto/DataSourceConfiguration.java +++ b/src/main/java/ru/rt/restream/reindexer/binding/cproto/DataSourceConfiguration.java @@ -16,6 +16,7 @@ package ru.rt.restream.reindexer.binding.cproto; +import io.micrometer.observation.ObservationRegistry; import org.apache.commons.lang3.mutable.MutableInt; import java.util.ArrayList; @@ -44,6 +45,11 @@ public class DataSourceConfiguration { */ private final SSLSocketFactory sslSocketFactory; + /** + * An {@link ObservationRegistry} to record connector's metrics and traces. + */ + private final ObservationRegistry observationRegistry; + /** * An index of the current active data source. */ @@ -54,6 +60,7 @@ private DataSourceConfiguration(Builder builder) { allowUnlistedDataSource = builder.allowUnlistedDataSource; active = builder.active; sslSocketFactory = builder.sslSocketFactory; + observationRegistry = builder.observationRegistry; } public static Builder builder() { @@ -86,6 +93,16 @@ public SSLSocketFactory getSslSocketFactory() { return sslSocketFactory; } + /** + * Returns an {@link ObservationRegistry} to record connector's metrics and traces. + * Defaults to {@link ObservationRegistry#NOOP}. + * + * @return the {@link ObservationRegistry} to use + */ + public ObservationRegistry getObservationRegistry() { + return observationRegistry; + } + /** * Returns the index of the current active data source. * @@ -124,6 +141,11 @@ public static class Builder { */ private SSLSocketFactory sslSocketFactory; + /** + * An {@link ObservationRegistry} to record connector's metrics and traces. + */ + private ObservationRegistry observationRegistry = ObservationRegistry.NOOP; + /** * An index of the current active data source. */ @@ -190,6 +212,18 @@ public Builder 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 Builder} for further customizations + */ + public Builder observationRegistry(ObservationRegistry observationRegistry) { + this.observationRegistry = Objects.requireNonNull(observationRegistry, "observationRegistry cannot be null"); + return this; + } + /** * Build and return a {@link DataSource} configuration. * diff --git a/src/main/java/ru/rt/restream/reindexer/binding/cproto/DataSourceFactoryStrategy.java b/src/main/java/ru/rt/restream/reindexer/binding/cproto/DataSourceFactoryStrategy.java index 02d3c7ac..483390c2 100644 --- a/src/main/java/ru/rt/restream/reindexer/binding/cproto/DataSourceFactoryStrategy.java +++ b/src/main/java/ru/rt/restream/reindexer/binding/cproto/DataSourceFactoryStrategy.java @@ -43,7 +43,7 @@ public enum DataSourceFactoryStrategy implements DataSourceFactory { public DataSource getDataSource(DataSourceConfiguration configuration) { List urls = configuration.getUrls(); configuration.setActive((configuration.getActive() + 1) % urls.size()); - return new PhysicalDataSource(urls.get(configuration.getActive()), configuration.getSslSocketFactory()); + return super.getDataSource(configuration); } }, @@ -55,7 +55,7 @@ public DataSource getDataSource(DataSourceConfiguration configuration) { public DataSource getDataSource(DataSourceConfiguration configuration) { List urls = configuration.getUrls(); configuration.setActive(ThreadLocalRandom.current().nextInt(urls.size())); - return new PhysicalDataSource(urls.get(configuration.getActive()), configuration.getSslSocketFactory()); + return super.getDataSource(configuration); } }, @@ -132,6 +132,16 @@ public DataSource getDataSource(DataSourceConfiguration configuration) { } }; + @Override + public DataSource getDataSource(DataSourceConfiguration configuration) { + String url = configuration.getUrls().get(configuration.getActive()); + PhysicalDataSource dataSource = new PhysicalDataSource(url, configuration.getSslSocketFactory()); + if (configuration.getObservationRegistry().isNoop()) { + return dataSource; + } + return new ObservationDataSource(dataSource, url, configuration.getObservationRegistry()); + } + /** * Get a list of online {@link Nodes.Node} of Reindexer cluster. * diff --git a/src/main/java/ru/rt/restream/reindexer/binding/cproto/ObservationDataSource.java b/src/main/java/ru/rt/restream/reindexer/binding/cproto/ObservationDataSource.java new file mode 100644 index 00000000..6e4f22e4 --- /dev/null +++ b/src/main/java/ru/rt/restream/reindexer/binding/cproto/ObservationDataSource.java @@ -0,0 +1,111 @@ +/* + * Copyright 2020-present Restream + * + * 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 + * + * http://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 ru.rt.restream.reindexer.binding.cproto; + +import io.micrometer.observation.Observation; +import io.micrometer.observation.ObservationRegistry; +import lombok.RequiredArgsConstructor; +import ru.rt.restream.reindexer.ReindexerResponse; +import ru.rt.restream.reindexer.exceptions.ReindexerExceptionFactory; + +import java.time.Duration; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ScheduledThreadPoolExecutor; + +/** + * A {@link DataSource} that wraps a target {@link DataSource} and instruments it using configured {@link ObservationRegistry}. + */ +@RequiredArgsConstructor +final class ObservationDataSource implements DataSource { + + private static final CommandObservationConvention CONVENTION = new CommandObservationConvention(); + + private final DataSource delegate; + + private final String url; + + private final ObservationRegistry registry; + + @Override + public Connection getConnection(Duration timeout, ScheduledThreadPoolExecutor scheduler) { + Connection connection = delegate.getConnection(timeout, scheduler); + return new ObservationConnection(connection); + } + + @RequiredArgsConstructor + private final class ObservationConnection implements Connection { + + private final Connection delegate; + + @Override + public ReindexerResponse rpcCall(int command, Object... args) { + CommandObservationContext context = new CommandObservationContext(command, args); + context.setRemoteServiceAddress(url); + Observation observation = Observation.createNotStarted(CONVENTION, () -> context, registry).start(); + try (Observation.Scope scope = observation.openScope()) { + ReindexerResponse response = delegate.rpcCall(command, args); + context.setResponse(response); + if (response.hasError()) { + observation.error(ReindexerExceptionFactory.fromResponse(response)); + } + return response; + } catch (Throwable t) { + observation.error(t); + throw t; + } finally { + observation.stop(); + } + } + + @Override + public CompletableFuture rpcCallAsync(int command, Object... args) { + CommandObservationContext context = new CommandObservationContext(command, args); + context.setRemoteServiceAddress(url); + Observation observation = Observation.createNotStarted(CONVENTION, () -> context, registry).start(); + CompletableFuture future; + try (Observation.Scope scope = observation.openScope()) { + future = delegate.rpcCallAsync(command, args); + } catch (Throwable t) { + observation.error(t); + observation.stop(); + throw t; + } + return future.whenComplete((response, error) -> { + if (error != null) { + observation.error(error); + } else { + context.setResponse(response); + if (response.hasError()) { + observation.error(ReindexerExceptionFactory.fromResponse(response)); + } + } + observation.stop(); + }); + } + + @Override + public boolean hasError() { + return delegate.hasError(); + } + + @Override + public void close() { + delegate.close(); + } + + } + +} diff --git a/src/test/java/ru/rt/restream/reindexer/db/ClearDbReindexer.java b/src/test/java/ru/rt/restream/reindexer/db/ClearDbReindexer.java index 0cfc80cb..aba1761e 100644 --- a/src/test/java/ru/rt/restream/reindexer/db/ClearDbReindexer.java +++ b/src/test/java/ru/rt/restream/reindexer/db/ClearDbReindexer.java @@ -34,7 +34,7 @@ public class ClearDbReindexer extends Reindexer { * Removes all registered namespaces. * TODO: to do refactoring after implementation of Reindexer.enumNamespaces */ - void clear() { + public void clear() { Binding binding = getBinding(); namespaceMap.values().stream() .map(ReindexerNamespace::getName) diff --git a/src/test/java/ru/rt/restream/reindexer/db/DbLocator.java b/src/test/java/ru/rt/restream/reindexer/db/DbLocator.java index 42f31758..6cbf90aa 100644 --- a/src/test/java/ru/rt/restream/reindexer/db/DbLocator.java +++ b/src/test/java/ru/rt/restream/reindexer/db/DbLocator.java @@ -16,6 +16,7 @@ package ru.rt.restream.reindexer.db; +import io.micrometer.observation.ObservationRegistry; import org.apache.commons.io.FileUtils; import ru.rt.restream.category.CprotoTest; import ru.rt.restream.reindexer.Reindexer; @@ -75,9 +76,13 @@ public class DbLocator { private static boolean serverStarted = false; public static ClearDbReindexer getDb(Type type) { + return getDb(type, ObservationRegistry.NOOP); + } + + public static ClearDbReindexer getDb(Type type, ObservationRegistry observationRegistry) { ClearDbReindexer db = instancesForUse.get(type); if (db == null) { - db = addDbInstance(type); + db = addDbInstance(type, observationRegistry); } return db; } @@ -98,7 +103,7 @@ static void closeAllDbInstances() throws IOException { serverStarted = false; } - private static ClearDbReindexer addDbInstance(Type type) { + private static ClearDbReindexer addDbInstance(Type type, ObservationRegistry observationRegistry) { switch (type) { case BUILTIN: ClearDbReindexer builtinDb = new ClearDbReindexer(ReindexerConfiguration.builder() @@ -107,11 +112,13 @@ private static ClearDbReindexer addDbInstance(Type type) { instancesForUse.put(Type.BUILTIN, builtinDb); instancesForClose.put(builtinDb, BUILTIN_DB_PATH); return builtinDb; + case OBSERVATION: case CPROTOS: case CPROTO: ReindexerConfiguration cprotoConfig = ReindexerConfiguration.builder() .connectionPoolSize(4) .sslSocketFactory(getSslSocketFactory(type)) + .observationRegistry(observationRegistry) .requestTimeout(Duration.ofSeconds(30L)); List urls = getCprotoDbUrlsFromProperty(); @@ -123,7 +130,7 @@ private static ClearDbReindexer addDbInstance(Type type) { urls.forEach(cprotoConfig::url); ClearDbReindexer cprotoDb = new ClearDbReindexer(cprotoConfig.getReindexer().getBinding()); - instancesForUse.put(Type.CPROTO, cprotoDb); + instancesForUse.put(type, cprotoDb); instancesForClose.put(cprotoDb, null); return cprotoDb; default: @@ -204,6 +211,7 @@ private static void copyResourceToReindexerDirectory(String fileName) { public enum Type { BUILTIN, + OBSERVATION, CPROTOS, CPROTO } diff --git a/src/test/java/ru/rt/restream/reindexer/observability/ReindexerObservabilityTest.java b/src/test/java/ru/rt/restream/reindexer/observability/ReindexerObservabilityTest.java new file mode 100644 index 00000000..e0861a2c --- /dev/null +++ b/src/test/java/ru/rt/restream/reindexer/observability/ReindexerObservabilityTest.java @@ -0,0 +1,149 @@ +/* + * Copyright 2020-present Restream + * + * 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 + * + * http://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 ru.rt.restream.reindexer.observability; + +import io.micrometer.core.instrument.MeterRegistry; +import io.micrometer.core.instrument.observation.DefaultMeterObservationHandler; +import io.micrometer.core.instrument.simple.SimpleMeterRegistry; +import io.micrometer.observation.ObservationRegistry; +import io.micrometer.tracing.exporter.FinishedSpan; +import io.micrometer.tracing.test.SampleTestRunner; +import lombok.Data; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.extension.ExtendWith; +import ru.rt.restream.reindexer.Namespace; +import ru.rt.restream.reindexer.NamespaceOptions; +import ru.rt.restream.reindexer.Query; +import ru.rt.restream.reindexer.Transaction; +import ru.rt.restream.reindexer.annotations.Reindex; +import ru.rt.restream.reindexer.db.ClearDbReindexer; +import ru.rt.restream.reindexer.db.DbCloseExtension; +import ru.rt.restream.reindexer.db.DbLocator; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Tests for Reindexer observability. + */ +@ExtendWith(DbCloseExtension.class) +public class ReindexerObservabilityTest extends SampleTestRunner { + + private static final SimpleMeterRegistry METER_REGISTRY = new SimpleMeterRegistry(); + + private static final ObservationRegistry OBSERVATION_REGISTRY = ObservationRegistry.create(); + + static { + OBSERVATION_REGISTRY.observationConfig().observationHandler(new DefaultMeterObservationHandler(METER_REGISTRY)); + } + + private static ClearDbReindexer db; + + @BeforeAll + static void beforeAll() { + db = DbLocator.getDb(DbLocator.Type.OBSERVATION, OBSERVATION_REGISTRY); + } + + @Override + protected MeterRegistry createMeterRegistry() { + return METER_REGISTRY; + } + + @Override + protected ObservationRegistry createObservationRegistry() { + return OBSERVATION_REGISTRY; + } + + @AfterEach + void tearDown() { + if (db != null) { + db.clear(); + } + } + + @Override + public SampleTestRunnerConsumer yourCode() { + return (tracer, meterRegistry) -> { + String namespaceName = "items"; + Namespace namespace = db.openNamespace(namespaceName, NamespaceOptions.defaultOptions(), TestItem.class); + + Transaction tx = namespace.beginTransaction(); + TestItem testItem = new TestItem(); + testItem.setId(123); + testItem.setName("TestName"); + testItem.setValue("TestValue"); + tx.insert(testItem); + + tx.query() + .where("id", Query.Condition.EQ, 123) + .set("name", "UpdatedName") + .update(); + + tx.query() + .where("id", Query.Condition.EQ, 123) + .delete(); + + tx.commit(); + + boolean exists = namespace.query() + .where("id", Query.Condition.EQ, 123) + .exists(); + assertThat(exists).isFalse(); + + namespace.query() + .where("id", Query.Condition.EQ, 123) + .set("name", "UpdatedName") + .update(); + + namespace.query() + .where("id", Query.Condition.EQ, 123) + .delete(); + + System.out.println(METER_REGISTRY.getMetersAsString()); + + assertThat(tracer.getFinishedSpans()) + .hasSize(14) + .extracting(FinishedSpan::getName) + .contains( + "reindexer.rpc.openNamespace", + "reindexer.rpc.addIndex", + "reindexer.rpc.addIndex", + "reindexer.rpc.addIndex", + "reindexer.rpc.startTransaction", + "reindexer.rpc.addTxItem", + "reindexer.rpc.selectQuery", + "reindexer.rpc.addTxItem", + "reindexer.rpc.commitTx", + "reindexer.rpc.selectQuery", + "reindexer.rpc.updateQuery", + "reindexer.rpc.updateQueryTx", + "reindexer.rpc.deleteQuery", + "reindexer.rpc.deleteQueryTx" + ); + }; + } + + @Data + public static class TestItem { + @Reindex(name = "id", isPrimaryKey = true) + private Integer id; + @Reindex(name = "name") + private String name; + @Reindex(name = "value") + private String value; + } + +}