Auto-bootstrap: add verbose logging

Log explicit messages around auto-bootstrapping and unnecessary/left-over secrets that are (still) available.

Author: snazy

polaris-core/src/main/java/org/apache/polaris/core/persistence/LocalPolarisMetaStoreManagerFactory.java

@@ -118,11 +118,6 @@ public synchronized Map<String, PrincipalSecretsResult> bootstrapRealms(
         PrincipalSecretsResult secretsResult =
             bootstrapServiceAndCreatePolarisPrincipalForRealm(
                 realmContext, metaStoreManagerMap.get(realmContext.getRealmIdentifier()));
-
-        if (rootCredentialsSet.credentials().containsKey(realm)) {
-          LOGGER.info("Bootstrapped realm {} using preset credentials.", realm);
-        }
-
         results.put(realmContext.getRealmIdentifier(), secretsResult);
       }
     }

quarkus/service/src/main/java/org/apache/polaris/service/quarkus/config/QuarkusProducers.java

@@ -32,6 +32,7 @@
 import jakarta.ws.rs.container.ContainerRequestContext;
 import jakarta.ws.rs.core.Context;
 import java.time.Clock;
+import java.util.stream.Collectors;
 import org.apache.polaris.core.PolarisCallContext;
 import org.apache.polaris.core.PolarisDefaultDiagServiceImpl;
 import org.apache.polaris.core.PolarisDiagnostics;
@@ -68,8 +69,11 @@
 import org.eclipse.microprofile.config.inject.ConfigProperty;
 import org.eclipse.microprofile.context.ManagedExecutor;
 import org.eclipse.microprofile.context.ThreadContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class QuarkusProducers {
+  private static final Logger LOGGER = LoggerFactory.getLogger(QuarkusProducers.class);
 
   @Produces
   @ApplicationScoped // cannot be singleton because it is mocked in tests
@@ -159,9 +163,68 @@ public void maybeBootstrap(
       MetaStoreManagerFactory factory,
       QuarkusPersistenceConfiguration config,
       RealmContextConfiguration realmContextConfiguration) {
+    var rootCredentialsSet = RootCredentialsSet.fromEnvironment();
+    var rootCredentials = rootCredentialsSet.credentials();
     if (config.isAutoBootstrap()) {
-      RootCredentialsSet rootCredentialsSet = RootCredentialsSet.fromEnvironment();
-      factory.bootstrapRealms(realmContextConfiguration.realms(), rootCredentialsSet);
+      var realmIds = realmContextConfiguration.realms();
+
+      LOGGER.info(
+          "Bootstrapping realm(s) {}, if necessary, from root credentials set provided via the environment variable {} or Java system property {} ...",
+          realmIds.stream().map(r -> "'" + r + "'").collect(Collectors.joining(", ")),
+          RootCredentialsSet.ENVIRONMENT_VARIABLE,
+          RootCredentialsSet.SYSTEM_PROPERTY);
+
+      var result = factory.bootstrapRealms(realmIds, rootCredentialsSet);
+
+      result.forEach(
+          (realm, secrets) -> {
+            var principalSecrets = secrets.getPrincipalSecrets();
+
+            var log =
+                LOGGER
+                    .atInfo()
+                    .addArgument(realm)
+                    .addArgument(RootCredentialsSet.ENVIRONMENT_VARIABLE)
+                    .addArgument(RootCredentialsSet.SYSTEM_PROPERTY);
+            if (rootCredentials.containsKey(realm)) {
+              log.log(
+                  "Realm '{}' automatically bootstrapped, credentials taken from root credentials set provided via the environment variable {} or Java system property {}, not printed to stdout.");
+            } else {
+              log.log(
+                  "Realm '{}' automatically bootstrapped, credentials were not present in root credentials set provided via the environment variable {} or Java system property {}, see separate message printed to stdout.");
+              String msg =
+                  String.format(
+                      "realm: %1s root principal credentials: %2s:%3s",
+                      realm,
+                      principalSecrets.getPrincipalClientId(),
+                      principalSecrets.getMainSecret());
+              System.out.println(msg);
+            }
+          });
+
+      var unusedRealmSecrets =
+          realmIds.stream()
+              .filter(rootCredentials::containsKey)
+              .filter(r -> !result.containsKey(r))
+              .map(r -> "'" + r + "'")
+              .collect(Collectors.joining(", "));
+      if (!unusedRealmSecrets.isEmpty()) {
+        // This is intentionally an error to highlight the importance of the situation.
+        LOGGER.error(
+            "The realms {} are already fully bootstrapped but the secrets are still available via the environment variable {} or Java system property {}. "
+                + "Remove this security sensitive information from the environment / Java system properties!",
+            unusedRealmSecrets,
+            RootCredentialsSet.ENVIRONMENT_VARIABLE,
+            RootCredentialsSet.SYSTEM_PROPERTY);
+      }
+    } else if (!rootCredentials.isEmpty()) {
+      // This is intentionally an error to highlight the importance of the situation.
+      LOGGER.error(
+          "Secrets for the realms {} are available via the environment variable {} or Java system property {}. "
+              + "Remove this security sensitive information from the environment / Java system properties!",
+          rootCredentials.keySet(),
+          RootCredentialsSet.ENVIRONMENT_VARIABLE,
+          RootCredentialsSet.SYSTEM_PROPERTY);
     }
   }