From 985258900082839f2bd6754ea0b679f031734572 Mon Sep 17 00:00:00 2001 From: Maryanto <54889592+maryantocinn@users.noreply.github.com> Date: Sun, 8 Mar 2026 00:20:30 +0700 Subject: [PATCH] Add Micrometer Observation support. (#3249) Signed-off-by: Maryanto <54889592+maryantocinn@users.noreply.github.com> --- pom.xml | 8 +- ...ultElasticsearchObservationConvention.java | 70 ++++++ .../client/elc/ElasticsearchObservation.java | 99 ++++++++ .../elc/ElasticsearchObservationContext.java | 83 +++++++ .../ElasticsearchObservationConvention.java | 34 +++ .../elc/ElasticsearchOperationName.java | 56 +++++ .../client/elc/ElasticsearchTemplate.java | 214 +++++++++++----- .../elc/ReactiveElasticsearchTemplate.java | 229 ++++++++++++++---- .../core/AbstractElasticsearchTemplate.java | 46 +++- ...AbstractReactiveElasticsearchTemplate.java | 48 +++- ...asticsearchObservationConventionTests.java | 202 +++++++++++++++ .../ElasticsearchObservationContextTests.java | 91 +++++++ ...icsearchObservationDocumentationTests.java | 207 ++++++++++++++++ .../elc/ObservabilityIntegrationTests.java | 164 +++++++++++++ 14 files changed, 1421 insertions(+), 130 deletions(-) create mode 100644 src/main/java/org/springframework/data/elasticsearch/client/elc/DefaultElasticsearchObservationConvention.java create mode 100644 src/main/java/org/springframework/data/elasticsearch/client/elc/ElasticsearchObservation.java create mode 100644 src/main/java/org/springframework/data/elasticsearch/client/elc/ElasticsearchObservationContext.java create mode 100644 src/main/java/org/springframework/data/elasticsearch/client/elc/ElasticsearchObservationConvention.java create mode 100644 src/main/java/org/springframework/data/elasticsearch/client/elc/ElasticsearchOperationName.java create mode 100644 src/test/java/org/springframework/data/elasticsearch/client/elc/DefaultElasticsearchObservationConventionTests.java create mode 100644 src/test/java/org/springframework/data/elasticsearch/client/elc/ElasticsearchObservationContextTests.java create mode 100644 src/test/java/org/springframework/data/elasticsearch/client/elc/ElasticsearchObservationDocumentationTests.java create mode 100644 src/test/java/org/springframework/data/elasticsearch/client/elc/ObservabilityIntegrationTests.java diff --git a/pom.xml b/pom.xml index e3a56fb8f..0ef62a3b8 100644 --- a/pom.xml +++ b/pom.xml @@ -355,7 +355,13 @@ test - + + io.micrometer + micrometer-observation-test + test + + + diff --git a/src/main/java/org/springframework/data/elasticsearch/client/elc/DefaultElasticsearchObservationConvention.java b/src/main/java/org/springframework/data/elasticsearch/client/elc/DefaultElasticsearchObservationConvention.java new file mode 100644 index 000000000..a24acd0f5 --- /dev/null +++ b/src/main/java/org/springframework/data/elasticsearch/client/elc/DefaultElasticsearchObservationConvention.java @@ -0,0 +1,70 @@ +/* + * Copyright 2026-present the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.data.elasticsearch.client.elc; + +import io.micrometer.common.KeyValues; + +/** + * Default {@link ElasticsearchObservationConvention} implementation. + * + * @author maryantocinn + * @since 6.1 + */ +public class DefaultElasticsearchObservationConvention implements ElasticsearchObservationConvention { + + public static final DefaultElasticsearchObservationConvention INSTANCE = new DefaultElasticsearchObservationConvention(); + + @Override + public String getName() { + return ElasticsearchObservation.ELASTICSEARCH_COMMAND_OBSERVATION.getName(); + } + + @Override + public String getContextualName(ElasticsearchObservationContext context) { + + String indexName = context.getIndexName(); + if (indexName != null) { + return context.getOperationName().getValue() + " " + indexName; + } + return context.getOperationName().getValue(); + } + + @Override + public KeyValues getLowCardinalityKeyValues(ElasticsearchObservationContext context) { + + KeyValues keyValues = KeyValues.of( + ElasticsearchObservation.LowCardinalityKeyNames.OPERATION.withValue(context.getOperationName().getValue())); + + String indexName = context.getIndexName(); + if (indexName != null) { + keyValues = keyValues.and(ElasticsearchObservation.LowCardinalityKeyNames.COLLECTION.withValue(indexName)); + } + + return keyValues; + } + + @Override + public KeyValues getHighCardinalityKeyValues(ElasticsearchObservationContext context) { + + Integer batchSize = context.getBatchSize(); + if (batchSize != null) { + return KeyValues.of( + ElasticsearchObservation.HighCardinalityKeyNames.BATCH_SIZE.withValue(String.valueOf(batchSize))); + } + + return KeyValues.empty(); + } +} diff --git a/src/main/java/org/springframework/data/elasticsearch/client/elc/ElasticsearchObservation.java b/src/main/java/org/springframework/data/elasticsearch/client/elc/ElasticsearchObservation.java new file mode 100644 index 000000000..3b2c2b149 --- /dev/null +++ b/src/main/java/org/springframework/data/elasticsearch/client/elc/ElasticsearchObservation.java @@ -0,0 +1,99 @@ +/* + * Copyright 2026-present the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.data.elasticsearch.client.elc; + +import io.micrometer.common.docs.KeyName; +import io.micrometer.observation.Observation; +import io.micrometer.observation.ObservationConvention; +import io.micrometer.observation.docs.ObservationDocumentation; + +/** + * {@link ObservationDocumentation} for Spring Data Elasticsearch template operations. + * + * @author maryantocinn + * @since 6.1 + */ +public enum ElasticsearchObservation implements ObservationDocumentation { + + /** + * Timer created around a Spring Data Elasticsearch template operation. + */ + ELASTICSEARCH_COMMAND_OBSERVATION { + @Override + public String getName() { + return "spring.data.elasticsearch.command"; + } + + @Override + public Class> getDefaultConvention() { + return DefaultElasticsearchObservationConvention.class; + } + + @Override + public KeyName[] getLowCardinalityKeyNames() { + return LowCardinalityKeyNames.values(); + } + + @Override + public KeyName[] getHighCardinalityKeyNames() { + return HighCardinalityKeyNames.values(); + } + }; + + /** + * Low cardinality key names for Spring Data Elasticsearch observations. These become metric dimensions and MUST be + * present on every observation to satisfy backends like Prometheus that require consistent tag key sets. + */ + enum LowCardinalityKeyNames implements KeyName { + + /** + * The Spring Data operation being performed (e.g., save, search, delete, bulk). + */ + OPERATION { + @Override + public String asString() { + return "spring.data.operation"; + } + }, + + /** + * The target collection (index) name. Only present when the operation targets a specific index. + */ + COLLECTION { + @Override + public String asString() { + return "spring.data.collection"; + } + } + } + + /** + * High cardinality key names for Spring Data Elasticsearch observations. These appear only on traces/spans, not on + * metrics, because their values are unbounded or optional per operation. + */ + enum HighCardinalityKeyNames implements KeyName { + + /** + * The number of operations included in a batch (bulk) request. Only present for bulk operations. + */ + BATCH_SIZE { + @Override + public String asString() { + return "spring.data.batch.size"; + } + } + } +} diff --git a/src/main/java/org/springframework/data/elasticsearch/client/elc/ElasticsearchObservationContext.java b/src/main/java/org/springframework/data/elasticsearch/client/elc/ElasticsearchObservationContext.java new file mode 100644 index 000000000..41222d90f --- /dev/null +++ b/src/main/java/org/springframework/data/elasticsearch/client/elc/ElasticsearchObservationContext.java @@ -0,0 +1,83 @@ +/* + * Copyright 2026-present the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.data.elasticsearch.client.elc; + +import org.jspecify.annotations.Nullable; +import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates; + +import io.micrometer.observation.Observation; + +/** + * {@link Observation.Context} for Spring Data Elasticsearch operations. One instance is created per observed operation. + * It carries contextual data that conventions use to produce observation names and key-values. + * + * @author maryantocinn + * @since 6.1 + */ +public class ElasticsearchObservationContext extends Observation.Context { + + private final ElasticsearchOperationName operationName; + @Nullable + private final IndexCoordinates indexCoordinates; + @Nullable + private Integer batchSize; + + public ElasticsearchObservationContext(ElasticsearchOperationName operationName, + @Nullable IndexCoordinates indexCoordinates) { + this.operationName = operationName; + this.indexCoordinates = indexCoordinates; + } + + /** + * @return the Spring Data operation being performed. + */ + public ElasticsearchOperationName getOperationName() { + return operationName; + } + + /** + * @return the target index coordinates, or {@literal null} if the operation is not index-specific. + */ + @Nullable + public IndexCoordinates getIndexCoordinates() { + return indexCoordinates; + } + + /** + * @return the comma-joined index name(s), or {@literal null} if no index coordinates are set. + */ + @Nullable + public String getIndexName() { + return indexCoordinates != null ? String.join(",", indexCoordinates.getIndexNames()) : null; + } + + /** + * @return the batch size, or {@literal null} if not a batch operation. + */ + @Nullable + public Integer getBatchSize() { + return batchSize; + } + + /** + * Set the number of operations included in a batch (bulk) request. + * + * @param batchSize the batch size, can be {@literal null} + */ + public void setBatchSize(@Nullable Integer batchSize) { + this.batchSize = batchSize; + } +} diff --git a/src/main/java/org/springframework/data/elasticsearch/client/elc/ElasticsearchObservationConvention.java b/src/main/java/org/springframework/data/elasticsearch/client/elc/ElasticsearchObservationConvention.java new file mode 100644 index 000000000..e9cdf79fa --- /dev/null +++ b/src/main/java/org/springframework/data/elasticsearch/client/elc/ElasticsearchObservationConvention.java @@ -0,0 +1,34 @@ +/* + * Copyright 2026-present the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.data.elasticsearch.client.elc; + +import io.micrometer.observation.Observation; +import io.micrometer.observation.ObservationConvention; + +/** + * {@link ObservationConvention} for Spring Data Elasticsearch operations. Implement this interface and register it as a + * bean to customize observation names and key-values. + * + * @author maryantocinn + * @since 6.1 + */ +public interface ElasticsearchObservationConvention extends ObservationConvention { + + @Override + default boolean supportsContext(Observation.Context context) { + return context instanceof ElasticsearchObservationContext; + } +} diff --git a/src/main/java/org/springframework/data/elasticsearch/client/elc/ElasticsearchOperationName.java b/src/main/java/org/springframework/data/elasticsearch/client/elc/ElasticsearchOperationName.java new file mode 100644 index 000000000..2c0b48560 --- /dev/null +++ b/src/main/java/org/springframework/data/elasticsearch/client/elc/ElasticsearchOperationName.java @@ -0,0 +1,56 @@ +/* + * Copyright 2026-present the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.data.elasticsearch.client.elc; + +/** + * Enumeration of Spring Data Elasticsearch operation names used in observations. + * + * @author maryantocinn + * @since 6.1 + */ +public enum ElasticsearchOperationName { + + SAVE("save"), // + INDEX("index"), // + GET("get"), // + MULTI_GET("multiGet"), // + EXISTS("exists"), // + DELETE("delete"), // + DELETE_BY_QUERY("deleteByQuery"), // + BULK("bulk"), // + UPDATE("update"), // + UPDATE_BY_QUERY("updateByQuery"), // + COUNT("count"), // + SEARCH("search"); + + private final String value; + + ElasticsearchOperationName(String value) { + this.value = value; + } + + /** + * @return the operation name as a string value used in observation key values. + */ + public String getValue() { + return value; + } + + @Override + public String toString() { + return value; + } +} diff --git a/src/main/java/org/springframework/data/elasticsearch/client/elc/ElasticsearchTemplate.java b/src/main/java/org/springframework/data/elasticsearch/client/elc/ElasticsearchTemplate.java index 571e74935..f383ac0f7 100644 --- a/src/main/java/org/springframework/data/elasticsearch/client/elc/ElasticsearchTemplate.java +++ b/src/main/java/org/springframework/data/elasticsearch/client/elc/ElasticsearchTemplate.java @@ -17,30 +17,11 @@ package org.springframework.data.elasticsearch.client.elc; import static org.springframework.data.elasticsearch.client.elc.TypeUtils.*; -import co.elastic.clients.elasticsearch.ElasticsearchClient; -import co.elastic.clients.elasticsearch._types.Time; -import co.elastic.clients.elasticsearch.core.*; -import co.elastic.clients.elasticsearch.core.bulk.BulkResponseItem; -import co.elastic.clients.elasticsearch.core.msearch.MultiSearchResponseItem; -import co.elastic.clients.elasticsearch.core.search.ResponseBody; -import co.elastic.clients.elasticsearch.sql.ElasticsearchSqlClient; -import co.elastic.clients.elasticsearch.sql.QueryResponse; -import co.elastic.clients.json.JsonpMapper; -import co.elastic.clients.transport.Version; - -import java.io.IOException; -import java.time.Duration; -import java.util.ArrayList; -import java.util.Collections; -import java.util.HashMap; -import java.util.Iterator; -import java.util.List; -import java.util.Map; -import java.util.stream.Collectors; - import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.jspecify.annotations.Nullable; +import org.springframework.beans.BeansException; +import org.springframework.context.ApplicationContext; import org.springframework.data.elasticsearch.BulkFailureException; import org.springframework.data.elasticsearch.client.UnsupportedBackendOperation; import org.springframework.data.elasticsearch.core.AbstractElasticsearchTemplate; @@ -62,6 +43,30 @@ import org.springframework.data.elasticsearch.core.script.Script; import org.springframework.data.elasticsearch.core.sql.SqlResponse; import org.springframework.util.Assert; +import java.io.IOException; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.function.Supplier; +import java.util.stream.Collectors; + +import co.elastic.clients.elasticsearch.ElasticsearchClient; +import co.elastic.clients.elasticsearch._types.Time; +import co.elastic.clients.elasticsearch.core.*; +import co.elastic.clients.elasticsearch.core.bulk.BulkResponseItem; +import co.elastic.clients.elasticsearch.core.msearch.MultiSearchResponseItem; +import co.elastic.clients.elasticsearch.core.search.ResponseBody; +import co.elastic.clients.elasticsearch.sql.ElasticsearchSqlClient; +import co.elastic.clients.elasticsearch.sql.QueryResponse; +import co.elastic.clients.json.JsonpMapper; +import co.elastic.clients.transport.Version; +import io.micrometer.observation.Observation; +import io.micrometer.observation.ObservationRegistry; + /** * Implementation of {@link org.springframework.data.elasticsearch.core.ElasticsearchOperations} using the new * Elasticsearch client. @@ -70,12 +75,15 @@ import org.springframework.util.Assert; * @author Hamid Rahimi * @author Illia Ulianov * @author Haibo Liu + * @author maryantocinn * @since 4.4 */ public class ElasticsearchTemplate extends AbstractElasticsearchTemplate { private static final Log LOGGER = LogFactory.getLog(ElasticsearchTemplate.class); + @Nullable private ElasticsearchObservationConvention observationConvention; + private final ElasticsearchClient client; private final ElasticsearchSqlClient sqlClient; private final RequestConverter requestConverter; @@ -113,6 +121,61 @@ public class ElasticsearchTemplate extends AbstractElasticsearchTemplate { protected AbstractElasticsearchTemplate doCopy() { return new ElasticsearchTemplate(client, elasticsearchConverter); } + + @Override + protected void customizeCopy(AbstractElasticsearchTemplate copy) { + + if (copy instanceof ElasticsearchTemplate elasticsearchTemplate) { + elasticsearchTemplate.observationConvention = this.observationConvention; + } + } + + @Override + public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { + + super.setApplicationContext(applicationContext); + + if (observationRegistry == ObservationRegistry.NOOP) { + applicationContext.getBeanProvider(ObservationRegistry.class).ifAvailable(this::setObservationRegistry); + } + + if (observationConvention == null) { + applicationContext.getBeanProvider(ElasticsearchObservationConvention.class) + .ifAvailable(this::setObservationConvention); + } + } + + /** + * Set a custom {@link ElasticsearchObservationConvention} to override the default convention. + * + * @param observationConvention can be {@literal null}. + * @since 6.1 + */ + public void setObservationConvention(@Nullable ElasticsearchObservationConvention observationConvention) { + this.observationConvention = observationConvention; + } + + private T observe(ElasticsearchOperationName operationName, @Nullable IndexCoordinates index, + Supplier action) { + Observation observation = createObservation(operationName, index, null); + return observation.observe(action); + } + + private T observe(ElasticsearchOperationName operationName, @Nullable IndexCoordinates index, int batchSize, + Supplier action) { + Observation observation = createObservation(operationName, index, batchSize); + return observation.observe(action); + } + + private Observation createObservation(ElasticsearchOperationName operationName, @Nullable IndexCoordinates index, + @Nullable Integer batchSize) { + + ElasticsearchObservationContext context = new ElasticsearchObservationContext(operationName, index); + context.setBatchSize(batchSize); + + return ElasticsearchObservation.ELASTICSEARCH_COMMAND_OBSERVATION.observation(observationConvention, + DefaultElasticsearchObservationConvention.INSTANCE, () -> context, observationRegistry); + } // endregion // region child templates @@ -137,16 +200,45 @@ public class ElasticsearchTemplate extends AbstractElasticsearchTemplate { // endregion // region document operations + @Override + public T save(T entity, IndexCoordinates index) { + return observe(ElasticsearchOperationName.SAVE, index, () -> super.save(entity, index)); + } + + @Override + public String index(IndexQuery query, IndexCoordinates index) { + return observe(ElasticsearchOperationName.INDEX, index, () -> super.index(query, index)); + } + + @Override + public boolean exists(String id, IndexCoordinates index) { + return observe(ElasticsearchOperationName.EXISTS, index, () -> super.exists(id, index)); + } + + @Override + public String delete(String id, IndexCoordinates index) { + return observe(ElasticsearchOperationName.DELETE, index, () -> super.delete(id, index)); + } + + @Override + public List bulkOperation(List queries, BulkOptions bulkOptions, + IndexCoordinates index) { + return observe(ElasticsearchOperationName.BULK, index, queries.size(), + () -> super.bulkOperation(queries, bulkOptions, index)); + } + @Override @Nullable public T get(String id, Class clazz, IndexCoordinates index) { - GetRequest getRequest = requestConverter.documentGetRequest(elasticsearchConverter.convertId(id), - routingResolver.getRouting(), index); - GetResponse getResponse = execute(client -> client.get(getRequest, EntityAsMap.class)); + return observe(ElasticsearchOperationName.GET, index, () -> { + GetRequest getRequest = requestConverter.documentGetRequest(elasticsearchConverter.convertId(id), + routingResolver.getRouting(), index); + GetResponse getResponse = execute(client -> client.get(getRequest, EntityAsMap.class)); - ReadDocumentCallback callback = new ReadDocumentCallback<>(elasticsearchConverter, clazz, index); - return callback.doWith(DocumentAdapters.from(getResponse)); + ReadDocumentCallback callback = new ReadDocumentCallback<>(elasticsearchConverter, clazz, index); + return callback.doWith(DocumentAdapters.from(getResponse)); + }); } @Override @@ -155,15 +247,17 @@ public class ElasticsearchTemplate extends AbstractElasticsearchTemplate { Assert.notNull(query, "query must not be null"); Assert.notNull(clazz, "clazz must not be null"); - MgetRequest request = requestConverter.documentMgetRequest(query, clazz, index); - MgetResponse result = execute(client -> client.mget(request, EntityAsMap.class)); + return observe(ElasticsearchOperationName.MULTI_GET, index, () -> { + MgetRequest request = requestConverter.documentMgetRequest(query, clazz, index); + MgetResponse result = execute(client -> client.mget(request, EntityAsMap.class)); - ReadDocumentCallback callback = new ReadDocumentCallback<>(elasticsearchConverter, clazz, index); + ReadDocumentCallback callback = new ReadDocumentCallback<>(elasticsearchConverter, clazz, index); - return DocumentAdapters.from(result).stream() // - .map(multiGetItem -> MultiGetItem.of( // - multiGetItem.isFailed() ? null : callback.doWith(multiGetItem.getItem()), multiGetItem.getFailure())) // - .collect(Collectors.toList()); + return DocumentAdapters.from(result).stream() // + .map(multiGetItem -> MultiGetItem.of( // + multiGetItem.isFailed() ? null : callback.doWith(multiGetItem.getItem()), multiGetItem.getFailure())) // + .collect(Collectors.toList()); + }); } @Override @@ -185,22 +279,26 @@ public class ElasticsearchTemplate extends AbstractElasticsearchTemplate { public ByQueryResponse delete(DeleteQuery query, Class clazz, IndexCoordinates index) { Assert.notNull(query, "query must not be null"); - DeleteByQueryRequest request = requestConverter.documentDeleteByQueryRequest(query, routingResolver.getRouting(), - clazz, index, getRefreshPolicy()); + return observe(ElasticsearchOperationName.DELETE_BY_QUERY, index, () -> { + DeleteByQueryRequest request = requestConverter.documentDeleteByQueryRequest(query, routingResolver.getRouting(), + clazz, index, getRefreshPolicy()); - DeleteByQueryResponse response = execute(client -> client.deleteByQuery(request)); + DeleteByQueryResponse response = execute(client -> client.deleteByQuery(request)); - return responseConverter.byQueryResponse(response); + return responseConverter.byQueryResponse(response); + }); } @Override public UpdateResponse update(UpdateQuery updateQuery, IndexCoordinates index) { - UpdateRequest request = requestConverter.documentUpdateRequest(updateQuery, index, getRefreshPolicy(), - routingResolver.getRouting()); - co.elastic.clients.elasticsearch.core.UpdateResponse response = execute( - client -> client.update(request, Document.class)); - return UpdateResponse.of(result(response.result())); + return observe(ElasticsearchOperationName.UPDATE, index, () -> { + UpdateRequest request = requestConverter.documentUpdateRequest(updateQuery, index, + getRefreshPolicy(), routingResolver.getRouting()); + co.elastic.clients.elasticsearch.core.UpdateResponse response = execute( + client -> client.update(request, Document.class)); + return UpdateResponse.of(result(response.result())); + }); } @Override @@ -209,11 +307,13 @@ public class ElasticsearchTemplate extends AbstractElasticsearchTemplate { Assert.notNull(updateQuery, "updateQuery must not be null"); Assert.notNull(index, "index must not be null"); - UpdateByQueryRequest request = requestConverter.documentUpdateByQueryRequest(updateQuery, index, - getRefreshPolicy()); + return observe(ElasticsearchOperationName.UPDATE_BY_QUERY, index, () -> { + UpdateByQueryRequest request = requestConverter.documentUpdateByQueryRequest(updateQuery, index, + getRefreshPolicy()); - UpdateByQueryResponse byQueryResponse = execute(client -> client.updateByQuery(request)); - return responseConverter.byQueryResponse(byQueryResponse); + UpdateByQueryResponse byQueryResponse = execute(client -> client.updateByQuery(request)); + return responseConverter.byQueryResponse(byQueryResponse); + }); } @Override @@ -328,12 +428,14 @@ public class ElasticsearchTemplate extends AbstractElasticsearchTemplate { Assert.notNull(query, "query must not be null"); Assert.notNull(index, "index must not be null"); - SearchRequest searchRequest = requestConverter.searchRequest(query, routingResolver.getRouting(), clazz, index, - true); + return observe(ElasticsearchOperationName.COUNT, index, () -> { + SearchRequest searchRequest = requestConverter.searchRequest(query, routingResolver.getRouting(), clazz, index, + true); - SearchResponse searchResponse = execute(client -> client.search(searchRequest, EntityAsMap.class)); + SearchResponse searchResponse = execute(client -> client.search(searchRequest, EntityAsMap.class)); - return searchResponse.hits().total().value(); + return searchResponse.hits().total().value(); + }); } @Override @@ -343,11 +445,13 @@ public class ElasticsearchTemplate extends AbstractElasticsearchTemplate { Assert.notNull(clazz, "clazz must not be null"); Assert.notNull(index, "index must not be null"); - if (query instanceof SearchTemplateQuery searchTemplateQuery) { - return doSearch(searchTemplateQuery, clazz, index); - } else { - return doSearch(query, clazz, index); - } + return observe(ElasticsearchOperationName.SEARCH, index, () -> { + if (query instanceof SearchTemplateQuery searchTemplateQuery) { + return doSearch(searchTemplateQuery, clazz, index); + } else { + return doSearch(query, clazz, index); + } + }); } protected SearchHits doSearch(Query query, Class clazz, IndexCoordinates index) { diff --git a/src/main/java/org/springframework/data/elasticsearch/client/elc/ReactiveElasticsearchTemplate.java b/src/main/java/org/springframework/data/elasticsearch/client/elc/ReactiveElasticsearchTemplate.java index a89fed31a..97289c85c 100644 --- a/src/main/java/org/springframework/data/elasticsearch/client/elc/ReactiveElasticsearchTemplate.java +++ b/src/main/java/org/springframework/data/elasticsearch/client/elc/ReactiveElasticsearchTemplate.java @@ -18,31 +18,13 @@ package org.springframework.data.elasticsearch.client.elc; import static co.elastic.clients.util.ApiTypeHelper.*; import static org.springframework.data.elasticsearch.client.elc.TypeUtils.*; -import co.elastic.clients.elasticsearch._types.Result; -import co.elastic.clients.elasticsearch.core.*; -import co.elastic.clients.elasticsearch.core.bulk.BulkResponseItem; -import co.elastic.clients.elasticsearch.core.search.ResponseBody; -import co.elastic.clients.json.JsonpMapper; -import co.elastic.clients.transport.Version; -import co.elastic.clients.transport.endpoints.BooleanResponse; -import org.jspecify.annotations.NonNull; -import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; -import reactor.util.function.Tuple2; - -import java.time.Duration; -import java.util.Collection; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.function.BiFunction; -import java.util.function.Function; -import java.util.stream.Collectors; - import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.jspecify.annotations.NonNull; import org.jspecify.annotations.Nullable; import org.reactivestreams.Publisher; +import org.springframework.beans.BeansException; +import org.springframework.context.ApplicationContext; import org.springframework.data.domain.Sort; import org.springframework.data.elasticsearch.BulkFailureException; import org.springframework.data.elasticsearch.NoSuchIndexException; @@ -53,6 +35,7 @@ import org.springframework.data.elasticsearch.core.AggregationContainer; import org.springframework.data.elasticsearch.core.IndexedObjectInformation; import org.springframework.data.elasticsearch.core.MultiGetItem; import org.springframework.data.elasticsearch.core.ReactiveIndexOperations; +import org.springframework.data.elasticsearch.core.SearchHit; import org.springframework.data.elasticsearch.core.cluster.ReactiveClusterOperations; import org.springframework.data.elasticsearch.core.convert.ElasticsearchConverter; import org.springframework.data.elasticsearch.core.document.Document; @@ -69,6 +52,28 @@ import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; import org.springframework.util.StringUtils; +import java.time.Duration; +import java.util.Collection; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.function.BiFunction; +import java.util.function.Function; +import java.util.stream.Collectors; + +import co.elastic.clients.elasticsearch._types.Result; +import co.elastic.clients.elasticsearch.core.*; +import co.elastic.clients.elasticsearch.core.bulk.BulkResponseItem; +import co.elastic.clients.elasticsearch.core.search.ResponseBody; +import co.elastic.clients.json.JsonpMapper; +import co.elastic.clients.transport.Version; +import co.elastic.clients.transport.endpoints.BooleanResponse; +import io.micrometer.observation.Observation; +import io.micrometer.observation.ObservationRegistry; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.util.function.Tuple2; + /** * Implementation of {@link org.springframework.data.elasticsearch.core.ReactiveElasticsearchOperations} using the new * Elasticsearch client. @@ -76,12 +81,15 @@ import org.springframework.util.StringUtils; * @author Peter-Josef Meisch * @author Illia Ulianov * @author Junghoon Ban + * @author maryantocinn * @since 4.4 */ public class ReactiveElasticsearchTemplate extends AbstractReactiveElasticsearchTemplate { private static final Log LOGGER = LogFactory.getLog(ReactiveElasticsearchTemplate.class); + @Nullable private ElasticsearchObservationConvention observationConvention; + private final ReactiveElasticsearchClient client; private final ReactiveElasticsearchSqlClient sqlClient; private final RequestConverter requestConverter; @@ -102,7 +110,105 @@ public class ReactiveElasticsearchTemplate extends AbstractReactiveElasticsearch exceptionTranslator = new ElasticsearchExceptionTranslator(jsonpMapper); } + @Override + public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { + + super.setApplicationContext(applicationContext); + + if (observationRegistry == ObservationRegistry.NOOP) { + applicationContext.getBeanProvider(ObservationRegistry.class).ifAvailable(this::setObservationRegistry); + } + + if (observationConvention == null) { + applicationContext.getBeanProvider(ElasticsearchObservationConvention.class) + .ifAvailable(this::setObservationConvention); + } + } + + /** + * Set a custom {@link ElasticsearchObservationConvention} to override the default convention. + * + * @param observationConvention can be {@literal null}. + * @since 6.1 + */ + public void setObservationConvention(@Nullable ElasticsearchObservationConvention observationConvention) { + this.observationConvention = observationConvention; + } + + private Mono observeMono(ElasticsearchOperationName operationName, @Nullable IndexCoordinates index, + Mono mono) { + return Mono.defer(() -> { + Observation observation = createObservation(operationName, index, null); + return mono.doOnError(observation::error) // + .doFinally(signalType -> observation.stop()) + .contextWrite(context -> context.put(Observation.class, observation)) + .doOnSubscribe(subscription -> observation.start()); + }); + } + + private Flux observeFlux(ElasticsearchOperationName operationName, @Nullable IndexCoordinates index, + Flux flux) { + return Flux.defer(() -> { + Observation observation = createObservation(operationName, index, null); + return flux.doOnError(observation::error) // + .doFinally(signalType -> observation.stop()) + .contextWrite(context -> context.put(Observation.class, observation)) + .doOnSubscribe(subscription -> observation.start()); + }); + } + + private Mono observeMono(ElasticsearchOperationName operationName, @Nullable IndexCoordinates index, + int batchSize, Mono mono) { + return Mono.defer(() -> { + Observation observation = createObservation(operationName, index, batchSize); + return mono.doOnError(observation::error) // + .doFinally(signalType -> observation.stop()) + .contextWrite(context -> context.put(Observation.class, observation)) + .doOnSubscribe(subscription -> observation.start()); + }); + } + + private Observation createObservation(ElasticsearchOperationName operationName, @Nullable IndexCoordinates index, + @Nullable Integer batchSize) { + + ElasticsearchObservationContext context = new ElasticsearchObservationContext(operationName, index); + context.setBatchSize(batchSize); + + return ElasticsearchObservation.ELASTICSEARCH_COMMAND_OBSERVATION.observation(observationConvention, + DefaultElasticsearchObservationConvention.INSTANCE, () -> context, observationRegistry); + } + // region Document operations + @Override + public Mono save(T entity, IndexCoordinates index) { + return observeMono(ElasticsearchOperationName.SAVE, index, super.save(entity, index)); + } + + @Override + public Mono exists(String id, IndexCoordinates index) { + return observeMono(ElasticsearchOperationName.EXISTS, index, super.exists(id, index)); + } + + @Override + public Mono delete(Object entity, IndexCoordinates index) { + return observeMono(ElasticsearchOperationName.DELETE, index, super.delete(entity, index)); + } + + @Override + public Mono delete(String id, IndexCoordinates index) { + return observeMono(ElasticsearchOperationName.DELETE, index, super.delete(id, index)); + } + + @Override + public Flux> search(Query query, Class entityType, Class resultType, IndexCoordinates index) { + return observeFlux(ElasticsearchOperationName.SEARCH, index, super.search(query, entityType, resultType, index)); + } + + @Override + public Mono count(Query query, Class entityType, IndexCoordinates index) { + return observeMono(ElasticsearchOperationName.COUNT, index, super.count(query, entityType, index)); + } + @Override protected Mono> doIndex(T entity, IndexCoordinates index) { @@ -124,7 +230,7 @@ public class ReactiveElasticsearchTemplate extends AbstractReactiveElasticsearch Assert.notNull(entitiesPublisher, "entitiesPublisher must not be null!"); - return entitiesPublisher // + return observeFlux(ElasticsearchOperationName.BULK, index, entitiesPublisher // .flatMapMany(entities -> Flux.fromIterable(entities) // .concatMap(entity -> maybeCallbackBeforeConvert(entity, index)) // ).collectList() // @@ -146,12 +252,12 @@ public class ReactiveElasticsearchTemplate extends AbstractReactiveElasticsearch response.index(), // response.seqNo(), // response.primaryTerm(), // - response.version()), - converter, + response.version()), // + converter, // routingResolver); return maybeCallbackAfterSave(updatedEntity, index); }); - }); + })); } @Override @@ -172,9 +278,11 @@ public class ReactiveElasticsearchTemplate extends AbstractReactiveElasticsearch public Mono delete(DeleteQuery query, Class entityType, IndexCoordinates index) { Assert.notNull(query, "query must not be null"); - DeleteByQueryRequest request = requestConverter.documentDeleteByQueryRequest(query, routingResolver.getRouting(), - entityType, index, getRefreshPolicy()); - return Mono.from(execute(client -> client.deleteByQuery(request))).map(responseConverter::byQueryResponse); + return observeMono(ElasticsearchOperationName.DELETE_BY_QUERY, index, Mono.defer(() -> { + DeleteByQueryRequest request = requestConverter.documentDeleteByQueryRequest(query, routingResolver.getRouting(), + entityType, index, getRefreshPolicy()); + return Mono.from(execute(client -> client.deleteByQuery(request))).map(responseConverter::byQueryResponse); + })); } @Override @@ -184,13 +292,15 @@ public class ReactiveElasticsearchTemplate extends AbstractReactiveElasticsearch Assert.notNull(entityType, "entityType must not be null"); Assert.notNull(index, "index must not be null"); - GetRequest getRequest = requestConverter.documentGetRequest(id, routingResolver.getRouting(), index); + return observeMono(ElasticsearchOperationName.GET, index, Mono.defer(() -> { + GetRequest getRequest = requestConverter.documentGetRequest(id, routingResolver.getRouting(), index); - Mono> getResponse = Mono - .from(execute(client -> client.get(getRequest, EntityAsMap.class))); + Mono> getResponse = Mono // + .from(execute(client -> client.get(getRequest, EntityAsMap.class))); - ReadDocumentCallback callback = new ReadDocumentCallback<>(converter, entityType, index); - return getResponse.flatMap(response -> callback.toEntity(DocumentAdapters.from(response))); + ReadDocumentCallback callback = new ReadDocumentCallback<>(converter, entityType, index); + return getResponse.flatMap(response -> callback.toEntity(DocumentAdapters.from(response))); + })); } @Override @@ -227,13 +337,15 @@ public class ReactiveElasticsearchTemplate extends AbstractReactiveElasticsearch Assert.notNull(updateQuery, "UpdateQuery must not be null"); Assert.notNull(index, "Index must not be null"); - UpdateRequest request = requestConverter.documentUpdateRequest(updateQuery, index, getRefreshPolicy(), - routingResolver.getRouting()); + return observeMono(ElasticsearchOperationName.UPDATE, index, Mono.defer(() -> { + UpdateRequest request = requestConverter.documentUpdateRequest(updateQuery, index, + getRefreshPolicy(), routingResolver.getRouting()); - return Mono.from(execute(client -> client.update(request, Document.class))).flatMap(response -> { - UpdateResponse.Result result = result(response.result()); - return result == null ? Mono.empty() : Mono.just(UpdateResponse.of(result)); - }); + return Mono.from(execute(client -> client.update(request, Document.class))).flatMap(response -> { + UpdateResponse.Result result = result(response.result()); + return result == null ? Mono.empty() : Mono.just(UpdateResponse.of(result)); + }); + })); } @Override @@ -248,7 +360,8 @@ public class ReactiveElasticsearchTemplate extends AbstractReactiveElasticsearch Assert.notNull(bulkOptions, "BulkOptions must not be null"); Assert.notNull(index, "Index must not be null"); - return doBulkOperation(queries, bulkOptions, index).then(); + return observeMono(ElasticsearchOperationName.BULK, index, queries.size(), + doBulkOperation(queries, bulkOptions, index).then()); } private Flux doBulkOperation(List queries, BulkOptions bulkOptions, IndexCoordinates index) { @@ -311,22 +424,24 @@ public class ReactiveElasticsearchTemplate extends AbstractReactiveElasticsearch Assert.notNull(query, "query must not be null"); Assert.notNull(clazz, "clazz must not be null"); - MgetRequest request = requestConverter.documentMgetRequest(query, clazz, index); + return observeFlux(ElasticsearchOperationName.MULTI_GET, index, Flux.defer(() -> { + MgetRequest request = requestConverter.documentMgetRequest(query, clazz, index); - ReadDocumentCallback callback = new ReadDocumentCallback<>(converter, clazz, index); + ReadDocumentCallback callback = new ReadDocumentCallback<>(converter, clazz, index); - Publisher> response = execute(client -> client.mget(request, EntityAsMap.class)); + Publisher> response = execute(client -> client.mget(request, EntityAsMap.class)); - return Mono.from(response)// - .flatMapMany(it -> Flux.fromIterable(DocumentAdapters.from(it))) // - .flatMap(multiGetItem -> { - if (multiGetItem.isFailed()) { - return Mono.just(MultiGetItem.of(null, multiGetItem.getFailure())); - } else { - return callback.toEntity(multiGetItem.getItem()) // - .map(t -> MultiGetItem.of(t, multiGetItem.getFailure())); - } - }); + return Mono.from(response)// + .flatMapMany(it -> Flux.fromIterable(DocumentAdapters.from(it))) // + .flatMap(multiGetItem -> { + if (multiGetItem.isFailed()) { + return Mono.just(MultiGetItem.of(null, multiGetItem.getFailure())); + } else { + return callback.toEntity(multiGetItem.getItem()) // + .map(t -> MultiGetItem.of(t, multiGetItem.getFailure())); + } + }); + })); } // endregion @@ -336,6 +451,14 @@ public class ReactiveElasticsearchTemplate extends AbstractReactiveElasticsearch return new ReactiveElasticsearchTemplate(client, converter); } + @Override + protected void customizeCopy(AbstractReactiveElasticsearchTemplate copy) { + + if (copy instanceof ReactiveElasticsearchTemplate reactiveTemplate) { + reactiveTemplate.observationConvention = this.observationConvention; + } + } + // region search operations @Override diff --git a/src/main/java/org/springframework/data/elasticsearch/core/AbstractElasticsearchTemplate.java b/src/main/java/org/springframework/data/elasticsearch/core/AbstractElasticsearchTemplate.java index 7993fd1bc..6d1ffc13a 100644 --- a/src/main/java/org/springframework/data/elasticsearch/core/AbstractElasticsearchTemplate.java +++ b/src/main/java/org/springframework/data/elasticsearch/core/AbstractElasticsearchTemplate.java @@ -15,15 +15,6 @@ */ package org.springframework.data.elasticsearch.core; -import java.time.Duration; -import java.util.Arrays; -import java.util.Collections; -import java.util.Iterator; -import java.util.List; -import java.util.Objects; -import java.util.concurrent.CompletableFuture; -import java.util.stream.Collectors; - import org.jspecify.annotations.Nullable; import org.springframework.beans.BeansException; import org.springframework.context.ApplicationContext; @@ -43,7 +34,6 @@ import org.springframework.data.elasticsearch.core.mapping.ElasticsearchPersiste import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates; import org.springframework.data.elasticsearch.core.mapping.SimpleElasticsearchMappingContext; import org.springframework.data.elasticsearch.core.query.BulkOptions; -import org.springframework.data.elasticsearch.core.query.ByQueryResponse; import org.springframework.data.elasticsearch.core.query.IndexQuery; import org.springframework.data.elasticsearch.core.query.IndexQueryBuilder; import org.springframework.data.elasticsearch.core.query.MoreLikeThisQuery; @@ -61,6 +51,17 @@ import org.springframework.data.util.Streamable; import org.springframework.util.Assert; import org.springframework.util.StringUtils; +import java.time.Duration; +import java.util.Arrays; +import java.util.Collections; +import java.util.Iterator; +import java.util.List; +import java.util.Objects; +import java.util.concurrent.CompletableFuture; +import java.util.stream.Collectors; + +import io.micrometer.observation.ObservationRegistry; + /** * This class contains methods that are common to different implementations of the {@link ElasticsearchOperations} * interface that use different clients, like the different Java clients from Elasticsearch or some external @@ -85,6 +86,7 @@ public abstract class AbstractElasticsearchTemplate implements ElasticsearchOper @Nullable protected EntityCallbacks entityCallbacks; @Nullable protected RefreshPolicy refreshPolicy; protected RoutingResolver routingResolver; + protected ObservationRegistry observationRegistry = ObservationRegistry.NOOP; public AbstractElasticsearchTemplate() { this(null); @@ -117,6 +119,8 @@ public abstract class AbstractElasticsearchTemplate implements ElasticsearchOper copy.setRoutingResolver(routingResolver); copy.setRefreshPolicy(refreshPolicy); + copy.setObservationRegistry(observationRegistry); + customizeCopy(copy); return copy; } @@ -171,6 +175,28 @@ public abstract class AbstractElasticsearchTemplate implements ElasticsearchOper return refreshPolicy; } + /** + * Set the {@link ObservationRegistry} to use for recording observations. + * + * @param observationRegistry must not be {@literal null}. + * @since 6.1 + */ + public void setObservationRegistry(ObservationRegistry observationRegistry) { + + Assert.notNull(observationRegistry, "observationRegistry must not be null"); + + this.observationRegistry = observationRegistry; + } + + /** + * Hook for subclasses to copy additional state during {@link #copy()}. Called after all common fields have been + * copied. The default implementation does nothing. + * + * @param copy the new template instance to customize + */ + protected void customizeCopy(AbstractElasticsearchTemplate copy) { + } + /** * logs the versions of the different Elasticsearch components. * diff --git a/src/main/java/org/springframework/data/elasticsearch/core/AbstractReactiveElasticsearchTemplate.java b/src/main/java/org/springframework/data/elasticsearch/core/AbstractReactiveElasticsearchTemplate.java index c82e39a40..a485ce10b 100644 --- a/src/main/java/org/springframework/data/elasticsearch/core/AbstractReactiveElasticsearchTemplate.java +++ b/src/main/java/org/springframework/data/elasticsearch/core/AbstractReactiveElasticsearchTemplate.java @@ -15,17 +15,6 @@ */ package org.springframework.data.elasticsearch.core; -import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; -import reactor.core.publisher.Sinks; -import reactor.util.function.Tuple2; - -import java.time.Duration; -import java.util.Collection; -import java.util.List; -import java.util.concurrent.atomic.AtomicBoolean; -import java.util.stream.Collectors; - import org.jspecify.annotations.Nullable; import org.reactivestreams.Subscriber; import org.reactivestreams.Subscription; @@ -59,6 +48,18 @@ import org.springframework.data.elasticsearch.support.VersionInfo; import org.springframework.data.mapping.callback.ReactiveEntityCallbacks; import org.springframework.util.Assert; +import java.time.Duration; +import java.util.Collection; +import java.util.List; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.stream.Collectors; + +import io.micrometer.observation.ObservationRegistry; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.core.publisher.Sinks; +import reactor.util.function.Tuple2; + /** * Base class keeping common code for implementations of the {@link ReactiveElasticsearchOperations} interface * independent of the used client. @@ -77,6 +78,7 @@ abstract public class AbstractReactiveElasticsearchTemplate protected RoutingResolver routingResolver; protected @Nullable ReactiveEntityCallbacks entityCallbacks; + protected ObservationRegistry observationRegistry = ObservationRegistry.NOOP; // region Initialization protected AbstractReactiveElasticsearchTemplate(@Nullable ElasticsearchConverter converter) { @@ -109,6 +111,8 @@ abstract public class AbstractReactiveElasticsearchTemplate } copy.setRoutingResolver(routingResolver); + copy.setObservationRegistry(observationRegistry); + customizeCopy(copy); return copy; } @@ -162,6 +166,28 @@ abstract public class AbstractReactiveElasticsearchTemplate this.entityCallbacks = entityCallbacks; } + /** + * Set the {@link ObservationRegistry} to use for recording observations. + * + * @param observationRegistry must not be {@literal null}. + * @since 6.1 + */ + public void setObservationRegistry(ObservationRegistry observationRegistry) { + + Assert.notNull(observationRegistry, "observationRegistry must not be null"); + + this.observationRegistry = observationRegistry; + } + + /** + * Hook for subclasses to copy additional state during {@link #copy()}. Called after all common fields have been + * copied. The default implementation does nothing. + * + * @param copy the new template instance to customize + */ + protected void customizeCopy(AbstractReactiveElasticsearchTemplate copy) { + } + /** * logs the versions of the different Elasticsearch components. * diff --git a/src/test/java/org/springframework/data/elasticsearch/client/elc/DefaultElasticsearchObservationConventionTests.java b/src/test/java/org/springframework/data/elasticsearch/client/elc/DefaultElasticsearchObservationConventionTests.java new file mode 100644 index 000000000..366fb876e --- /dev/null +++ b/src/test/java/org/springframework/data/elasticsearch/client/elc/DefaultElasticsearchObservationConventionTests.java @@ -0,0 +1,202 @@ +/* + * Copyright 2026-present the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.data.elasticsearch.client.elc; + +import static org.assertj.core.api.Assertions.*; + +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates; + +import io.micrometer.common.KeyValue; +import io.micrometer.common.KeyValues; +import io.micrometer.observation.Observation; + +/** + * Unit tests for {@link DefaultElasticsearchObservationConvention}. + * + * @author maryantocinn + */ +class DefaultElasticsearchObservationConventionTests { + + private final DefaultElasticsearchObservationConvention convention = DefaultElasticsearchObservationConvention.INSTANCE; + + @Test + @DisplayName("should return observation name matching ObservationDocumentation") + void shouldReturnCorrectName() { + assertThat(convention.getName()).isEqualTo("spring.data.elasticsearch.command"); + } + + @Test + @DisplayName("should support ElasticsearchObservationContext") + void shouldSupportElasticsearchContext() { + + ElasticsearchObservationContext context = new ElasticsearchObservationContext(ElasticsearchOperationName.SEARCH, + IndexCoordinates.of("my-index")); + + assertThat(convention.supportsContext(context)).isTrue(); + } + + @Test + @DisplayName("should not support unrelated context") + void shouldNotSupportUnrelatedContext() { + + Observation.Context otherContext = new Observation.Context(); + + assertThat(convention.supportsContext(otherContext)).isFalse(); + } + + // region contextual name + + @Test + @DisplayName("contextual name should be 'operation index' when index is present") + void contextualNameWithIndex() { + + ElasticsearchObservationContext context = new ElasticsearchObservationContext(ElasticsearchOperationName.SEARCH, + IndexCoordinates.of("products")); + + assertThat(convention.getContextualName(context)).isEqualTo("search products"); + } + + @Test + @DisplayName("contextual name should be just the operation when index is null") + void contextualNameWithoutIndex() { + + ElasticsearchObservationContext context = new ElasticsearchObservationContext(ElasticsearchOperationName.SEARCH, + null); + + assertThat(convention.getContextualName(context)).isEqualTo("search"); + } + + @Test + @DisplayName("contextual name should include comma-joined indices for multi-index operations") + void contextualNameWithMultipleIndices() { + + ElasticsearchObservationContext context = new ElasticsearchObservationContext(ElasticsearchOperationName.SEARCH, + IndexCoordinates.of("index-a", "index-b")); + + assertThat(convention.getContextualName(context)).isEqualTo("search index-a,index-b"); + } + + // endregion + + // region low cardinality key values + + @Test + @DisplayName("should always include spring.data.operation") + void shouldIncludeRequiredKeyValues() { + + ElasticsearchObservationContext context = new ElasticsearchObservationContext(ElasticsearchOperationName.GET, null); + + KeyValues keyValues = convention.getLowCardinalityKeyValues(context); + + assertThat(keyValues).contains(KeyValue.of("spring.data.operation", "get")); + } + + @Test + @DisplayName("should include spring.data.collection when index is present") + void shouldIncludeCollectionWhenIndexPresent() { + + ElasticsearchObservationContext context = new ElasticsearchObservationContext(ElasticsearchOperationName.SEARCH, + IndexCoordinates.of("products")); + + KeyValues keyValues = convention.getLowCardinalityKeyValues(context); + + assertThat(keyValues).contains(KeyValue.of("spring.data.collection", "products")); + } + + @Test + @DisplayName("should not include spring.data.collection when index is null") + void shouldNotIncludeCollectionWhenNull() { + + ElasticsearchObservationContext context = new ElasticsearchObservationContext(ElasticsearchOperationName.SEARCH, + null); + + KeyValues keyValues = convention.getLowCardinalityKeyValues(context); + + assertThat(keyValues.stream().map(KeyValue::getKey)).doesNotContain("spring.data.collection"); + } + + @Test + @DisplayName("should include spring.data.batch.size as high cardinality when batch size is set") + void shouldIncludeBatchSizeAsHighCardinality() { + + ElasticsearchObservationContext context = new ElasticsearchObservationContext(ElasticsearchOperationName.BULK, + IndexCoordinates.of("products")); + context.setBatchSize(5); + + KeyValues highCardValues = convention.getHighCardinalityKeyValues(context); + + assertThat(highCardValues).contains(KeyValue.of("spring.data.batch.size", "5")); + } + + @Test + @DisplayName("should not include spring.data.batch.size in low cardinality key values") + void shouldNotIncludeBatchSizeInLowCardinality() { + + ElasticsearchObservationContext context = new ElasticsearchObservationContext(ElasticsearchOperationName.BULK, + IndexCoordinates.of("products")); + context.setBatchSize(5); + + KeyValues keyValues = convention.getLowCardinalityKeyValues(context); + + assertThat(keyValues.stream().map(KeyValue::getKey)).doesNotContain("spring.data.batch.size"); + } + + @Test + @DisplayName("should return empty high cardinality key values when batch size is null") + void shouldReturnEmptyHighCardinalityWhenNoBatchSize() { + + ElasticsearchObservationContext context = new ElasticsearchObservationContext(ElasticsearchOperationName.SEARCH, + IndexCoordinates.of("products")); + + KeyValues highCardValues = convention.getHighCardinalityKeyValues(context); + + assertThat(highCardValues.stream().map(KeyValue::getKey)).doesNotContain("spring.data.batch.size"); + } + + @Test + @DisplayName("should produce correct key values for a full bulk operation") + void shouldProduceCorrectKeyValuesForBulk() { + + ElasticsearchObservationContext context = new ElasticsearchObservationContext(ElasticsearchOperationName.BULK, + IndexCoordinates.of("orders")); + context.setBatchSize(100); + + KeyValues lowCardValues = convention.getLowCardinalityKeyValues(context); + KeyValues highCardValues = convention.getHighCardinalityKeyValues(context); + + assertThat(lowCardValues).contains(KeyValue.of("spring.data.operation", "bulk"), + KeyValue.of("spring.data.collection", "orders")); + assertThat(highCardValues).contains(KeyValue.of("spring.data.batch.size", "100")); + } + + @Test + @DisplayName("should produce correct key values for each operation name") + void shouldProduceCorrectOperationNames() { + + for (ElasticsearchOperationName operationName : ElasticsearchOperationName.values()) { + ElasticsearchObservationContext context = new ElasticsearchObservationContext(operationName, + IndexCoordinates.of("test-index")); + + KeyValues keyValues = convention.getLowCardinalityKeyValues(context); + + assertThat(keyValues).contains(KeyValue.of("spring.data.operation", operationName.getValue())); + } + } + + // endregion +} diff --git a/src/test/java/org/springframework/data/elasticsearch/client/elc/ElasticsearchObservationContextTests.java b/src/test/java/org/springframework/data/elasticsearch/client/elc/ElasticsearchObservationContextTests.java new file mode 100644 index 000000000..dbaa4f754 --- /dev/null +++ b/src/test/java/org/springframework/data/elasticsearch/client/elc/ElasticsearchObservationContextTests.java @@ -0,0 +1,91 @@ +/* + * Copyright 2026-present the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.data.elasticsearch.client.elc; + +import static org.assertj.core.api.Assertions.*; + +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates; + +/** + * Unit tests for {@link ElasticsearchObservationContext}. + * + * @author maryantocinn + */ +class ElasticsearchObservationContextTests { + + @Test + @DisplayName("should carry operation name and index coordinates") + void shouldCarryOperationNameAndIndex() { + + IndexCoordinates index = IndexCoordinates.of("my-index"); + ElasticsearchObservationContext context = new ElasticsearchObservationContext(ElasticsearchOperationName.SEARCH, + index); + + assertThat(context.getOperationName()).isEqualTo(ElasticsearchOperationName.SEARCH); + assertThat(context.getIndexCoordinates()).isEqualTo(index); + assertThat(context.getIndexName()).isEqualTo("my-index"); + } + + @Test + @DisplayName("should return null index name when index coordinates are null") + void shouldReturnNullIndexNameWhenNull() { + + ElasticsearchObservationContext context = new ElasticsearchObservationContext(ElasticsearchOperationName.SEARCH, + null); + + assertThat(context.getIndexCoordinates()).isNull(); + assertThat(context.getIndexName()).isNull(); + } + + @Test + @DisplayName("should join multiple index names with comma") + void shouldJoinMultipleIndexNames() { + + IndexCoordinates index = IndexCoordinates.of("index-a", "index-b", "index-c"); + ElasticsearchObservationContext context = new ElasticsearchObservationContext(ElasticsearchOperationName.SEARCH, + index); + + assertThat(context.getIndexName()).isEqualTo("index-a,index-b,index-c"); + } + + @Test + @DisplayName("should store and retrieve batch size") + void shouldStoreAndRetrieveBatchSize() { + + ElasticsearchObservationContext context = new ElasticsearchObservationContext(ElasticsearchOperationName.BULK, + IndexCoordinates.of("my-index")); + + assertThat(context.getBatchSize()).isNull(); + + context.setBatchSize(42); + + assertThat(context.getBatchSize()).isEqualTo(42); + } + + @Test + @DisplayName("should allow null batch size") + void shouldAllowNullBatchSize() { + + ElasticsearchObservationContext context = new ElasticsearchObservationContext(ElasticsearchOperationName.BULK, + IndexCoordinates.of("my-index")); + context.setBatchSize(10); + context.setBatchSize(null); + + assertThat(context.getBatchSize()).isNull(); + } +} diff --git a/src/test/java/org/springframework/data/elasticsearch/client/elc/ElasticsearchObservationDocumentationTests.java b/src/test/java/org/springframework/data/elasticsearch/client/elc/ElasticsearchObservationDocumentationTests.java new file mode 100644 index 000000000..0f100dcfe --- /dev/null +++ b/src/test/java/org/springframework/data/elasticsearch/client/elc/ElasticsearchObservationDocumentationTests.java @@ -0,0 +1,207 @@ +/* + * Copyright 2026-present the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.data.elasticsearch.client.elc; + +import static org.assertj.core.api.Assertions.*; + +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates; + +import java.util.Arrays; +import java.util.List; +import java.util.stream.Collectors; + +import io.micrometer.common.docs.KeyName; +import io.micrometer.observation.Observation; +import io.micrometer.observation.ObservationRegistry; +import io.micrometer.observation.tck.TestObservationRegistry; +import io.micrometer.observation.tck.TestObservationRegistryAssert; + +/** + * Tests for {@link ElasticsearchObservation} and the end-to-end observation lifecycle. + * + * @author maryantocinn + */ +class ElasticsearchObservationDocumentationTests { + + @Test + @DisplayName("observation name should be 'spring.data.elasticsearch.command'") + void shouldHaveCorrectObservationName() { + assertThat(ElasticsearchObservation.ELASTICSEARCH_COMMAND_OBSERVATION.getName()).isEqualTo( + "spring.data.elasticsearch.command"); + } + + @Test + @DisplayName("default convention should be DefaultElasticsearchObservationConvention") + void shouldHaveCorrectDefaultConvention() { + assertThat(ElasticsearchObservation.ELASTICSEARCH_COMMAND_OBSERVATION.getDefaultConvention()).isEqualTo( + DefaultElasticsearchObservationConvention.class); + } + + @Test + @DisplayName("should declare all expected low cardinality key names") + void shouldDeclareExpectedLowCardinalityKeyNames() { + + KeyName[] keyNames = ElasticsearchObservation.ELASTICSEARCH_COMMAND_OBSERVATION.getLowCardinalityKeyNames(); + List keyStrings = Arrays.stream(keyNames).map(KeyName::asString).collect(Collectors.toList()); + + assertThat(keyStrings).containsExactlyInAnyOrder("spring.data.operation", "spring.data.collection"); + } + + @Test + @DisplayName("should declare spring.data.batch.size as high cardinality key name") + void shouldDeclareHighCardinalityKeyNames() { + + KeyName[] keyNames = ElasticsearchObservation.ELASTICSEARCH_COMMAND_OBSERVATION.getHighCardinalityKeyNames(); + List keyStrings = Arrays.stream(keyNames).map(KeyName::asString).collect(Collectors.toList()); + + assertThat(keyStrings).containsExactly("spring.data.batch.size"); + } + + @Test + @DisplayName("should record observation with correct key values using TestObservationRegistry") + void shouldRecordObservationWithKeyValues() { + + TestObservationRegistry registry = TestObservationRegistry.create(); + + ElasticsearchObservationContext context = new ElasticsearchObservationContext(ElasticsearchOperationName.SEARCH, + IndexCoordinates.of("products")); + + Observation observation = ElasticsearchObservation.ELASTICSEARCH_COMMAND_OBSERVATION.observation(null, + DefaultElasticsearchObservationConvention.INSTANCE, () -> context, registry); + + observation.start(); + observation.stop(); + + TestObservationRegistryAssert.assertThat(registry).doesNotHaveAnyRemainingCurrentObservation() + .hasObservationWithNameEqualTo("spring.data.elasticsearch.command").that() + .hasLowCardinalityKeyValue("spring.data.operation", "search") + .hasLowCardinalityKeyValue("spring.data.collection", "products").hasContextualNameEqualTo("search products"); + } + + @Test + @DisplayName("should record observation without index when null") + void shouldRecordObservationWithoutIndex() { + + TestObservationRegistry registry = TestObservationRegistry.create(); + + ElasticsearchObservationContext context = new ElasticsearchObservationContext(ElasticsearchOperationName.DELETE, + null); + + Observation observation = ElasticsearchObservation.ELASTICSEARCH_COMMAND_OBSERVATION.observation(null, + DefaultElasticsearchObservationConvention.INSTANCE, () -> context, registry); + + observation.start(); + observation.stop(); + + TestObservationRegistryAssert.assertThat(registry) + .hasObservationWithNameEqualTo("spring.data.elasticsearch.command").that() + .hasLowCardinalityKeyValue("spring.data.operation", "delete") + .doesNotHaveLowCardinalityKeyValue("spring.data.collection", "products").hasContextualNameEqualTo("delete"); + } + + @Test + @DisplayName("should record observation with batch size for bulk operations") + void shouldRecordObservationWithBatchSize() { + + TestObservationRegistry registry = TestObservationRegistry.create(); + + ElasticsearchObservationContext context = new ElasticsearchObservationContext(ElasticsearchOperationName.BULK, + IndexCoordinates.of("logs")); + context.setBatchSize(25); + + Observation observation = ElasticsearchObservation.ELASTICSEARCH_COMMAND_OBSERVATION.observation(null, + DefaultElasticsearchObservationConvention.INSTANCE, () -> context, registry); + + observation.start(); + observation.stop(); + + TestObservationRegistryAssert.assertThat(registry) + .hasObservationWithNameEqualTo("spring.data.elasticsearch.command").that() + .hasLowCardinalityKeyValue("spring.data.operation", "bulk") + .hasHighCardinalityKeyValue("spring.data.batch.size", "25").hasContextualNameEqualTo("bulk logs"); + } + + @Test + @DisplayName("should record error on observation") + void shouldRecordErrorOnObservation() { + + TestObservationRegistry registry = TestObservationRegistry.create(); + + ElasticsearchObservationContext context = new ElasticsearchObservationContext(ElasticsearchOperationName.SEARCH, + IndexCoordinates.of("products")); + + Observation observation = ElasticsearchObservation.ELASTICSEARCH_COMMAND_OBSERVATION.observation(null, + DefaultElasticsearchObservationConvention.INSTANCE, () -> context, registry); + + RuntimeException error = new RuntimeException("connection refused"); + observation.start(); + observation.error(error); + observation.stop(); + + TestObservationRegistryAssert.assertThat(registry) + .hasObservationWithNameEqualTo("spring.data.elasticsearch.command").that() + .hasLowCardinalityKeyValue("spring.data.operation", "search").hasBeenStopped(); + } + + @Test + @DisplayName("NOOP registry should not record any observations") + void noopRegistryShouldNotRecord() { + + ElasticsearchObservationContext context = new ElasticsearchObservationContext(ElasticsearchOperationName.SEARCH, + IndexCoordinates.of("products")); + + Observation observation = ElasticsearchObservation.ELASTICSEARCH_COMMAND_OBSERVATION.observation(null, + DefaultElasticsearchObservationConvention.INSTANCE, () -> context, ObservationRegistry.NOOP); + + observation.start(); + observation.stop(); + + assertThat(observation.isNoop()).isTrue(); + } + + @Test + @DisplayName("custom convention should override default key values") + void customConventionShouldOverride() { + + TestObservationRegistry registry = TestObservationRegistry.create(); + + ElasticsearchObservationConvention customConvention = new ElasticsearchObservationConvention() { + @Override + public String getName() { + return "custom.elasticsearch.command"; + } + + @Override + public String getContextualName(ElasticsearchObservationContext context) { + return "custom " + context.getOperationName().getValue(); + } + }; + + ElasticsearchObservationContext context = new ElasticsearchObservationContext(ElasticsearchOperationName.SEARCH, + IndexCoordinates.of("products")); + + Observation observation = ElasticsearchObservation.ELASTICSEARCH_COMMAND_OBSERVATION.observation(customConvention, + DefaultElasticsearchObservationConvention.INSTANCE, () -> context, registry); + + observation.start(); + observation.stop(); + + TestObservationRegistryAssert.assertThat(registry).hasObservationWithNameEqualTo("custom.elasticsearch.command") + .that().hasContextualNameEqualTo("custom search"); + } +} diff --git a/src/test/java/org/springframework/data/elasticsearch/client/elc/ObservabilityIntegrationTests.java b/src/test/java/org/springframework/data/elasticsearch/client/elc/ObservabilityIntegrationTests.java new file mode 100644 index 000000000..4fcb36925 --- /dev/null +++ b/src/test/java/org/springframework/data/elasticsearch/client/elc/ObservabilityIntegrationTests.java @@ -0,0 +1,164 @@ +/* + * Copyright 2026-present the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.data.elasticsearch.client.elc; + +import static org.assertj.core.api.Assertions.*; + +import io.micrometer.observation.tck.TestObservationRegistry; +import io.micrometer.observation.tck.TestObservationRegistryAssert; + +import org.jspecify.annotations.Nullable; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Order; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Import; +import org.springframework.data.annotation.Id; +import org.springframework.data.elasticsearch.annotations.Document; +import org.springframework.data.elasticsearch.annotations.Field; +import org.springframework.data.elasticsearch.annotations.FieldType; +import org.springframework.data.elasticsearch.core.ElasticsearchOperations; +import org.springframework.data.elasticsearch.core.SearchHits; +import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates; +import org.springframework.data.elasticsearch.junit.jupiter.ElasticsearchTemplateConfiguration; +import org.springframework.data.elasticsearch.junit.jupiter.SpringIntegrationTest; +import org.springframework.data.elasticsearch.utils.IndexNameProvider; +import org.springframework.test.context.ContextConfiguration; + +/** + * Integration tests verifying that Micrometer observations are recorded for Spring Data Elasticsearch template + * operations when a {@link TestObservationRegistry} is wired into the application context. + * + * @author maryantocinn + * @since 6.1 + */ +@SpringIntegrationTest +@ContextConfiguration(classes = { ObservabilityIntegrationTests.Config.class }) +@DisplayName("Observability Integration Tests") +class ObservabilityIntegrationTests { + + @Configuration + @Import({ ElasticsearchTemplateConfiguration.class }) + static class Config { + + @Bean + IndexNameProvider indexNameProvider() { + return new IndexNameProvider("observability-it"); + } + + @Bean + TestObservationRegistry observationRegistry() { + return TestObservationRegistry.create(); + } + } + + @Autowired private ElasticsearchOperations operations; + @Autowired private IndexNameProvider indexNameProvider; + @Autowired private TestObservationRegistry observationRegistry; + + @BeforeEach + void setUp() { + indexNameProvider.increment(); + operations.indexOps(SampleEntity.class).createWithMapping(); + observationRegistry.clear(); + } + + @Test + @Order(Integer.MAX_VALUE) + void cleanup() { + operations.indexOps(IndexCoordinates.of(indexNameProvider.getPrefix() + "*")).delete(); + } + + @Test + @DisplayName("should record observation for save operation") + void shouldRecordObservationForSave() { + + SampleEntity entity = new SampleEntity(); + entity.setId("1"); + entity.setMessage("hello"); + + operations.save(entity); + + TestObservationRegistryAssert.assertThat(observationRegistry) + .hasObservationWithNameEqualTo("spring.data.elasticsearch.command") + .that() + .hasLowCardinalityKeyValue("spring.data.operation", "save") + .hasBeenStopped(); + } + + @Test + @DisplayName("should record observation for search operation") + void shouldRecordObservationForSearch() { + + SampleEntity entity = new SampleEntity(); + entity.setId("1"); + entity.setMessage("hello"); + operations.save(entity); + observationRegistry.clear(); + + SearchHits hits = operations.search(operations.matchAllQuery(), SampleEntity.class); + + assertThat(hits.getTotalHits()).isGreaterThanOrEqualTo(1); + TestObservationRegistryAssert.assertThat(observationRegistry) + .hasObservationWithNameEqualTo("spring.data.elasticsearch.command") + .that() + .hasLowCardinalityKeyValue("spring.data.operation", "search") + .hasBeenStopped(); + } + + @Test + @DisplayName("should record observation with collection name") + void shouldRecordObservationWithCollectionName() { + + SampleEntity entity = new SampleEntity(); + entity.setId("1"); + entity.setMessage("hello"); + + operations.save(entity); + + TestObservationRegistryAssert.assertThat(observationRegistry) + .hasObservationWithNameEqualTo("spring.data.elasticsearch.command") + .that() + .hasLowCardinalityKeyValue("spring.data.collection", indexNameProvider.indexName()); + } + + @Document(indexName = "#{@indexNameProvider.indexName()}") + static class SampleEntity { + @Nullable @Id private String id; + @Nullable @Field(type = FieldType.Text) private String message; + + @Nullable + public String getId() { + return id; + } + + public void setId(@Nullable String id) { + this.id = id; + } + + @Nullable + public String getMessage() { + return message; + } + + public void setMessage(@Nullable String message) { + this.message = message; + } + } +}