AWS CloudWatch Event Sink Implementation

gradle/libs.versions.toml

@@ -72,6 +72,7 @@ jakarta-validation-api = { module = "jakarta.validation:jakarta.validation-api",
 jakarta-ws-rs-api = { module = "jakarta.ws.rs:jakarta.ws.rs-api", version = "4.0.0" }
 javax-servlet-api = { module = "javax.servlet:javax.servlet-api", version = "4.0.1" }
 junit-bom = { module = "org.junit:junit-bom", version = "5.13.1" }
+localstack = { module = "org.testcontainers:localstack", version = "1.19.7" }
 logback-classic = { module = "ch.qos.logback:logback-classic", version = "1.5.18" }
 micrometer-bom = { module = "io.micrometer:micrometer-bom", version = "1.15.1" }
 microprofile-fault-tolerance-api = { module = "org.eclipse.microprofile.fault-tolerance:microprofile-fault-tolerance-api", version = "4.1.2" }

runtime/defaults/src/main/resources/application.properties

@@ -126,6 +126,10 @@ polaris.secrets-manager.type=in-memory
 polaris.file-io.type=default
 
 polaris.event-listener.type=no-op
+# polaris.event-listener.type=aws-cloudwatch
+# polaris.event-listener.aws-cloudwatch.log-group=test-group
+# polaris.event-listener.aws-cloudwatch.log-stream=test-stream
+# polaris.event-listener.aws-cloudwatch.region=us-west-2
 
 polaris.log.request-id-header-name=Polaris-Request-Id
 # polaris.log.mdc.aid=polaris

runtime/service/src/main/java/org/apache/polaris/service/quarkus/events/AwsCloudwatchConfig.java

@@ -0,0 +1,28 @@
+/*
+ * 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.polaris.service.quarkus.events;
+
+public interface AwsCloudwatchConfig {
+  String logGroup();
+
+  String logStream();
+
+  String region();
+}

runtime/service/src/main/java/org/apache/polaris/service/quarkus/events/QuarkusPolarisEventListenerConfiguration.java

@@ -20,13 +20,29 @@
 
 import io.quarkus.runtime.annotations.StaticInitSafe;
 import io.smallrye.config.ConfigMapping;
+import io.smallrye.config.WithName;
+import java.util.Optional;
+import org.apache.polaris.service.events.EventListenerConfiguration;
+import org.apache.polaris.service.events.PolarisEventListener;
 
 @StaticInitSafe
 @ConfigMapping(prefix = "polaris.event-listener")
-public interface QuarkusPolarisEventListenerConfiguration {
+public interface QuarkusPolarisEventListenerConfiguration extends EventListenerConfiguration {
   /**
-   * The type of the event listener to use. Must be a registered {@link
-   * org.apache.polaris.service.events.PolarisEventListener} identifier.
+   * The type of the event listener to use. Must be a registered {@link PolarisEventListener}
+   * identifier.
    */
   String type();
+
+  @WithName("aws-cloudwatch.log-group")
+  @Override
+  Optional<String> awsCloudwatchlogGroup();
+
+  @WithName("aws-cloudwatch.log-stream")
+  @Override
+  Optional<String> awsCloudwatchlogStream();
+
+  @WithName("aws-cloudwatch.region")
+  @Override
+  Optional<String> awsCloudwatchRegion();
 }

service/common/build.gradle.kts

@@ -82,6 +82,7 @@ dependencies {
   implementation("software.amazon.awssdk:sts")
   implementation("software.amazon.awssdk:iam-policy-builder")
   implementation("software.amazon.awssdk:s3")
+  implementation("software.amazon.awssdk:cloudwatchlogs")
 
   implementation(platform(libs.azuresdk.bom))
   implementation("com.azure:azure-core")
@@ -94,6 +95,8 @@ dependencies {
   testImplementation("org.junit.jupiter:junit-jupiter")
   testImplementation(libs.assertj.core)
   testImplementation(libs.mockito.core)
+  testImplementation(libs.localstack)
+  testImplementation("org.testcontainers:testcontainers")
   testRuntimeOnly("org.junit.platform:junit-platform-launcher")
 
   testImplementation(libs.logback.classic)

service/common/src/main/java/org/apache/polaris/service/admin/PolarisServiceImpl.java

@@ -80,6 +80,8 @@
 import org.apache.polaris.service.admin.api.PolarisPrincipalsApiService;
 import org.apache.polaris.service.config.RealmEntityManagerFactory;
 import org.apache.polaris.service.config.ReservedProperties;
+import org.apache.polaris.service.events.AfterCatalogCreatedEvent;
+import org.apache.polaris.service.events.PolarisEventListener;
 import org.apache.polaris.service.types.PolicyIdentifier;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -97,6 +99,7 @@ public class PolarisServiceImpl
   private final UserSecretsManagerFactory userSecretsManagerFactory;
   private final CallContext callContext;
   private final ReservedProperties reservedProperties;
+  private final PolarisEventListener polarisEventListener;
 
   @Inject
   public PolarisServiceImpl(
@@ -105,13 +108,15 @@ public PolarisServiceImpl(
       UserSecretsManagerFactory userSecretsManagerFactory,
       PolarisAuthorizer polarisAuthorizer,
       CallContext callContext,
-      ReservedProperties reservedProperties) {
+      ReservedProperties reservedProperties,
+      PolarisEventListener polarisEventListener) {
     this.entityManagerFactory = entityManagerFactory;
     this.metaStoreManagerFactory = metaStoreManagerFactory;
     this.userSecretsManagerFactory = userSecretsManagerFactory;
     this.polarisAuthorizer = polarisAuthorizer;
     this.callContext = callContext;
     this.reservedProperties = reservedProperties;
+    this.polarisEventListener = polarisEventListener;
     // FIXME: This is a hack to set the current context for downstream calls.
     CallContext.setCurrentContext(callContext);
   }
@@ -150,6 +155,8 @@ public Response createCatalog(
     validateExternalCatalog(catalog);
     Catalog newCatalog = new CatalogEntity(adminService.createCatalog(request)).asCatalog();
     LOGGER.info("Created new catalog {}", newCatalog);
+    polarisEventListener.onAfterCatalogCreated(
+        new AfterCatalogCreatedEvent(newCatalog), callContext);
     return Response.status(Response.Status.CREATED).build();
   }
 

service/common/src/main/java/org/apache/polaris/service/events/AfterCatalogCreatedEvent.java

@@ -0,0 +1,29 @@
+/*
+ * 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.polaris.service.events;
+
+import org.apache.polaris.core.admin.model.Catalog;
+
+/**
+ * Emitted after Polaris creates a catalog (internal or external). This is not emitted if there's an
+ * exception while created.
+ *
+ * @param catalog The catalog that was created
+ */
+public record AfterCatalogCreatedEvent(Catalog catalog) implements PolarisEvent {}

service/common/src/main/java/org/apache/polaris/service/events/AwsCloudWatchEventListener.java

@@ -0,0 +1,228 @@
+/*
+ * 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.polaris.service.events;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.annotations.VisibleForTesting;
+import io.smallrye.common.annotation.Identifier;
+import jakarta.annotation.PostConstruct;
+import jakarta.annotation.PreDestroy;
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.inject.Inject;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import org.apache.polaris.core.context.CallContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.cloudwatchlogs.CloudWatchLogsClient;
+import software.amazon.awssdk.services.cloudwatchlogs.model.CreateLogGroupRequest;
+import software.amazon.awssdk.services.cloudwatchlogs.model.CreateLogStreamRequest;
+import software.amazon.awssdk.services.cloudwatchlogs.model.DescribeLogStreamsRequest;
+import software.amazon.awssdk.services.cloudwatchlogs.model.DescribeLogStreamsResponse;
+import software.amazon.awssdk.services.cloudwatchlogs.model.InputLogEvent;
+import software.amazon.awssdk.services.cloudwatchlogs.model.InvalidSequenceTokenException;
+import software.amazon.awssdk.services.cloudwatchlogs.model.LogStream;
+import software.amazon.awssdk.services.cloudwatchlogs.model.PutLogEventsRequest;
+import software.amazon.awssdk.services.cloudwatchlogs.model.PutLogEventsResponse;
+import software.amazon.awssdk.services.cloudwatchlogs.model.ResourceAlreadyExistsException;
+
+@ApplicationScoped
+@Identifier("aws-cloudwatch")
+public class AwsCloudWatchEventListener extends PolarisEventListener {
+  private static final Logger LOGGER = LoggerFactory.getLogger(AwsCloudWatchEventListener.class);
+  private final ObjectMapper objectMapper = new ObjectMapper();
+  private static final int MAX_BATCH_SIZE = 10_000;
+  private static final int MAX_WAIT_MS = 5000;
+
+  private final BlockingQueue<EventAndTimestamp> queue = new LinkedBlockingQueue<>();
+  private CloudWatchLogsClient client;
+  private volatile String sequenceToken;
+
+  private volatile boolean running = true;
+
+  ExecutorService executorService;
+
+  private Future<?> backgroundTask;
+
+  private final String logGroup;
+  private final String logStream;
+  private final Region region;
+
+  @Inject
+  public AwsCloudWatchEventListener(
+      EventListenerConfiguration config, ExecutorService executorService) {
+    this.executorService = executorService;
+
+    this.logStream = config.awsCloudwatchlogStream().orElse("polaris-cloudwatch-default-stream");
+    this.logGroup = config.awsCloudwatchlogGroup().orElse("polaris-cloudwatch-default-group");
+    this.region = Region.of(config.awsCloudwatchRegion().orElse("us-east-1"));
+  }
+
+  @PostConstruct
+  void start() {
+    this.client = createCloudWatchClient();
+    ensureLogGroupAndStream();
+    backgroundTask = executorService.submit(this::processQueue);
+  }
+
+  protected CloudWatchLogsClient createCloudWatchClient() {
+    return CloudWatchLogsClient.builder().region(region).build();
+  }
+
+  private void processQueue() {
+    while (running || !queue.isEmpty()) {
+      drainQueue();
+    }
+  }
+
+  @VisibleForTesting
+  public void drainQueue() {
+    List<EventAndTimestamp> drainedEvents = new ArrayList<>();
+    List<InputLogEvent> transformedEvents = new ArrayList<>();
+    try {
+      EventAndTimestamp first = queue.poll(MAX_WAIT_MS, TimeUnit.MILLISECONDS);
+
+      if (first != null) {
+        drainedEvents.add(first);
+        queue.drainTo(drainedEvents, MAX_BATCH_SIZE - 1);
+      } else {
+        return;
+      }
+
+      drainedEvents.forEach(event -> transformedEvents.add(createLogEvent(event)));
+
+      sendToCloudWatch(transformedEvents);
+    } catch (Exception e) {
+      LOGGER.error("Error writing logs to CloudWatch: {}", e.getMessage());
+      LOGGER.error("Events not logged: {}", transformedEvents);
+      queue.addAll(drainedEvents);
+    }
+  }
+
+  private InputLogEvent createLogEvent(EventAndTimestamp eventAndTimestamp) {
+    return InputLogEvent.builder()
+        .message(eventAndTimestamp.event)
+        .timestamp(eventAndTimestamp.timestamp)
+        .build();
+  }
+
+  private void sendToCloudWatch(List<InputLogEvent> events) {
+    events.sort(Comparator.comparingLong(InputLogEvent::timestamp));
+
+    PutLogEventsRequest.Builder requestBuilder =
+        PutLogEventsRequest.builder()
+            .logGroupName(logGroup)
+            .logStreamName(logStream)
+            .logEvents(events);
+
+    synchronized (this) {
+      if (sequenceToken != null) {
+        requestBuilder.sequenceToken(sequenceToken);
+      }
+
+      try {
+        PutLogEventsResponse response = client.putLogEvents(requestBuilder.build());
+        sequenceToken = response.nextSequenceToken();
+      } catch (InvalidSequenceTokenException e) {
+        sequenceToken = getSequenceToken();
+        requestBuilder.sequenceToken(sequenceToken);
+        PutLogEventsResponse retryResponse = client.putLogEvents(requestBuilder.build());
+        sequenceToken = retryResponse.nextSequenceToken();
+      }
+    }
+  }
+
+  private void ensureLogGroupAndStream() {
+    try {
+      client.createLogGroup(CreateLogGroupRequest.builder().logGroupName(logGroup).build());
+    } catch (ResourceAlreadyExistsException ignored) {
+    }
+
+    try {
+      client.createLogStream(
+          CreateLogStreamRequest.builder().logGroupName(logGroup).logStreamName(logStream).build());
+    } catch (ResourceAlreadyExistsException ignored) {
+    }
+
+    sequenceToken = getSequenceToken();
+  }
+
+  private String getSequenceToken() {
+    DescribeLogStreamsResponse response =
+        client.describeLogStreams(
+            DescribeLogStreamsRequest.builder()
+                .logGroupName(logGroup)
+                .logStreamNamePrefix(logStream)
+                .build());
+
+    return response.logStreams().stream()
+        .filter(s -> logStream.equals(s.logStreamName()))
+        .map(LogStream::uploadSequenceToken)
+        .filter(Objects::nonNull)
+        .findFirst()
+        .orElse(null);
+  }
+
+  @PreDestroy
+  void shutdown() {
+    running = false;
+    if (backgroundTask != null) {
+      try {
+        backgroundTask.get(10, TimeUnit.SECONDS);
+      } catch (Exception e) {
+        LOGGER.error("Error waiting for background logging task to finish: {}", e.getMessage());
+      }
+    }
+    if (client != null) {
+      client.close();
+    }
+  }
+
+  private record EventAndTimestamp(String event, long timestamp) {}
+
+  private long getCurrentTimestamp(CallContext callContext) {
+    return callContext.getPolarisCallContext().getClock().millis();
+  }
+
+  // Event overrides below
+  @Override
+  public void onAfterCatalogCreated(AfterCatalogCreatedEvent event, CallContext callContext) {
+    try {
+      Map<String, Object> json = objectMapper.convertValue(event.catalog(), Map.class);
+      json.put("realm", callContext.getRealmContext().getRealmIdentifier());
+      json.put("event_type", event.getClass().getSimpleName());
+      queue.add(
+          new EventAndTimestamp(
+              objectMapper.writeValueAsString(json), getCurrentTimestamp(callContext)));
+    } catch (JsonProcessingException e) {
+      LOGGER.error("Error processing event into JSON string: {}", e.getMessage());
+    }
+  }
+}