This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new d5f34a8bd8ac CAMEL-25729: camel-aws-secrets-manager - read every
CloudTrail event since the last check in the refresh task (#28036)
d5f34a8bd8ac is described below
commit d5f34a8bd8ac7cc43c3333bb251c7511e4e60af6
Author: allthingssecurity <[email protected]>
AuthorDate: Mon Oct 19 00:35:52 2026 +0530
CAMEL-25729: camel-aws-secrets-manager - read every CloudTrail event since
the last check in the refresh task (#28036)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../vault/CloudTrailReloadTriggerTask.java | 81 ++++++--
.../CloudTrailReloadTriggerTaskLookupTest.java | 230 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 10 +
3 files changed, 305 insertions(+), 16 deletions(-)
diff --git
a/components/camel-aws/camel-aws-secrets-manager/src/main/java/org/apache/camel/component/aws/secretsmanager/vault/CloudTrailReloadTriggerTask.java
b/components/camel-aws/camel-aws-secrets-manager/src/main/java/org/apache/camel/component/aws/secretsmanager/vault/CloudTrailReloadTriggerTask.java
index 0795267005bc..060f5656cd00 100644
---
a/components/camel-aws/camel-aws-secrets-manager/src/main/java/org/apache/camel/component/aws/secretsmanager/vault/CloudTrailReloadTriggerTask.java
+++
b/components/camel-aws/camel-aws-secrets-manager/src/main/java/org/apache/camel/component/aws/secretsmanager/vault/CloudTrailReloadTriggerTask.java
@@ -18,6 +18,7 @@ package org.apache.camel.component.aws.secretsmanager.vault;
import java.net.URI;
import java.time.Instant;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
@@ -99,6 +100,12 @@ public class CloudTrailReloadTriggerTask extends
ServiceSupport implements Camel
private static final String SECRETSMANAGER_UPDATE_SECRET_EVENT =
"UpdateSecret";
+ // the largest page that LookupEvents accepts
+ private static final int LOOKUP_MAX_RESULTS = 50;
+ // the events that change the value of a secret, looked up one by one
+ private static final List<String> UPDATE_EVENT_NAMES
+ = List.of(SECRETSMANAGER_UPDATE_EVENT,
SECRETSMANAGER_UPDATE_SECRET_EVENT);
+
private CamelContext camelContext;
private boolean reloadEnabled = true;
private String secrets;
@@ -106,6 +113,8 @@ public class CloudTrailReloadTriggerTask extends
ServiceSupport implements Camel
private SqsClient sqsClient;
private SecretsManagerPropertiesFunction propertiesFunction;
private volatile Instant lastTime;
+ // the ids of the events seen at exactly lastTime, which the next lookup
returns again (its startTime is inclusive)
+ private final Set<String> lastTimeEventIds = new HashSet<>();
private volatile Instant lastCheckTime;
private volatile Instant lastReloadTime;
private final Map<String, Instant> updates = new HashMap<>();
@@ -220,6 +229,10 @@ public class CloudTrailReloadTriggerTask extends
ServiceSupport implements Camel
}
cloudTrailClient = isOverrideEndpoint
?
clientBuilder.endpointOverride(URI.create(uriEndpointOverride)).build() :
clientBuilder.build();
+ if (lastTime == null) {
+ // look for the updates made from now on, not in the event
history of the last 90 days
+ lastTime = Instant.now();
+ }
} else {
SqsClientBuilder clientBuilder = SqsClient.builder();
if (ObjectHelper.isNotEmpty(accessKey) &&
ObjectHelper.isNotEmpty(secretKey) && ObjectHelper.isNotEmpty(region)) {
@@ -263,6 +276,31 @@ public class CloudTrailReloadTriggerTask extends
ServiceSupport implements Camel
}
updates.clear();
+ lastTime = null;
+ lastTimeEventIds.clear();
+ }
+
+ /**
+ * Adds every event with the given event name since the last check to the
list, reading all pages.
+ */
+ private void lookupEvents(String eventName, List<Event> events) {
+ String nextToken = null;
+ do {
+ LookupEventsRequest.Builder eventsRequestBuilder =
LookupEventsRequest.builder()
+ .maxResults(LOOKUP_MAX_RESULTS)
+
.lookupAttributes(LookupAttribute.builder().attributeKey(LookupAttributeKey.EVENT_NAME)
+ .attributeValue(eventName).build());
+ if (lastTime != null) {
+ // inclusive, so an event in the same second as the newest
event seen is not skipped
+ eventsRequestBuilder.startTime(lastTime);
+ }
+ if (nextToken != null) {
+ eventsRequestBuilder.nextToken(nextToken);
+ }
+ LookupEventsResponse response =
cloudTrailClient.lookupEvents(eventsRequestBuilder.build());
+ events.addAll(response.events());
+ nextToken = response.nextToken();
+ } while (ObjectHelper.isNotEmpty(nextToken));
}
@Override
@@ -273,26 +311,30 @@ public class CloudTrailReloadTriggerTask extends
ServiceSupport implements Camel
if (!useSqsNotification) {
try {
- LookupEventsRequest.Builder eventsRequestBuilder =
LookupEventsRequest.builder()
- .maxResults(100)
-
.lookupAttributes(LookupAttribute.builder().attributeKey(LookupAttributeKey.EVENT_SOURCE)
-
.attributeValue(SECRETSMANAGER_AMAZONAWS_COM).build());
-
- if (lastTime != null) {
- eventsRequestBuilder.startTime(lastTime.plusMillis(1000));
- }
-
- LookupEventsRequest lookupEventsRequest =
eventsRequestBuilder.build();
-
- LookupEventsResponse response =
cloudTrailClient.lookupEvents(lookupEventsRequest);
- List<Event> events = response.events();
-
- if (!events.isEmpty()) {
- lastTime = events.get(0).eventTime();
+ // read every page of the update events since the last check.
LookupEvents takes a single lookup
+ // attribute, so there is one lookup per event name: filtering
on the event source instead would also
+ // return every read (GetSecretValue, DescribeSecret), and the
pages would grow with the read traffic
+ List<Event> events = new ArrayList<>();
+ for (String eventName : UPDATE_EVENT_NAMES) {
+ lookupEvents(eventName, events);
}
LOG.debug("Found {} events", events.size());
+ Instant newest = lastTime;
+ Set<String> newestEventIds = new HashSet<>();
for (Event event : events) {
+ if (lastTimeEventIds.contains(event.eventId())) {
+ // already seen by the previous check
+ continue;
+ }
+ Instant eventTime = event.eventTime();
+ if (newest == null || eventTime.isAfter(newest)) {
+ newest = eventTime;
+ newestEventIds.clear();
+ newestEventIds.add(event.eventId());
+ } else if (eventTime.equals(newest)) {
+ newestEventIds.add(event.eventId());
+ }
if
(event.eventSource().equalsIgnoreCase(SECRETSMANAGER_AMAZONAWS_COM)) {
if
(event.eventName().equalsIgnoreCase(SECRETSMANAGER_UPDATE_EVENT)
||
event.eventName().equalsIgnoreCase(SECRETSMANAGER_UPDATE_SECRET_EVENT)) {
@@ -311,6 +353,13 @@ public class CloudTrailReloadTriggerTask extends
ServiceSupport implements Camel
}
}
}
+ if (newest != null && newest.equals(lastTime)) {
+ // no newer event: the events at lastTime seen before are
still returned by the next lookup
+ newestEventIds.addAll(lastTimeEventIds);
+ }
+ lastTime = newest;
+ lastTimeEventIds.clear();
+ lastTimeEventIds.addAll(newestEventIds);
} catch (Exception e) {
LOG.warn(
"Error during AWS Secrets Refresh Task due to {}. This
exception is ignored. Will try again on next run.",
diff --git
a/components/camel-aws/camel-aws-secrets-manager/src/test/java/org/apache/camel/component/aws/secretsmanager/vault/CloudTrailReloadTriggerTaskLookupTest.java
b/components/camel-aws/camel-aws-secrets-manager/src/test/java/org/apache/camel/component/aws/secretsmanager/vault/CloudTrailReloadTriggerTaskLookupTest.java
new file mode 100644
index 000000000000..4888bf91084b
--- /dev/null
+++
b/components/camel-aws/camel-aws-secrets-manager/src/test/java/org/apache/camel/component/aws/secretsmanager/vault/CloudTrailReloadTriggerTaskLookupTest.java
@@ -0,0 +1,230 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.aws.secretsmanager.vault;
+
+import java.lang.reflect.Field;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.List;
+
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.spi.ContextReloadStrategy;
+import org.apache.camel.vault.AwsVaultConfiguration;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import software.amazon.awssdk.services.cloudtrail.CloudTrailClient;
+import software.amazon.awssdk.services.cloudtrail.model.Event;
+import software.amazon.awssdk.services.cloudtrail.model.LookupAttribute;
+import software.amazon.awssdk.services.cloudtrail.model.LookupAttributeKey;
+import software.amazon.awssdk.services.cloudtrail.model.LookupEventsRequest;
+import software.amazon.awssdk.services.cloudtrail.model.LookupEventsResponse;
+import software.amazon.awssdk.services.cloudtrail.model.Resource;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests the CloudTrail lookups of the refresh task against a fake {@code
LookupEvents} that answers like the service:
+ * only the events that match the single lookup attribute and have an event
time at or after {@code startTime}, newest
+ * first, at most 50 per page, and a next token while events are left.
+ */
+class CloudTrailReloadTriggerTaskLookupTest {
+
+ private static final String SOURCE = "secretsmanager.amazonaws.com";
+ private static final int MAX_PAGE_SIZE = 50;
+
+ private final List<Event> visibleEvents = new ArrayList<>();
+ private final List<LookupEventsRequest> requests = new ArrayList<>();
+ private final CloudTrailClient client = mock(CloudTrailClient.class);
+ private final ContextReloadStrategy reload =
mock(ContextReloadStrategy.class);
+ private DefaultCamelContext context;
+ private int eventIds;
+
+ @BeforeEach
+ void setUp() throws Exception {
+
when(client.lookupEvents(any(LookupEventsRequest.class))).thenAnswer(inv ->
lookup(inv.getArgument(0)));
+ context = new DefaultCamelContext();
+ context.addService(reload);
+ context.start();
+ }
+
+ @AfterEach
+ void tearDown() {
+ context.stop();
+ }
+
+ private LookupEventsResponse lookup(LookupEventsRequest request) {
+ requests.add(request);
+ // the service takes at most one lookup attribute
+ assertTrue(request.lookupAttributes().size() <= 1, "LookupEvents takes
at most one lookup attribute");
+ LookupAttribute attribute = request.lookupAttributes().isEmpty() ?
null : request.lookupAttributes().get(0);
+ List<Event> matching = visibleEvents.stream()
+ .filter(e -> attribute == null || matches(attribute, e))
+ .filter(e -> request.startTime() == null ||
!e.eventTime().isBefore(request.startTime()))
+ .sorted(Comparator.comparing(Event::eventTime).reversed())
+ .toList();
+ int pageSize = request.maxResults() == null ? MAX_PAGE_SIZE :
Math.min(request.maxResults(), MAX_PAGE_SIZE);
+ int from = request.nextToken() == null ? 0 :
Integer.parseInt(request.nextToken());
+ int to = Math.min(from + pageSize, matching.size());
+ LookupEventsResponse.Builder response =
LookupEventsResponse.builder().events(matching.subList(from, to));
+ if (to < matching.size()) {
+ response.nextToken(Integer.toString(to));
+ }
+ return response.build();
+ }
+
+ private static boolean matches(LookupAttribute attribute, Event event) {
+ return switch (attribute.attributeKey()) {
+ case EVENT_NAME ->
attribute.attributeValue().equals(event.eventName());
+ case EVENT_SOURCE ->
attribute.attributeValue().equals(event.eventSource());
+ default -> throw new IllegalArgumentException("Unexpected lookup
attribute " + attribute);
+ };
+ }
+
+ private void addEvent(String name, String secret, Instant time) {
+ visibleEvents.add(Event.builder().eventId("id-" +
eventIds++).eventSource(SOURCE).eventName(name)
+
.eventTime(time).resources(Resource.builder().resourceName(secret).build()).build());
+ }
+
+ private CloudTrailReloadTriggerTask task(Instant lastTime) throws
Exception {
+ CloudTrailReloadTriggerTask task = new CloudTrailReloadTriggerTask();
+ task.setCamelContext(context);
+ setField(task, "cloudTrailClient", client);
+ setField(task, "secrets", "tracked-secret");
+ setField(task, "lastTime", lastTime);
+ return task;
+ }
+
+ private static void setField(Object target, String name, Object value)
throws Exception {
+ Field f = CloudTrailReloadTriggerTask.class.getDeclaredField(name);
+ f.setAccessible(true);
+ f.set(target, value);
+ }
+
+ @Test
+ void updateBehindMoreThanOnePageOfOtherEventsReloadsTheContext() throws
Exception {
+ Instant t0 = Instant.parse("2026-10-17T10:00:00Z");
+ CloudTrailReloadTriggerTask task = task(t0);
+ // the secret is updated, then other secrets are updated 60 times
before the next check
+ addEvent("PutSecretValue", "tracked-secret", t0.plusSeconds(1));
+ for (int i = 0; i < 60; i++) {
+ addEvent("PutSecretValue", "other-secret", t0.plusSeconds(2 + i %
20));
+ }
+
+ task.run();
+
+ assertTrue(task.getUpdates().containsKey("tracked-secret"), "the
update on the second page must be seen");
+ verify(reload, times(1)).onReload(task);
+ assertTrue(requests.stream().allMatch(r -> r.maxResults() != null &&
r.maxResults() <= MAX_PAGE_SIZE),
+ "LookupEvents accepts at most 50 results per page");
+ }
+
+ @Test
+ void updateInTheSameSecondAsTheNewestEventSeenIsNotSkipped() throws
Exception {
+ Instant t0 = Instant.parse("2026-10-17T10:00:00Z");
+ CloudTrailReloadTriggerTask task = task(t0);
+ Instant t1 = t0.plusSeconds(10);
+ addEvent("UpdateSecret", "other-secret", t1);
+
+ task.run();
+ assertFalse(task.getUpdates().containsKey("tracked-secret"));
+
+ // CloudTrail event times are in seconds: an update in the same second
becomes visible after the first check
+ addEvent("PutSecretValue", "tracked-secret", t1);
+ task.run();
+
+ assertTrue(task.getUpdates().containsKey("tracked-secret"), "the
update in the same second must be seen");
+ verify(reload, times(1)).onReload(task);
+
+ // the events already seen at that second do not reload the context
again
+ task.run();
+ verify(reload, times(1)).onReload(task);
+ }
+
+ @Test
+ void firstCheckLooksUpFromTheStartOfTheTask() throws Exception {
+ AwsVaultConfiguration aws = context.getVaultConfiguration().aws();
+ aws.setAccessKey("key");
+ aws.setSecretKey("secret");
+ aws.setRegion("eu-west-1");
+ aws.setSecrets("tracked-secret");
+ CloudTrailReloadTriggerTask task = new CloudTrailReloadTriggerTask();
+ task.setCamelContext(context);
+ Instant before = Instant.now();
+ task.start();
+ try {
+ Field f =
CloudTrailReloadTriggerTask.class.getDeclaredField("cloudTrailClient");
+ f.setAccessible(true);
+ ((CloudTrailClient) f.get(task)).close();
+ f.set(task, client);
+ // an update made long before the task started
+ addEvent("PutSecretValue", "tracked-secret",
before.minusSeconds(86400));
+
+ task.run();
+
+ assertEquals(2, requests.size(), "one lookup per update event
name");
+ for (LookupEventsRequest request : requests) {
+ assertNotNull(request.startTime(), "the first lookup must not
read the whole event history");
+
assertFalse(request.startTime().isBefore(before.minusSeconds(1)));
+ }
+ assertFalse(task.getUpdates().containsKey("tracked-secret"));
+ verify(reload, times(0)).onReload(task);
+ } finally {
+ setField(task, "cloudTrailClient", null);
+ task.stop();
+ }
+ }
+
+ @Test
+ void readsOfSecretsDoNotAddLookupRequests() throws Exception {
+ Instant t0 = Instant.parse("2026-10-17T10:00:00Z");
+ CloudTrailReloadTriggerTask task = task(t0);
+ // a busy account: other applications read secrets 200 times between
two checks
+ for (int i = 0; i < 200; i++) {
+ addEvent(i % 2 == 0 ? "GetSecretValue" : "DescribeSecret",
"other-secret", t0.plusSeconds(1 + i % 30));
+ }
+ addEvent("UpdateSecret", "tracked-secret", t0.plusSeconds(5));
+
+ task.run();
+
+ assertTrue(task.getUpdates().containsKey("tracked-secret"), "an
UpdateSecret event must be seen");
+ verify(reload, times(1)).onReload(task);
+ assertEquals(2, requests.size(), "one request per update event name,
whatever the number of reads");
+ assertEquals(List.of("PutSecretValue", "UpdateSecret"),
+ requests.stream().map(r -> {
+ LookupAttribute a = r.lookupAttributes().get(0);
+ assertEquals(LookupAttributeKey.EVENT_NAME,
a.attributeKey());
+ return a.attributeValue();
+ }).toList());
+
+ // the next check looks up from the newest update seen and does not
reload again
+ requests.clear();
+ task.run();
+ verify(reload, times(1)).onReload(task);
+ assertEquals(2, requests.size());
+ assertTrue(requests.stream().allMatch(r ->
t0.plusSeconds(5).equals(r.startTime())));
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index 68a45bb42317..5b1e7e99e39d 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -1258,6 +1258,16 @@ MINA does not pass a failed handshake on as an error:
the producer fails with a
`org.apache.mina.filter.ssl.SslHandler`. A producer that starts timing out
after the upgrade should be checked for a
certificate that does not list the host.
+=== camel-aws-secrets-manager - the CloudTrail refresh task reads every event
since the last check
+
+With `camel.vault.aws.refreshEnabled=true` and without SQS notification, each
check now reads every page of the
+`PutSecretValue` and `UpdateSecret` events since the previous check, including
the events in the same second as the
+newest event seen before, so an update is no longer missed when more events
than one page happened in between. The
+first check looks for the updates made after the task started: an update made
before the start no longer reloads the
+context. A check sends two `LookupEvents` requests, one per event name, and
one more per further 50 updates (before:
+always one request, for all Secrets Manager events); reads of secrets no
longer add requests. LookupEvents allows two
+requests per second per account and Region.
+
=== camel-aws2-kinesis - the consumer continues after the last record when a
shard iterator expired
When a cached shard iterator has expired (an iterator is valid for 5 minutes,
so after a route was suspended for
