AWS CloudWatch Event Sink Implementation

gradle/libs.versions.toml

@@ -74,6 +74,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.4" }
+localstack = { module = "org.testcontainers:localstack", version = "1.19.7" }
 keycloak-admin-client = { module = "org.keycloak:keycloak-admin-client", version = "26.0.6" }
 logback-classic = { module = "ch.qos.logback:logback-classic", version = "1.5.18" }
 micrometer-bom = { module = "io.micrometer:micrometer-bom", version = "1.15.3" }

runtime/service/build.gradle.kts

@@ -103,6 +103,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("software.amazon.awssdk:apache-client") {
     exclude("commons-logging", "commons-logging")
   }
@@ -144,6 +145,9 @@ dependencies {
   testImplementation("io.quarkus:quarkus-rest-client")
   testImplementation("io.quarkus:quarkus-rest-client-jackson")
   testImplementation("io.rest-assured:rest-assured")
+  testImplementation(libs.localstack)
+  testImplementation("org.testcontainers:testcontainers")
+  testImplementation(project(":polaris-container-spec-helper"))
 
   testImplementation(libs.threeten.extra)
   testImplementation(libs.hawkular.agent.prometheus.scraper)

runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListenerConfiguration.java

@@ -25,8 +25,8 @@
 @ConfigMapping(prefix = "polaris.event-listener")
 public interface PolarisEventListenerConfiguration {
   /**
-   * 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();
 }

runtime/service/src/main/java/org/apache/polaris/service/events/jsonEventListener/PropertyMapEventListener.java

@@ -0,0 +1,43 @@
+/*
+ * 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.jsonEventListener;
+
+import java.util.HashMap;
+import org.apache.polaris.service.events.AfterTableRefreshedEvent;
+import org.apache.polaris.service.events.PolarisEventListener;
+
+/**
+ * This class provides a common framework for transforming Polaris events into a HashMap, which can
+ * be used to transform the event further, such as transforming into a JSON string, and send them to
+ * various destinations. Concrete implementations should override the
+ * {{@code @link#transformAndSendEvent(HashMap)}} method to define how the event data should be
+ * transformed into a JSON string, transmitted, and/or stored.
+ */
+public abstract class PropertyMapEventListener extends PolarisEventListener {
+  protected abstract void transformAndSendEvent(HashMap<String, Object> properties);
+
+  @Override
+  public void onAfterTableRefreshed(AfterTableRefreshedEvent event) {
+    HashMap<String, Object> properties = new HashMap<>();
+    properties.put("event_type", event.getClass().getSimpleName());
+    properties.put("table_identifier", event.tableIdentifier().toString());
+    transformAndSendEvent(properties);
+  }
+}

runtime/service/src/main/java/org/apache/polaris/service/events/jsonEventListener/aws/cloudwatch/AwsCloudWatchConfiguration.java

@@ -0,0 +1,31 @@
+/*
+ * 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.jsonEventListener.aws.cloudwatch;
+
+/** Configuration interface for AWS CloudWatch event listener settings. */
+public interface AwsCloudWatchConfiguration {
+  String awsCloudWatchLogGroup();
+
+  String awsCloudWatchLogStream();
+
+  String awsCloudWatchRegion();
+
+  boolean synchronousMode();
+}

runtime/service/src/main/java/org/apache/polaris/service/events/jsonEventListener/aws/cloudwatch/AwsCloudWatchEventListener.java

@@ -0,0 +1,190 @@
+/*
+ * 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.jsonEventListener.aws.cloudwatch;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.smallrye.common.annotation.Identifier;
+import jakarta.annotation.PostConstruct;
+import jakarta.annotation.PreDestroy;
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.inject.Inject;
+import jakarta.ws.rs.core.Context;
+import jakarta.ws.rs.core.SecurityContext;
+import java.time.Clock;
+import java.util.HashMap;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Supplier;
+import org.apache.polaris.core.auth.PolarisPrincipal;
+import org.apache.polaris.core.context.CallContext;
+import org.apache.polaris.service.config.PolarisIcebergObjectMapperCustomizer;
+import org.apache.polaris.service.events.jsonEventListener.PropertyMapEventListener;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.cloudwatchlogs.CloudWatchLogsAsyncClient;
+import software.amazon.awssdk.services.cloudwatchlogs.model.CreateLogGroupRequest;
+import software.amazon.awssdk.services.cloudwatchlogs.model.CreateLogStreamRequest;
+import software.amazon.awssdk.services.cloudwatchlogs.model.DescribeLogGroupsRequest;
+import software.amazon.awssdk.services.cloudwatchlogs.model.DescribeLogStreamsRequest;
+import software.amazon.awssdk.services.cloudwatchlogs.model.InputLogEvent;
+import software.amazon.awssdk.services.cloudwatchlogs.model.PutLogEventsRequest;
+import software.amazon.awssdk.services.cloudwatchlogs.model.PutLogEventsResponse;
+
+@ApplicationScoped
+@Identifier("aws-cloudwatch")
+public class AwsCloudWatchEventListener extends PropertyMapEventListener {
+  private static final Logger LOGGER = LoggerFactory.getLogger(AwsCloudWatchEventListener.class);
+  final ObjectMapper objectMapper = new ObjectMapper();
+
+  private CloudWatchLogsAsyncClient client;
+
+  private final String logGroup;
+  private final String logStream;
+  private final Region region;
+  private final boolean synchronousMode;
+  private final Clock clock;
+
+  @Inject CallContext callContext;
+
+  @Context SecurityContext securityContext;
+
+  @Inject
+  public AwsCloudWatchEventListener(
+      AwsCloudWatchConfiguration config,
+      Clock clock,
+      PolarisIcebergObjectMapperCustomizer customizer) {
+    this.logStream = config.awsCloudWatchLogStream();
+    this.logGroup = config.awsCloudWatchLogGroup();
+    this.region = Region.of(config.awsCloudWatchRegion());
+    this.synchronousMode = config.synchronousMode();
+    this.clock = clock;
+    customizer.customize(this.objectMapper);
+  }
+
+  @PostConstruct
+  void start() {
+    this.client = createCloudWatchAsyncClient();
+    ensureLogGroupAndStream();
+  }
+
+  protected CloudWatchLogsAsyncClient createCloudWatchAsyncClient() {
+    return CloudWatchLogsAsyncClient.builder().region(region).build();
+  }
+
+  private void ensureLogGroupAndStream() {
+    ensureResourceExists(
+        () ->
+            client
+                .describeLogGroups(
+                    DescribeLogGroupsRequest.builder().logGroupNamePrefix(logGroup).build())
+                .join()
+                .logGroups()
+                .stream()
+                .anyMatch(g -> g.logGroupName().equals(logGroup)),
+        () ->
+            client
+                .createLogGroup(CreateLogGroupRequest.builder().logGroupName(logGroup).build())
+                .join(),
+        "group",
+        logGroup);
+    ensureResourceExists(
+        () ->
+            client
+                .describeLogStreams(
+                    DescribeLogStreamsRequest.builder()
+                        .logGroupName(logGroup)
+                        .logStreamNamePrefix(logStream)
+                        .build())
+                .join()
+                .logStreams()
+                .stream()
+                .anyMatch(s -> s.logStreamName().equals(logStream)),
+        () ->
+            client
+                .createLogStream(
+                    CreateLogStreamRequest.builder()
+                        .logGroupName(logGroup)
+                        .logStreamName(logStream)
+                        .build())
+                .join(),
+        "stream",
+        logStream);
+  }
+
+  private static void ensureResourceExists(
+      Supplier<Boolean> existsCheck,
+      Runnable createAction,
+      String resourceType,
+      String resourceName) {
+    if (existsCheck.get()) {
+      LOGGER.debug("Log {} [{}] already exists", resourceType, resourceName);
+    } else {
+      LOGGER.debug("Attempting to create log {}: {}", resourceType, resourceName);
+      createAction.run();
+    }
+  }
+
+  @PreDestroy
+  void shutdown() {
+    if (client != null) {
+      client.close();
+      client = null;
+    }
+  }
+
+  @Override
+  protected void transformAndSendEvent(HashMap<String, Object> properties) {
+    properties.put("realm_id", callContext.getRealmContext().getRealmIdentifier());
+    properties.put("principal", securityContext.getUserPrincipal().getName());
+    properties.put(
+        "activated_roles", ((PolarisPrincipal) securityContext.getUserPrincipal()).getRoles());
+    // TODO: Add request ID when it is available
+    String eventAsJson;
+    try {
+      eventAsJson = objectMapper.writeValueAsString(properties);
+    } catch (JsonProcessingException e) {
+      LOGGER.error("Error processing event into JSON string: ", e);
+      LOGGER.debug("Failed to convert the following object into JSON string: {}", properties);
+      return;
+    }
+    InputLogEvent inputLogEvent =
+        InputLogEvent.builder().message(eventAsJson).timestamp(clock.millis()).build();
+    PutLogEventsRequest.Builder requestBuilder =
+        PutLogEventsRequest.builder()
+            .logGroupName(logGroup)
+            .logStreamName(logStream)
+            .logEvents(List.of(inputLogEvent));
+    CompletableFuture<PutLogEventsResponse> future =
+        client
+            .putLogEvents(requestBuilder.build())
+            .whenComplete(
+                (resp, err) -> {
+                  if (err != null) {
+                    LOGGER.error(
+                        "Error writing log to CloudWatch. Event: {}, Error: ", inputLogEvent, err);
+                  }
+                });
+    if (synchronousMode) {
+      future.join();
+    }
+  }
+}