diff --git a/changelog/unreleased/pr-27031.toml b/changelog/unreleased/pr-27031.toml new file mode 100644 index 000000000000..377c52ecf1e6 --- /dev/null +++ b/changelog/unreleased/pr-27031.toml @@ -0,0 +1,5 @@ +type = "fixed" +message = "Fixed bulk indexing requests being dropped instead of retried when OpenSearch returns an incomplete response during shard reallocation." + +issues = ["26853"] +pulls = ["27031"] diff --git a/graylog-storage-opensearch3/src/main/java/org/graylog/storage/opensearch3/MessagesAdapterOS.java b/graylog-storage-opensearch3/src/main/java/org/graylog/storage/opensearch3/MessagesAdapterOS.java index f7f86b6d8609..28a46086fd86 100644 --- a/graylog-storage-opensearch3/src/main/java/org/graylog/storage/opensearch3/MessagesAdapterOS.java +++ b/graylog-storage-opensearch3/src/main/java/org/graylog/storage/opensearch3/MessagesAdapterOS.java @@ -19,7 +19,9 @@ import com.codahale.metrics.Meter; import com.codahale.metrics.MetricRegistry; import com.fasterxml.jackson.databind.ObjectMapper; +import com.google.common.base.Throwables; import jakarta.inject.Inject; +import org.graylog2.indexer.IncompleteBulkResponseException; import org.graylog2.indexer.messages.ChunkedBulkIndexer; import org.graylog2.indexer.messages.DocumentNotFoundException; import org.graylog2.indexer.messages.Indexable; @@ -36,6 +38,7 @@ import org.opensearch.client.opensearch._types.ErrorCause; import org.opensearch.client.opensearch._types.ErrorResponse; import org.opensearch.client.opensearch._types.OpenSearchException; +import org.opensearch.client.opensearch._types.ShardSearchFailure; import org.opensearch.client.opensearch.core.BulkRequest; import org.opensearch.client.opensearch.core.BulkResponse; import org.opensearch.client.opensearch.core.GetRequest; @@ -46,6 +49,7 @@ import org.opensearch.client.opensearch.indices.AnalyzeResponse; import org.opensearch.client.opensearch.indices.analyze.AnalyzeToken; import org.opensearch.client.transport.httpclient5.ResponseException; +import org.opensearch.client.util.MissingRequiredPropertyException; import java.io.IOException; import java.util.Arrays; @@ -144,10 +148,35 @@ private ChunkedBulkIndexer.BulkIndexResult runBulkRequest(int indexedSuccessfull throw new org.graylog2.indexer.ElasticsearchException(e); } catch (IOException e) { throw new org.graylog2.indexer.ElasticsearchException(e); + } catch (RuntimeException e) { + if (isIncompleteShardFailureResponse(e)) { + // Can happen while shards are being reallocated during a rolling restart: OpenSearch omits the + // "shard" field on a shard failure entry, which the opensearch-java 3.x client fails to parse. + // Messages#createBulkRequestRetryerBuilder() retries this exception type explicitly. + throw new IncompleteBulkResponseException( + "Received an incomplete bulk response from OpenSearch, this can happen while shards are being reallocated", + e); + } + throw e; } return new ChunkedBulkIndexer.BulkIndexResult(indexingResultsFrom(result, chunk), () -> buildFailureMessage(result), result.items().size()); } + // The opensearch-java client throws this whenever ANY response it deserializes is missing a field it + // considers required, so we only treat it as our known, retryable rolling-restart scenario when the + // missing property is specifically ShardSearchFailure#shard. Anything else (a different missing field, + // or a genuinely incompatible/broken response) should keep propagating instead of being retried forever. + // Note: this is the same "_shards.failures[]" schema field that was named ShardFailure#shard in + // opensearch-java <= 3.3.0; the client renamed the generated class to ShardSearchFailure afterwards. If + // opensearch.client.version is bumped and this stops matching, MessagesAdapterOSTest will fail loudly. + private boolean isIncompleteShardFailureResponse(RuntimeException e) { + return Throwables.getCausalChain(e).stream() + .filter(MissingRequiredPropertyException.class::isInstance) + .map(MissingRequiredPropertyException.class::cast) + .anyMatch(missingProperty -> ShardSearchFailure.class.equals(missingProperty.getObjectClass()) + && "shard".equals(missingProperty.getPropertyName())); + } + private OpenSearchException toOpenSearchException(ResponseException re) { String[] split = re.getMessage().split("\n"); if (split.length != 2) { diff --git a/graylog-storage-opensearch3/src/test/java/org/graylog/storage/opensearch3/MessagesAdapterOSTest.java b/graylog-storage-opensearch3/src/test/java/org/graylog/storage/opensearch3/MessagesAdapterOSTest.java new file mode 100644 index 000000000000..c62593a9174c --- /dev/null +++ b/graylog-storage-opensearch3/src/test/java/org/graylog/storage/opensearch3/MessagesAdapterOSTest.java @@ -0,0 +1,144 @@ +/* + * Copyright (C) 2020 Graylog, Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the Server Side Public License, version 1, + * as published by MongoDB, Inc. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Server Side Public License for more details. + * + * You should have received a copy of the Server Side Public License + * along with this program. If not, see + * . + */ +package org.graylog.storage.opensearch3; + +import com.codahale.metrics.MetricRegistry; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.graylog.storage.opensearch3.testing.client.mock.ServerlessOpenSearchClient; +import org.graylog2.indexer.IncompleteBulkResponseException; +import org.graylog2.indexer.messages.ChunkedBulkIndexer; +import org.graylog2.indexer.messages.Indexable; +import org.graylog2.indexer.messages.IndexingRequest; +import org.graylog2.indexer.results.ResultMessageFactory; +import org.graylog2.shared.utilities.ExceptionUtils; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; +import org.opensearch.client.util.MissingRequiredPropertyException; + +import java.util.List; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +@ExtendWith(MockitoExtension.class) +class MessagesAdapterOSTest { + + @Mock + private ResultMessageFactory resultMessageFactory; + + /** + * Reproduces a rolling restart where a replica shard is still being promoted: OpenSearch reports a shard-level + * write failure whose {@code _shards.failures[]} entry omits the "shard" number. The opensearch-java client + * treats that as a required property and refuses to parse the response at all. + */ + private static final String BULK_RESPONSE_WITH_INCOMPLETE_SHARD_FAILURE = """ + { + "took": 1, + "errors": true, + "items": [ + { + "index": { + "_index": "graylog_0", + "_id": "message-id", + "status": 200, + "_shards": { + "total": 2, + "successful": 1, + "failed": 1, + "failures": [ + { + "index": "graylog_0", + "node": "node-1", + "reason": { + "type": "unavailable_shards_exception", + "reason": "primary shard is not active" + } + } + ] + } + } + } + ] + } + """; + + /** + * A missing "status" is a required-property failure too, but it has nothing to do with shard reallocation. + * It must not be mistaken for the scenario above and silently retried forever. + */ + private static final String BULK_RESPONSE_MISSING_UNRELATED_REQUIRED_PROPERTY = """ + { + "took": 1, + "errors": true, + "items": [ + { + "index": { + "_index": "graylog_0", + "_id": "message-id" + } + } + ] + } + """; + + @Test + void bulkIndexTreatsIncompleteShardFailureResponseAsRetryable() { + final OfficialOpensearchClient officialOpensearchClient = ServerlessOpenSearchClient.builder() + .stubResponse("POST", "/_bulk", BULK_RESPONSE_WITH_INCOMPLETE_SHARD_FAILURE) + .build(); + + final MessagesAdapterOS messagesAdapterOS = new MessagesAdapterOS(resultMessageFactory, officialOpensearchClient, + new MetricRegistry(), new ChunkedBulkIndexer(), new ObjectMapper()); + + final Indexable message = mock(Indexable.class); + when(message.getId()).thenReturn("message-id"); + when(message.toElasticSearchObject(any(), any())).thenReturn(Map.of("message", "test")); + + final List request = List.of(IndexingRequest.create("graylog_0", message)); + + assertThatThrownBy(() -> messagesAdapterOS.bulkIndex(request)) + .isInstanceOf(IncompleteBulkResponseException.class) + .satisfies(e -> assertThat(ExceptionUtils.hasCauseOf(e, MissingRequiredPropertyException.class)).isTrue()); + } + + @Test + void bulkIndexDoesNotTreatUnrelatedMissingPropertyAsRetryable() { + final OfficialOpensearchClient officialOpensearchClient = ServerlessOpenSearchClient.builder() + .stubResponse("POST", "/_bulk", BULK_RESPONSE_MISSING_UNRELATED_REQUIRED_PROPERTY) + .build(); + + final MessagesAdapterOS messagesAdapterOS = new MessagesAdapterOS(resultMessageFactory, officialOpensearchClient, + new MetricRegistry(), new ChunkedBulkIndexer(), new ObjectMapper()); + + final Indexable message = mock(Indexable.class); + when(message.getId()).thenReturn("message-id"); + when(message.toElasticSearchObject(any(), any())).thenReturn(Map.of("message", "test")); + + final List request = List.of(IndexingRequest.create("graylog_0", message)); + + // Must propagate as the original, unwrapped exception rather than being turned into a retryable one. + assertThatThrownBy(() -> messagesAdapterOS.bulkIndex(request)) + .isInstanceOf(MissingRequiredPropertyException.class) + .hasMessageContaining("status"); + } +} diff --git a/graylog2-server/src/main/java/org/graylog2/indexer/IncompleteBulkResponseException.java b/graylog2-server/src/main/java/org/graylog2/indexer/IncompleteBulkResponseException.java new file mode 100644 index 000000000000..327c8611ec4b --- /dev/null +++ b/graylog2-server/src/main/java/org/graylog2/indexer/IncompleteBulkResponseException.java @@ -0,0 +1,34 @@ +/* + * Copyright (C) 2020 Graylog, Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the Server Side Public License, version 1, + * as published by MongoDB, Inc. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Server Side Public License for more details. + * + * You should have received a copy of the Server Side Public License + * along with this program. If not, see + * . + */ +package org.graylog2.indexer; + +/** + * Thrown by a {@code MessagesAdapter} when an indexer's bulk response cannot be parsed because it omitted + * shard-failure detail while a shard was being reallocated, e.g. during a rolling restart. Retried like other + * transient indexing failures, see {@code Messages#createBulkRequestRetryerBuilder()}. + *

+ * See opensearch-java#551: the + * client treats the shard number on a shard-failure entry as a required field, but OpenSearch can omit it while + * the failing shard is still being promoted. A client-side fix is proposed in + * opensearch-java#2023 (unmerged + * as of this writing); once a released client version includes it, this workaround can likely be removed. + */ +public class IncompleteBulkResponseException extends ElasticsearchException { + public IncompleteBulkResponseException(String message, Throwable cause) { + super(message, cause); + } +} diff --git a/graylog2-server/src/main/java/org/graylog2/indexer/messages/Messages.java b/graylog2-server/src/main/java/org/graylog2/indexer/messages/Messages.java index 487ab21e1c0f..75837dd5c19c 100644 --- a/graylog2-server/src/main/java/org/graylog2/indexer/messages/Messages.java +++ b/graylog2-server/src/main/java/org/graylog2/indexer/messages/Messages.java @@ -26,6 +26,7 @@ import jakarta.inject.Inject; import jakarta.inject.Singleton; import org.graylog.failure.FailureSubmissionService; +import org.graylog2.indexer.IncompleteBulkResponseException; import org.graylog2.indexer.InvalidWriteTargetException; import org.graylog2.indexer.MasterNotDiscoveredException; import org.graylog2.indexer.results.ResultMessage; @@ -67,7 +68,8 @@ private RetryerBuilder createBulkRequestRetryerBuilder() { return RetryerBuilder.newBuilder() .retryIfException(t -> ExceptionUtils.hasCauseOf(t, IOException.class) || t instanceof InvalidWriteTargetException - || t instanceof MasterNotDiscoveredException) + || t instanceof MasterNotDiscoveredException + || t instanceof IncompleteBulkResponseException) .withWaitStrategy(WaitStrategies.exponentialWait(MAX_WAIT_TIME.getQuantity(), MAX_WAIT_TIME.getUnit())) .withRetryListener(new RetryListener() { @Override diff --git a/graylog2-server/src/test/java/org/graylog2/indexer/messages/MessagesTest.java b/graylog2-server/src/test/java/org/graylog2/indexer/messages/MessagesTest.java index f863f91b9ea6..b640bc28cbb5 100644 --- a/graylog2-server/src/test/java/org/graylog2/indexer/messages/MessagesTest.java +++ b/graylog2-server/src/test/java/org/graylog2/indexer/messages/MessagesTest.java @@ -18,6 +18,8 @@ import com.google.common.collect.ImmutableList; import org.graylog.failure.FailureSubmissionService; +import org.graylog2.indexer.ElasticsearchException; +import org.graylog2.indexer.IncompleteBulkResponseException; import org.graylog2.plugin.Message; import org.graylog2.plugin.MessageFactory; import org.graylog2.plugin.TestMessageFactory; @@ -35,6 +37,7 @@ import org.mockito.quality.Strictness; import java.io.IOException; +import java.net.SocketTimeoutException; import java.util.Collection; import java.util.Collections; import java.util.Comparator; @@ -250,6 +253,58 @@ public void bulkIndexRequests_nothingPropagatedToFailureSubmissionServiceWhenThe verifyNoInteractions(failureSubmissionService); } + @Test + public void bulkIndexRequests_retriesWhenAdapterThrowsExceptionWithIOExceptionCause() throws Exception { + // given + final DateTime ts = Tools.nowUTC(); + final Message message1 = message("msg-1", ts); + + final List indexingRequest = ImmutableList.of(IndexingRequest.create("", message1)); + + // A generic transient transport failure (e.g. a real connection problem), as any MessagesAdapter might + // throw it, unrelated to the more specific IncompleteBulkResponseException case tested below. + final ElasticsearchException transientFailure = new ElasticsearchException( + "Could not reach the indexer", new SocketTimeoutException("connect timed out")); + + when(messagesAdapter.bulkIndex(indexingRequest)) + .thenThrow(transientFailure) + .thenReturn(IndexingResults.create(List.of(new IndexingSuccess(message1, "index_1")), List.of())); + + // when + final IndexingResults indexingResults = messages.bulkIndexRequests(indexingRequest, false); + + // then + verify(messagesAdapter, times(2)).bulkIndex(indexingRequest); + assertThat(indexingResults.errors()).isEmpty(); + assertThat(indexingResults.successes()).hasSize(1); + } + + @Test + public void bulkIndexRequests_retriesWhenAdapterThrowsIncompleteBulkResponseException() throws Exception { + // given + final DateTime ts = Tools.nowUTC(); + final Message message1 = message("msg-1", ts); + + final List indexingRequest = ImmutableList.of(IndexingRequest.create("", message1)); + + // What MessagesAdapterOS throws when OpenSearch returns an incomplete bulk response (e.g. a + // "_shards.failures[]" entry missing its "shard" field) while shards are being reallocated. + final IncompleteBulkResponseException transientFailure = new IncompleteBulkResponseException( + "Received an incomplete bulk response from OpenSearch", new RuntimeException("missing property")); + + when(messagesAdapter.bulkIndex(indexingRequest)) + .thenThrow(transientFailure) + .thenReturn(IndexingResults.create(List.of(new IndexingSuccess(message1, "index_1")), List.of())); + + // when + final IndexingResults indexingResults = messages.bulkIndexRequests(indexingRequest, false); + + // then + verify(messagesAdapter, times(2)).bulkIndex(indexingRequest); + assertThat(indexingResults.errors()).isEmpty(); + assertThat(indexingResults.successes()).hasSize(1); + } + private List createSuccessFromMessages(List messageList) { return messageList.stream().map(m -> new IndexingSuccess(m.message(), "index_2")).collect(Collectors.toList()); }