Skip to content

Commit b09f1d7

Browse files
martijnvgmartijnvg
authored
Add additional debug logging for indexing failure. (#117728)
This change adds a new IndexingOperationListener that logs debug logging for indexing failures. Today `IndexShard` logger needs to be set to TRACE in order to see indexing failures. This logger is very verbose and logs many other aspects of indexing at trace level. This new logger is targeted just for seeing indexing failures. Relates #117700
1 parent 5b5c9ad commit b09f1d7

File tree

3 files changed

+194
-1
lines changed

3 files changed

+194
-1
lines changed

‎server/src/main/java/org/elasticsearch/index/shard/IndexShard.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -345,8 +345,9 @@ public IndexShard(
345345
this.mapperService = mapperService;
346346
this.indexCache = indexCache;
347347
this.internalIndexingStats = new InternalIndexingStats();
348+
var indexingFailuresDebugListener = new IndexingFailuresDebugListener(this);
348349
this.indexingOperationListeners = new IndexingOperationListener.CompositeListener(
349-
CollectionUtils.appendToCopyNoNullElements(listeners, internalIndexingStats),
350+
CollectionUtils.appendToCopyNoNullElements(listeners, internalIndexingStats, indexingFailuresDebugListener),
350351
logger
351352
);
352353
this.bulkOperationListener = new ShardBulkStats();

‎server/src/main/java/org/elasticsearch/index/shard/IndexingFailuresDebugListener.java‎

Lines changed: 54 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,54 @@
1+
/*
2+
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
3+
* or more contributor license agreements. Licensed under the "Elastic License
4+
* 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
5+
* Public License v 1"; you may not use this file except in compliance with, at
6+
* your election, the "Elastic License 2.0", the "GNU Affero General Public
7+
* License v3.0 only", or the "Server Side Public License, v 1".
8+
*/
9+
10+
package org.elasticsearch.index.shard;
11+
12+
import org.apache.logging.log4j.LogManager;
13+
import org.apache.logging.log4j.Logger;
14+
import org.elasticsearch.index.engine.Engine;
15+
16+
import static org.elasticsearch.core.Strings.format;
17+
18+
public class IndexingFailuresDebugListener implements IndexingOperationListener {
19+
20+
private static final Logger LOGGER = LogManager.getLogger(IndexingFailuresDebugListener.class);
21+
22+
private final IndexShard indexShard;
23+
24+
public IndexingFailuresDebugListener(IndexShard indexShard) {
25+
this.indexShard = indexShard;
26+
}
27+
28+
@Override
29+
public void postIndex(ShardId shardId, Engine.Index index, Engine.IndexResult result) {
30+
if (LOGGER.isDebugEnabled()) {
31+
if (result.getResultType() == Engine.Result.Type.FAILURE) {
32+
postIndex(shardId, index, result.getFailure());
33+
}
34+
}
35+
}
36+
37+
@Override
38+
public void postIndex(ShardId shardId, Engine.Index index, Exception ex) {
39+
if (LOGGER.isDebugEnabled()) {
40+
LOGGER.debug(
41+
() -> format(
42+
"index-fail [%s] seq# [%s] allocation-id [%s] primaryTerm [%s] operationPrimaryTerm [%s] origin [%s]",
43+
index.id(),
44+
index.seqNo(),
45+
indexShard.routingEntry().allocationId(),
46+
index.primaryTerm(),
47+
indexShard.getOperationPrimaryTerm(),
48+
index.origin()
49+
),
50+
ex
51+
);
52+
}
53+
}
54+
}

‎server/src/test/java/org/elasticsearch/index/shard/IndexingFailuresDebugListenerTests.java‎

Lines changed: 138 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,138 @@
1+
/*
2+
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
3+
* or more contributor license agreements. Licensed under the "Elastic License
4+
* 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
5+
* Public License v 1"; you may not use this file except in compliance with, at
6+
* your election, the "Elastic License 2.0", the "GNU Affero General Public
7+
* License v3.0 only", or the "Server Side Public License, v 1".
8+
*/
9+
10+
package org.elasticsearch.index.shard;
11+
12+
import org.apache.logging.log4j.Level;
13+
import org.apache.logging.log4j.LogManager;
14+
import org.apache.logging.log4j.Logger;
15+
import org.elasticsearch.cluster.routing.ShardRoutingState;
16+
import org.elasticsearch.cluster.routing.TestShardRouting;
17+
import org.elasticsearch.common.logging.Loggers;
18+
import org.elasticsearch.common.logging.MockAppender;
19+
import org.elasticsearch.index.engine.Engine;
20+
import org.elasticsearch.index.engine.EngineTestCase;
21+
import org.elasticsearch.index.mapper.ParsedDocument;
22+
import org.elasticsearch.index.mapper.Uid;
23+
import org.elasticsearch.test.ESTestCase;
24+
import org.junit.AfterClass;
25+
import org.junit.BeforeClass;
26+
27+
import static org.hamcrest.Matchers.equalTo;
28+
import static org.hamcrest.Matchers.nullValue;
29+
import static org.mockito.Mockito.mock;
30+
import static org.mockito.Mockito.when;
31+
32+
public class IndexingFailuresDebugListenerTests extends ESTestCase {
33+
34+
static MockAppender appender;
35+
static Logger testLogger1 = LogManager.getLogger(IndexingFailuresDebugListener.class);
36+
static Level origLogLevel = testLogger1.getLevel();
37+
38+
@BeforeClass
39+
public static void init() throws IllegalAccessException {
40+
appender = new MockAppender("mock_appender");
41+
appender.start();
42+
Loggers.addAppender(testLogger1, appender);
43+
Loggers.setLevel(testLogger1, randomBoolean() ? Level.DEBUG : Level.TRACE);
44+
}
45+
46+
@AfterClass
47+
public static void cleanup() {
48+
Loggers.removeAppender(testLogger1, appender);
49+
appender.stop();
50+
51+
Loggers.setLevel(testLogger1, origLogLevel);
52+
}
53+
54+
public void testPostIndexException() {
55+
var shardId = ShardId.fromString("[index][123]");
56+
var mockShard = mock(IndexShard.class);
57+
var shardRouting = TestShardRouting.newShardRouting(shardId, "node-id", true, ShardRoutingState.STARTED);
58+
when(mockShard.routingEntry()).thenReturn(shardRouting);
59+
when(mockShard.getOperationPrimaryTerm()).thenReturn(1L);
60+
IndexingFailuresDebugListener indexingFailuresDebugListener = new IndexingFailuresDebugListener(mockShard);
61+
62+
ParsedDocument doc = EngineTestCase.createParsedDoc("1", null);
63+
Engine.Index index = new Engine.Index(Uid.encodeId("doc_id"), 1, doc);
64+
indexingFailuresDebugListener.postIndex(shardId, index, new RuntimeException("test exception"));
65+
String message = appender.getLastEventAndReset().getMessage().getFormattedMessage();
66+
assertThat(
67+
message,
68+
equalTo(
69+
"index-fail [1] seq# [-2] allocation-id ["
70+
+ shardRouting.allocationId()
71+
+ "] primaryTerm [1] operationPrimaryTerm [1] origin [PRIMARY]"
72+
)
73+
);
74+
}
75+
76+
public void testPostIndexExceptionInfoLevel() {
77+
var previousLevel = testLogger1.getLevel();
78+
try {
79+
Loggers.setLevel(testLogger1, randomBoolean() ? Level.INFO : Level.WARN);
80+
var shardId = ShardId.fromString("[index][123]");
81+
var mockShard = mock(IndexShard.class);
82+
var shardRouting = TestShardRouting.newShardRouting(shardId, "node-id", true, ShardRoutingState.STARTED);
83+
when(mockShard.routingEntry()).thenReturn(shardRouting);
84+
when(mockShard.getOperationPrimaryTerm()).thenReturn(1L);
85+
IndexingFailuresDebugListener indexingFailuresDebugListener = new IndexingFailuresDebugListener(mockShard);
86+
87+
ParsedDocument doc = EngineTestCase.createParsedDoc("1", null);
88+
Engine.Index index = new Engine.Index(Uid.encodeId("doc_id"), 1, doc);
89+
indexingFailuresDebugListener.postIndex(shardId, index, new RuntimeException("test exception"));
90+
assertThat(appender.getLastEventAndReset(), nullValue());
91+
} finally {
92+
Loggers.setLevel(testLogger1, previousLevel);
93+
}
94+
}
95+
96+
public void testPostIndexFailure() {
97+
var shardId = ShardId.fromString("[index][123]");
98+
var mockShard = mock(IndexShard.class);
99+
var shardRouting = TestShardRouting.newShardRouting(shardId, "node-id", true, ShardRoutingState.STARTED);
100+
when(mockShard.routingEntry()).thenReturn(shardRouting);
101+
when(mockShard.getOperationPrimaryTerm()).thenReturn(1L);
102+
IndexingFailuresDebugListener indexingFailuresDebugListener = new IndexingFailuresDebugListener(mockShard);
103+
104+
ParsedDocument doc = EngineTestCase.createParsedDoc("1", null);
105+
Engine.Index index = new Engine.Index(Uid.encodeId("doc_id"), 1, doc);
106+
Engine.IndexResult indexResult = mock(Engine.IndexResult.class);
107+
when(indexResult.getResultType()).thenReturn(Engine.Result.Type.FAILURE);
108+
when(indexResult.getFailure()).thenReturn(new RuntimeException("test exception"));
109+
indexingFailuresDebugListener.postIndex(shardId, index, indexResult);
110+
String message = appender.getLastEventAndReset().getMessage().getFormattedMessage();
111+
assertThat(
112+
message,
113+
equalTo(
114+
"index-fail [1] seq# [-2] allocation-id ["
115+
+ shardRouting.allocationId()
116+
+ "] primaryTerm [1] operationPrimaryTerm [1] origin [PRIMARY]"
117+
)
118+
);
119+
}
120+
121+
public void testPostIndex() {
122+
var shardId = ShardId.fromString("[index][123]");
123+
var mockShard = mock(IndexShard.class);
124+
var shardRouting = TestShardRouting.newShardRouting(shardId, "node-id", true, ShardRoutingState.STARTED);
125+
when(mockShard.routingEntry()).thenReturn(shardRouting);
126+
when(mockShard.getOperationPrimaryTerm()).thenReturn(1L);
127+
IndexingFailuresDebugListener indexingFailuresDebugListener = new IndexingFailuresDebugListener(mockShard);
128+
129+
ParsedDocument doc = EngineTestCase.createParsedDoc("1", null);
130+
Engine.Index index = new Engine.Index(Uid.encodeId("doc_id"), 1, doc);
131+
Engine.IndexResult indexResult = mock(Engine.IndexResult.class);
132+
when(indexResult.getResultType()).thenReturn(Engine.Result.Type.SUCCESS);
133+
when(indexResult.getFailure()).thenReturn(new RuntimeException("test exception"));
134+
indexingFailuresDebugListener.postIndex(shardId, index, indexResult);
135+
assertThat(appender.getLastEventAndReset(), nullValue());
136+
}
137+
138+
}

0 commit comments

Comments
 (0)