[backport camel-4.14.x] CAMEL-24341: camel-google-secret-manager - fix GCP vault refresh task defects

components/camel-google/camel-google-secret-manager/pom.xml

@@ -83,5 +83,10 @@
             <artifactId>camel-test-junit5</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.assertj</groupId>
+            <artifactId>assertj-core</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 </project>

components/camel-google/camel-google-secret-manager/src/main/java/org/apache/camel/component/google/secret/manager/GoogleSecretManagerPropertiesFunction.java

@@ -55,8 +55,8 @@
  * Otherwise it is possible to specify the credentials as properties:
  *
  * <ul>
- * <li><tt>camel.vault.aws.serviceAccountKey</tt></li>
- * <li><tt>camel.vault.aws.projectId</tt></li>
+ * <li><tt>camel.vault.gcp.serviceAccountKey</tt></li>
+ * <li><tt>camel.vault.gcp.projectId</tt></li>
  * </ul>
  * <p/>
  *
@@ -79,6 +79,8 @@ public class GoogleSecretManagerPropertiesFunction extends ServiceSupport implem
     private static final String CAMEL_VAULT_GCP_PROJECT_ID = "CAMEL_VAULT_GCP_PROJECT_ID";
     private static final String CAMEL_VAULT_GCP_USE_DEFAULT_INSTANCE = "CAMEL_VAULT_GCP_USE_DEFAULT_INSTANCE";
 
+    private static final ObjectMapper MAPPER = new ObjectMapper();
+
     boolean useDefaultInstance;
 
     private CamelContext camelContext;
@@ -207,8 +209,7 @@ private String getSecretFromSource(
                 returnValue = response.getPayload().getData().toStringUtf8();
             }
             if (ObjectHelper.isNotEmpty(subkey) && ObjectHelper.isNotEmpty(returnValue)) {
-                ObjectMapper mapper = new ObjectMapper();
-                JsonNode actualObj = mapper.readTree(returnValue);
+                JsonNode actualObj = MAPPER.readTree(returnValue);
                 JsonNode field = actualObj.get(subkey);
                 if (ObjectHelper.isNotEmpty(field)) {
                     returnValue = field.textValue();

components/camel-google/camel-google-secret-manager/src/main/java/org/apache/camel/component/google/secret/manager/vault/PubsubReloadTriggerTask.java

@@ -24,6 +24,7 @@
 import java.util.Map;
 import java.util.Set;
 
+import com.google.api.core.ApiService;
 import com.google.api.gax.core.FixedCredentialsProvider;
 import com.google.auth.Credentials;
 import com.google.auth.oauth2.GoogleCredentials;
@@ -89,7 +90,7 @@ public boolean isReloadEnabled() {
     }
 
     /**
-     * Whether Camel should be reloaded on AWS secret updated
+     * Whether Camel should be reloaded on GCP secret updated
      */
     public void setReloadEnabled(boolean reloadEnabled) {
         this.reloadEnabled = reloadEnabled;
@@ -120,15 +121,17 @@ public Instant getLastReloadTime() {
     protected void doStart() throws Exception {
         super.doStart();
 
+        // specific secrets
+        secrets = camelContext.getVaultConfiguration().gcp().getSecrets();
+
         // auto-detect secrets in-use
         PropertiesComponent pc = camelContext.getPropertiesComponent();
         PropertiesFunction pf = pc.getPropertiesFunction("gcp");
         if (pf instanceof GoogleSecretManagerPropertiesFunction) {
             propertiesFunction = (GoogleSecretManagerPropertiesFunction) pf;
             LOG.debug("Auto-detecting secrets from properties-function: {}", pf.getName());
         }
-        // specific secrets
-        secrets = camelContext.getVaultConfiguration().aws().getSecrets();
+
         if (ObjectHelper.isEmpty(secrets) && propertiesFunction == null) {
             throw new IllegalArgumentException("Secrets must be configured on GCP vault configuration");
         }
@@ -187,7 +190,14 @@ protected void doShutdown() throws Exception {
     public void run() {
         lastCheckTime = Instant.now();
 
-        subscriber.startAsync().awaitRunning();
+        if (subscriber == null) {
+            return;
+        }
+        // the subscriber is push based, so it only has to be started once. A Google ApiService can only be
+        // started while it is still NEW, starting it again on every period would fail with an IllegalStateException
+        if (subscriber.state() == ApiService.State.NEW) {
+            subscriber.startAsync().awaitRunning();
+        }
     }
 
     protected boolean matchSecret(String name) {
@@ -220,14 +230,13 @@ public class FilteringEventMessageReceiver implements MessageReceiver {
         private static final String SECRET_UPDATE = "SECRET_UPDATE";
         private static final String SECRET_VERSION_ADD = "SECRET_VERSION_ADD";
 
-        private boolean triggerReloading;
-
         @Override
         public void receiveMessage(PubsubMessage message, AckReplyConsumer consumer) {
+            boolean triggerReloading = false;
             String secretId = message.getAttributesMap().get("secretId");
             String eventType = message.getAttributesMap().get("eventType");
-            if (eventType.equalsIgnoreCase(SECRET_UPDATE) || eventType.equalsIgnoreCase(SECRET_VERSION_ADD)) {
-                if (matchSecret(secretId)) {
+            if (SECRET_UPDATE.equalsIgnoreCase(eventType) || SECRET_VERSION_ADD.equalsIgnoreCase(eventType)) {
+                if (secretId != null && matchSecret(secretId)) {
                     int secretNameBeginInd = secretId.lastIndexOf("/") + 1;
                     updates.put(secretId.substring(secretNameBeginInd),
                             Instant.ofEpochSecond(message.getPublishTime().getSeconds(), message.getPublishTime().getNanos()));

components/camel-google/camel-google-secret-manager/src/test/java/org/apache/camel/component/google/secret/manager/vault/PubsubReloadTriggerTaskTest.java

@@ -0,0 +1,156 @@
+/*
+ * 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.google.secret.manager.vault;
+
+import java.lang.reflect.Field;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import com.google.cloud.pubsub.v1.AckReplyConsumer;
+import com.google.pubsub.v1.PubsubMessage;
+import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.support.DefaultContextReloadStrategy;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+class PubsubReloadTriggerTaskTest {
+
+    private static final class CountingReloadStrategy extends DefaultContextReloadStrategy {
+        private final AtomicInteger reloads = new AtomicInteger();
+
+        @Override
+        public void onReload(Object source) {
+            reloads.incrementAndGet();
+        }
+    }
+
+    private static final class RecordingAck implements AckReplyConsumer {
+        private boolean acked;
+
+        @Override
+        public void ack() {
+            acked = true;
+        }
+
+        @Override
+        public void nack() {
+        }
+    }
+
+    private static void setField(Object target, String name, Object value) throws Exception {
+        Field f = PubsubReloadTriggerTask.class.getDeclaredField(name);
+        f.setAccessible(true);
+        f.set(target, value);
+    }
+
+    private static Object getField(Object target, String name) throws Exception {
+        Field f = PubsubReloadTriggerTask.class.getDeclaredField(name);
+        f.setAccessible(true);
+        return f.get(target);
+    }
+
+    private static PubsubMessage message(String eventType, String secretId) {
+        PubsubMessage.Builder builder = PubsubMessage.newBuilder();
+        if (eventType != null) {
+            builder.putAttributes("eventType", eventType);
+        }
+        if (secretId != null) {
+            builder.putAttributes("secretId", secretId);
+        }
+        return builder.build();
+    }
+
+    @Test
+    void secretsAreReadFromTheGcpVaultConfiguration() throws Exception {
+        DefaultCamelContext context = new DefaultCamelContext();
+        context.getVaultConfiguration().gcp().setSecrets("gcp-secret");
+        context.getVaultConfiguration().aws().setSecrets("aws-secret");
+
+        PubsubReloadTriggerTask task = new PubsubReloadTriggerTask();
+        task.setCamelContext(context);
+
+        // no GCP credentials are configured, so starting the task fails once it reaches the gcp properties
+        // function. By then the secrets to watch must already have been read from the GCP vault configuration
+        assertThatThrownBy(task::start).isInstanceOf(RuntimeCamelException.class);
+        assertThat(getField(task, "secrets")).isEqualTo("gcp-secret");
+    }
+
+    @Test
+    void reloadIsTriggeredOnlyForMatchingSecrets() throws Exception {
+        DefaultCamelContext context = new DefaultCamelContext();
+        CountingReloadStrategy reload = new CountingReloadStrategy();
+        context.addService(reload);
+        context.start();
+
+        PubsubReloadTriggerTask task = new PubsubReloadTriggerTask();
+        task.setCamelContext(context);
+        setField(task, "secrets", "tracked");
+
+        PubsubReloadTriggerTask.FilteringEventMessageReceiver receiver = task.new FilteringEventMessageReceiver();
+
+        RecordingAck matching = new RecordingAck();
+        receiver.receiveMessage(message("SECRET_UPDATE", "projects/p/secrets/tracked"), matching);
+        assertThat(reload.reloads.get()).isEqualTo(1);
+        assertThat(matching.acked).isTrue();
+        assertThat(task.getUpdates()).containsKey("tracked");
+
+        // a message for a secret that is not watched must not trigger another reload
+        RecordingAck other = new RecordingAck();
+        receiver.receiveMessage(message("SECRET_UPDATE", "projects/p/secrets/other"), other);
+        assertThat(reload.reloads.get()).isEqualTo(1);
+        assertThat(other.acked).isTrue();
+
+        // neither must an event type that is not a secret update
+        RecordingAck unrelated = new RecordingAck();
+        receiver.receiveMessage(message("SECRET_DELETE", "projects/p/secrets/tracked"), unrelated);
+        assertThat(reload.reloads.get()).isEqualTo(1);
+        assertThat(unrelated.acked).isTrue();
+
+        context.stop();
+    }
+
+    @Test
+    void messagesWithoutAttributesAreAcknowledged() throws Exception {
+        DefaultCamelContext context = new DefaultCamelContext();
+        PubsubReloadTriggerTask task = new PubsubReloadTriggerTask();
+        task.setCamelContext(context);
+        setField(task, "secrets", "tracked");
+
+        PubsubReloadTriggerTask.FilteringEventMessageReceiver receiver = task.new FilteringEventMessageReceiver();
+
+        // a message published on the subscription by something other than the secret manager event feed
+        RecordingAck noAttributes = new RecordingAck();
+        assertThatCode(() -> receiver.receiveMessage(message(null, null), noAttributes)).doesNotThrowAnyException();
+        assertThat(noAttributes.acked).isTrue();
+
+        RecordingAck noSecretId = new RecordingAck();
+        assertThatCode(() -> receiver.receiveMessage(message("SECRET_UPDATE", null), noSecretId)).doesNotThrowAnyException();
+        assertThat(noSecretId.acked).isTrue();
+    }
+
+    @Test
+    void runWithoutSubscriberDoesNotThrow() {
+        PubsubReloadTriggerTask task = new PubsubReloadTriggerTask();
+        task.setCamelContext(new DefaultCamelContext());
+
+        assertThatCode(task::run).doesNotThrowAnyException();
+        assertThat(task.getLastCheckTime()).isNotNull();
+    }
+}