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..b7fea5ef --- /dev/null +++ b/src/main/java/ru/rt/restream/reindexer/binding/cproto/CommandObservationContext.java @@ -0,0 +1,39 @@ +/* + * 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.transport.Kind; +import io.micrometer.observation.transport.RequestReplySenderContext; +import lombok.Getter; +import ru.rt.restream.reindexer.ReindexerResponse; + +/** + * A context for command observation. + */ +@Getter +final class CommandObservationContext extends RequestReplySenderContext { + + private final int command; + + private final Object[] arguments; + + CommandObservationContext(int command, Object[] arguments) { + super((carrier, key, value) -> {}, Kind.CLIENT); + this.command = command; + this.arguments = arguments; + } + +} 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..989bd1ab --- /dev/null +++ b/src/main/java/ru/rt/restream/reindexer/binding/cproto/CommandObservationConvention.java @@ -0,0 +1,217 @@ +/* + * 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.common.KeyValues; +import io.micrometer.observation.Observation; +import io.micrometer.observation.ObservationConvention; +import org.apache.commons.lang3.ArrayUtils; +import ru.rt.restream.reindexer.binding.Binding; +import ru.rt.restream.reindexer.binding.definition.NamespaceDefinition; + +import java.net.URI; + +/** + * An {@link ObservationConvention} to handle {@link CommandObservationContext} observations. + */ +final class CommandObservationConvention implements ObservationConvention { + + private static final String OBSERVATION_NAME = "reindexer.rpc"; + + private final Gson gson = new GsonBuilder() + .setFieldNamingPolicy(FieldNamingPolicy.LOWER_CASE_WITH_UNDERSCORES) + .create(); + + @Override + public String getName() { + return OBSERVATION_NAME; + } + + @Override + public String getContextualName(CommandObservationContext context) { + return OBSERVATION_NAME + "." + getCommandName(context.getCommand()); + } + + @Override + public KeyValues getLowCardinalityKeyValues(CommandObservationContext context) { + String command = getCommandName(context.getCommand()); + String collection = getCollectionName(context); + String responseStatusCode = context.getResponse() != null ? String.valueOf(context.getResponse().getCode()) : ""; + String networkTransport = ""; + String namespace = ""; + String serverAddress = ""; + String serverPort = ""; + if (context.getRemoteServiceAddress() != null) { + URI uri = URI.create(context.getRemoteServiceAddress()); + networkTransport = uri.getScheme(); + namespace = uri.getPath().substring(1); + serverAddress = uri.getHost(); + serverPort = String.valueOf(uri.getPort()); + } + return KeyValues.of( + "db.system.name", "reindexer", + "db.command.name", command, + "db.namespace", namespace, + "db.collection.name", collection, + "network.transport", networkTransport, + "server.address", serverAddress, + "server.port", serverPort, + "db.response.status_code", responseStatusCode + ); + } + + @Override + public KeyValues getHighCardinalityKeyValues(CommandObservationContext context) { + String transactionId = getTransactionId(context); + String requestId = getRequestId(context); + return KeyValues.of( + "db.reindexer.tx_id", transactionId, + "db.reindexer.rq_id", requestId + ); + } + + @Override + public boolean supportsContext(Observation.Context context) { + return context instanceof CommandObservationContext; + } + + private String getCommandName(int command) { + 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); + } + } + + private String getCollectionName(CommandObservationContext context) { + switch (context.getCommand()) { + case Binding.OPEN_NAMESPACE: + 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. + String value = ArrayUtils.get(context.getArguments(), 0, "").toString(); + if (context.getCommand() == Binding.OPEN_NAMESPACE) { + // For openNamespace command, the [0] argument is a JSON string representing the namespace definition. + try { + NamespaceDefinition namespace = gson.fromJson(value, NamespaceDefinition.class); + return namespace.getName() != null ? namespace.getName() : ""; + } catch (JsonSyntaxException ignored) { + // Return an empty string if the JSON string is invalid. + return ""; + } + } + return value; + default: + return ""; + } + } + + private String getTransactionId(CommandObservationContext context) { + switch (context.getCommand()) { + case Binding.ADD_TX_ITEM: + // Command arguments[5] is the transaction id. + return ArrayUtils.get(context.getArguments(), 5, "").toString(); + case Binding.UPDATE_QUERY_TX: + case Binding.DELETE_QUERY_TX: + // Command arguments[1] is the transaction id. + return ArrayUtils.get(context.getArguments(), 1, "").toString(); + case Binding.START_TRANSACTION: + // Response arguments[0] is the transaction id. + return context.getResponse() != null + ? ArrayUtils.get(context.getResponse().getArguments(), 0, "").toString() + : ""; + case Binding.COMMIT_TX: + case Binding.ROLLBACK_TX: + // Command arguments[0] is the transaction id. + return ArrayUtils.get(context.getArguments(), 0, "").toString(); + default: + return ""; + } + } + + private String getRequestId(CommandObservationContext context) { + switch (context.getCommand()) { + case Binding.FETCH_RESULTS: + case Binding.CLOSE_RESULTS: + // Command arguments[0] is the request id. + return ArrayUtils.get(context.getArguments(), 0, "").toString(); + case Binding.SELECT: + case Binding.SELECT_SQL: + // Response arguments[1] is the request id. + return context.getResponse() != null + ? ArrayUtils.get(context.getResponse().getArguments(), 1, "").toString() + : ""; + default: + return ""; + } + } + +} 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..99c6f755 --- /dev/null +++ b/src/test/java/ru/rt/restream/reindexer/observability/ReindexerObservabilityTest.java @@ -0,0 +1,127 @@ +/* + * 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.commit(); + + boolean exists = namespace.query() + .where("id", Query.Condition.EQ, 123) + .exists(); + assertThat(exists).isTrue(); + + System.out.println(METER_REGISTRY.getMetersAsString()); + + assertThat(tracer.getFinishedSpans()) + .hasSize(10) + .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" + ); + }; + } + + @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; + } + +}