Skip to content

Commit 7371e61

Browse files
author
Brett Zeligson
committed
Add Retryable/Non-Retryable Exception + API Calls Metrics for O365
Signed-off-by: Brett Zeligson <brettzel@amazon.com>
1 parent c5190dc commit 7371e61

2 files changed

Lines changed: 99 additions & 26 deletions

File tree

‎data-prepper-plugins/saas-source-plugins/microsoft-office365-source/src/main/java/org/opensearch/dataprepper/plugins/source/microsoft_office365/Office365CrawlerClient.java‎

Lines changed: 26 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -56,6 +56,9 @@ public class Office365CrawlerClient implements CrawlerClient<DimensionalTimeSlic
5656
private static final String BUFFER_WRITE_RETRY_SUCCESS = "bufferWriteRetrySuccess";
5757
private static final String BUFFER_WRITE_RETRY_ATTEMPTS = "bufferWriteRetryAttempts";
5858
private static final String BUFFER_WRITE_FAILURES = "bufferWriteFailures";
59+
private static final String NON_RETRYABLE_ERRORS = "nonRetryableErrors";
60+
private static final String RETRYABLE_ERRORS = "retryableErrors";
61+
private static final String API_CALLS = "apiCalls";
5962
private static final int BUFFER_TIMEOUT_IN_SECONDS = 10;
6063
private static final String CONTENT_ID = "contentId";
6164
private static final String CONTENT_URI = "contentUri";
@@ -68,6 +71,9 @@ public class Office365CrawlerClient implements CrawlerClient<DimensionalTimeSlic
6871
private final Counter bufferWriteRetrySuccessCounter;
6972
private final Counter bufferWriteRetryAttemptsCounter;
7073
private final Counter bufferWriteFailuresCounter;
74+
private final Counter nonRetryableErrorsCounter;
75+
private final Counter retryableErrorsCounter;
76+
private final Counter apiCallsCounter;
7177
private ObjectMapper objectMapper;
7278

7379
public Office365CrawlerClient(final Office365Service service,
@@ -84,6 +90,9 @@ public Office365CrawlerClient(final Office365Service service,
8490
this.bufferWriteRetrySuccessCounter = pluginMetrics.counter(BUFFER_WRITE_RETRY_SUCCESS);
8591
this.bufferWriteRetryAttemptsCounter = pluginMetrics.counter(BUFFER_WRITE_RETRY_ATTEMPTS);
8692
this.bufferWriteFailuresCounter = pluginMetrics.counter(BUFFER_WRITE_FAILURES);
93+
this.nonRetryableErrorsCounter = pluginMetrics.counter(NON_RETRYABLE_ERRORS);
94+
this.retryableErrorsCounter = pluginMetrics.counter(RETRYABLE_ERRORS);
95+
this.apiCallsCounter = pluginMetrics.counter(API_CALLS);
8796
}
8897

8998
@VisibleForTesting
@@ -109,6 +118,7 @@ public void executePartition(final DimensionalTimeSliceWorkerProgressState state
109118
List<Record<Event>> records = new ArrayList<>();
110119

111120
do {
121+
apiCallsCounter.increment();
112122
AuditLogsResponse response =
113123
service.searchAuditLogs(logType, startTime, endTime, nextPageUri);
114124

@@ -121,7 +131,7 @@ public void executePartition(final DimensionalTimeSliceWorkerProgressState state
121131
records.add(record);
122132
}
123133
} catch (Office365Exception e) {
124-
134+
handleExceptionMetrics(e);
125135
log.error(NOISY, "{} error processing audit log: {}",
126136
e.isRetryable() ? "Retryable" : "Non-retryable", logId, e);
127137
if (e.isRetryable()) {
@@ -132,6 +142,7 @@ public void executePartition(final DimensionalTimeSliceWorkerProgressState state
132142
}
133143
} catch (Exception e) {
134144
// Unexpected errors are treated as retryable to be safe
145+
handleExceptionMetrics(e);
135146
log.error(NOISY, "Unexpected error processing audit log: {}", logId, e);
136147
throw new RuntimeException("Unexpected error processing audit log: " + logId, e);
137148
}
@@ -151,18 +162,32 @@ public void executePartition(final DimensionalTimeSliceWorkerProgressState state
151162
});
152163

153164
} catch (Exception e) {
165+
handleExceptionMetrics(e);
154166
log.error(NOISY, "Failed to process partition for log type {} from {} to {}",
155167
logType, startTime, endTime, e);
156168
throw e;
157169
}
158170
}
159171

172+
private void handleExceptionMetrics(Exception e) {
173+
if (e instanceof Office365Exception) {
174+
if (((Office365Exception) e).isRetryable()) {
175+
retryableErrorsCounter.increment();
176+
} else {
177+
nonRetryableErrorsCounter.increment();
178+
}
179+
} else {
180+
retryableErrorsCounter.increment();
181+
}
182+
}
183+
160184
private Record<Event> processAuditLog(Map<String, Object> metadata) throws Office365Exception {
161185
String contentUri = (String) metadata.get(CONTENT_URI);
162186
if (contentUri == null) {
163187
throw new Office365Exception("Missing contentUri in metadata", false);
164188
}
165189

190+
apiCallsCounter.increment();
166191
String logContent = service.getAuditLog(contentUri);
167192
if (logContent == null) {
168193
throw new Office365Exception("Received null log content for URI: " + contentUri, false);

‎data-prepper-plugins/saas-source-plugins/microsoft-office365-source/src/test/java/org/opensearch/dataprepper/plugins/source/microsoft_office365/Office365CrawlerClientTest.java‎

Lines changed: 73 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@
3030
import org.opensearch.dataprepper.plugins.source.source_crawler.coordination.state.DimensionalTimeSliceWorkerProgressState;
3131
import org.opensearch.dataprepper.metrics.PluginMetrics;
3232
import org.opensearch.dataprepper.plugins.source.microsoft_office365.service.Office365Service;
33+
import org.opensearch.dataprepper.plugins.source.microsoft_office365.exception.Office365Exception;
3334
import org.slf4j.Logger;
3435
import org.slf4j.LoggerFactory;
3536

@@ -53,8 +54,10 @@
5354
import static org.mockito.Mockito.doAnswer;
5455
import static org.mockito.Mockito.doThrow;
5556
import static org.mockito.Mockito.mock;
57+
import static org.mockito.Mockito.never;
5658
import static org.mockito.Mockito.verify;
5759
import static org.mockito.Mockito.when;
60+
import static org.mockito.Mockito.times;
5861

5962
@ExtendWith(MockitoExtension.class)
6063
@MockitoSettings(strictness = Strictness.LENIENT)
@@ -80,6 +83,15 @@ class Office365CrawlerClientTest {
8083
@Mock
8184
private Timer bufferWriteLatencyTimer;
8285

86+
@Mock
87+
private Counter apiCallsCounter;
88+
89+
@Mock
90+
private Counter nonRetryableErrorsCounter;
91+
92+
@Mock
93+
private Counter retryableErrorsCounter;
94+
8395
@Mock
8496
private static Logger log;
8597

@@ -92,7 +104,16 @@ static void setupLogger() {
92104
@BeforeEach
93105
void setUp() {
94106
when(pluginMetrics.timer(anyString())).thenReturn(bufferWriteLatencyTimer);
95-
when(pluginMetrics.counter(anyString())).thenReturn(mock(Counter.class));
107+
// Configure specific counters
108+
when(pluginMetrics.counter("apiCalls")).thenReturn(apiCallsCounter);
109+
when(pluginMetrics.counter("nonRetryableErrors")).thenReturn(nonRetryableErrorsCounter);
110+
when(pluginMetrics.counter("retryableErrors")).thenReturn(retryableErrorsCounter);
111+
// Configure other known counters
112+
when(pluginMetrics.counter("bufferWriteAttempts")).thenReturn(mock(Counter.class));
113+
when(pluginMetrics.counter("bufferWriteSuccess")).thenReturn(mock(Counter.class));
114+
when(pluginMetrics.counter("bufferWriteRetrySuccess")).thenReturn(mock(Counter.class));
115+
when(pluginMetrics.counter("bufferWriteRetryAttempts")).thenReturn(mock(Counter.class));
116+
when(pluginMetrics.counter("bufferWriteFailures")).thenReturn(mock(Counter.class));
96117
when(state.getStartTime()).thenReturn(Instant.now().minus(Duration.ofHours(1)));
97118
when(state.getEndTime()).thenReturn(Instant.now());
98119
when(state.getDimensionType()).thenReturn("Exchange");
@@ -111,15 +132,14 @@ void testExecutePartition() throws Exception {
111132
AuditLogsResponse response = new AuditLogsResponse(
112133
Arrays.asList(Map.of(
113134
"contentId", "ID1",
114-
"contentUri", "uri1"
115-
)), null);
135+
"contentUri", "uri1")),
136+
null);
116137

117138
when(service.searchAuditLogs(
118139
eq("Exchange"),
119140
any(Instant.class),
120141
any(Instant.class),
121-
isNull()
122-
)).thenReturn(response);
142+
isNull())).thenReturn(response);
123143

124144
when(service.getAuditLog(anyString()))
125145
.thenReturn("{\"Workload\":\"Exchange\",\"Operation\":\"Test\"}");
@@ -143,6 +163,11 @@ void testExecutePartition() throws Exception {
143163
assertNotNull(record.getData());
144164
assertEquals("Exchange", record.getData().getMetadata().getAttribute("contentType"));
145165
}
166+
167+
// Verify counters - 2 API calls: searchAuditLogs + getAuditLog
168+
verify(apiCallsCounter, times(2)).increment();
169+
verify(nonRetryableErrorsCounter, never()).increment();
170+
verify(retryableErrorsCounter, never()).increment();
146171
}
147172

148173
@Test
@@ -154,18 +179,18 @@ void testExecutePartitionWithJsonProcessingError() throws Exception {
154179
AuditLogsResponse response = new AuditLogsResponse(
155180
Arrays.asList(Map.of(
156181
"contentId", "ID1",
157-
"contentUri", "uri1"
158-
)), null);
182+
"contentUri", "uri1")),
183+
null);
159184

160185
when(service.searchAuditLogs(
161186
anyString(),
162187
any(Instant.class),
163188
any(Instant.class),
164-
any()
165-
)).thenReturn(response);
189+
any())).thenReturn(response);
166190

167191
when(service.getAuditLog(anyString())).thenReturn("{\"invalid\":json}");
168-
when(mockObjectMapper.readTree(anyString())).thenThrow(new JsonProcessingException("Test error") {});
192+
when(mockObjectMapper.readTree(anyString())).thenThrow(new JsonProcessingException("Test error") {
193+
});
169194

170195
doAnswer(invocation -> {
171196
Runnable runnable = invocation.getArgument(0);
@@ -176,6 +201,11 @@ void testExecutePartitionWithJsonProcessingError() throws Exception {
176201
client.executePartition(state, buffer, acknowledgementSet);
177202

178203
verify(buffer).writeAll(argThat(list -> list.isEmpty()), anyInt());
204+
205+
// Verify counters - 2 API calls: searchAuditLogs + getAuditLog, JSON processing errors are non-retryable
206+
verify(apiCallsCounter, times(2)).increment();
207+
verify(nonRetryableErrorsCounter).increment();
208+
verify(retryableErrorsCounter, never()).increment();
179209
}
180210

181211
@Test
@@ -185,15 +215,14 @@ void testBufferWriteWithAcknowledgements() throws Exception {
185215
AuditLogsResponse response = new AuditLogsResponse(
186216
Arrays.asList(Map.of(
187217
"contentId", "ID1",
188-
"contentUri", "uri1"
189-
)), null);
218+
"contentUri", "uri1")),
219+
null);
190220

191221
when(service.searchAuditLogs(
192222
anyString(),
193223
any(Instant.class),
194224
any(Instant.class),
195-
any()
196-
)).thenReturn(response);
225+
any())).thenReturn(response);
197226

198227
when(service.getAuditLog(anyString()))
199228
.thenReturn("{\"Workload\":\"Exchange\",\"Operation\":\"Test\"}");
@@ -210,6 +239,11 @@ void testBufferWriteWithAcknowledgements() throws Exception {
210239
verify(acknowledgementSet).add(any(Event.class));
211240
verify(acknowledgementSet).complete();
212241
verify(buffer).writeAll(any(), anyInt());
242+
243+
// Verify counters - 2 API calls: searchAuditLogs + getAuditLog
244+
verify(apiCallsCounter, times(2)).increment();
245+
verify(nonRetryableErrorsCounter, never()).increment();
246+
verify(retryableErrorsCounter, never()).increment();
213247
}
214248

215249
@Test
@@ -219,15 +253,14 @@ void testBufferWriteTimeout() throws Exception {
219253
AuditLogsResponse response = new AuditLogsResponse(
220254
Arrays.asList(Map.of(
221255
"contentId", "ID1",
222-
"contentUri", "uri1"
223-
)), null);
256+
"contentUri", "uri1")),
257+
null);
224258

225259
when(service.searchAuditLogs(
226260
anyString(),
227261
any(Instant.class),
228262
any(Instant.class),
229-
any()
230-
)).thenReturn(response);
263+
any())).thenReturn(response);
231264

232265
when(service.getAuditLog(anyString()))
233266
.thenReturn("{\"Workload\":\"Exchange\",\"Operation\":\"Test\"}");
@@ -247,6 +280,12 @@ void testBufferWriteTimeout() throws Exception {
247280

248281
assertEquals("Error writing to buffer", exception.getMessage());
249282
verify(buffer).writeAll(any(), anyInt());
283+
284+
// Verify counters - 2 API calls: searchAuditLogs + getAuditLog, and retryable error counter
285+
// since buffer errors are retryable
286+
verify(apiCallsCounter, times(2)).increment();
287+
verify(nonRetryableErrorsCounter, never()).increment();
288+
verify(retryableErrorsCounter).increment();
250289
}
251290

252291
@Test
@@ -256,14 +295,14 @@ void testNonRetryableError() throws Exception {
256295
AuditLogsResponse response = new AuditLogsResponse(
257296
Arrays.asList(Map.of(
258297
"contentId", "ID1",
259-
"contentUri", "uri1")), null);
298+
"contentUri", "uri1")),
299+
null);
260300

261301
when(service.searchAuditLogs(
262302
anyString(),
263303
any(Instant.class),
264304
any(Instant.class),
265-
any()
266-
)).thenReturn(response);
305+
any())).thenReturn(response);
267306

268307
when(service.getAuditLog(anyString())).thenReturn(null);
269308

@@ -276,6 +315,11 @@ void testNonRetryableError() throws Exception {
276315
client.executePartition(state, buffer, acknowledgementSet);
277316

278317
verify(buffer).writeAll(argThat(list -> list.isEmpty()), anyInt());
318+
319+
// Verify counters - 2 API calls: searchAuditLogs + getAuditLog, null content causes non-retryable error
320+
verify(apiCallsCounter, times(2)).increment();
321+
verify(nonRetryableErrorsCounter).increment();
322+
verify(retryableErrorsCounter, never()).increment();
279323
}
280324

281325
@Test
@@ -285,15 +329,14 @@ void testMissingWorkloadField() throws Exception {
285329
AuditLogsResponse response = new AuditLogsResponse(
286330
Arrays.asList(Map.of(
287331
"contentId", "ID1",
288-
"contentUri", "uri1"
289-
)), null);
332+
"contentUri", "uri1")),
333+
null);
290334

291335
when(service.searchAuditLogs(
292336
anyString(),
293337
any(Instant.class),
294338
any(Instant.class),
295-
any()
296-
)).thenReturn(response);
339+
any())).thenReturn(response);
297340

298341
when(service.getAuditLog(anyString())).thenReturn("{\"Operation\":\"Test\"}");
299342

@@ -306,5 +349,10 @@ void testMissingWorkloadField() throws Exception {
306349
client.executePartition(state, buffer, acknowledgementSet);
307350

308351
verify(buffer).writeAll(argThat(list -> list.isEmpty()), anyInt());
352+
353+
// Verify counters - 2 API calls: searchAuditLogs + getAuditLog, missing workload field causes non-retryable error
354+
verify(apiCallsCounter, times(2)).increment();
355+
verify(nonRetryableErrorsCounter).increment();
356+
verify(retryableErrorsCounter, never()).increment();
309357
}
310358
}

0 commit comments

Comments
 (0)