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 0795267005bcf..060f5656cd000 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 @@ 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 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 lastTimeEventIds = new HashSet<>(); private volatile Instant lastCheckTime; private volatile Instant lastReloadTime; private final Map updates = new HashMap<>(); @@ -220,6 +229,10 @@ protected void doStart() throws Exception { } 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 @@ protected void doShutdown() throws Exception { } 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 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 void run() { 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 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 events = new ArrayList<>(); + for (String eventName : UPDATE_EVENT_NAMES) { + lookupEvents(eventName, events); } LOG.debug("Found {} events", events.size()); + Instant newest = lastTime; + Set 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 void run() { } } } + 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 0000000000000..4888bf91084bc --- /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 visibleEvents = new ArrayList<>(); + private final List 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 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 f1cc998b316ea..803ea826c9d98 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 @@ -1118,6 +1118,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