[#5556] feat(iceberg) support audit framework for Iceberg REST server (#5580)

### What changes were proposed in this pull request?

1. expand `BaseEvent` to represent general event information, like
`OperationType` `OperationStatus`, etc.
2. add `SimpleFormatterV2` to convert all `Event` (include
GravitinoServerEvent and Iceberg REST Event) to an audit object

### Why are the changes needed?

Fix: #5556 

### Does this PR introduce _any_ user-facing change?

For Gravitino server:
1. If user use default audit configuration, will add `remoteAddr` and
`eventSource` to the audit log
2. If user use custom audit writer or formatter, there is no affect.

For Gravitino Iceberg REST server:
new feature, add configuration.

### How was this patch tested?

add `gravitino.audit.enabled = true`, check audit log content in
Gravitino server and Gravitino IcebergRESTServer.

Author: FANNG1

core/src/main/java/org/apache/gravitino/Configs.java

@@ -24,7 +24,7 @@
 import java.util.List;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.gravitino.audit.FileAuditWriter;
-import org.apache.gravitino.audit.SimpleFormatter;
+import org.apache.gravitino.audit.v2.SimpleFormatterV2;
 import org.apache.gravitino.config.ConfigBuilder;
 import org.apache.gravitino.config.ConfigConstants;
 import org.apache.gravitino.config.ConfigEntry;
@@ -366,5 +366,5 @@ private Configs() {}
           .doc("Gravitino event log formatter class name")
           .version(ConfigConstants.VERSION_0_7_0)
           .stringConf()
-          .createWithDefault(SimpleFormatter.class.getName());
+          .createWithDefault(SimpleFormatterV2.class.getName());
 }

core/src/main/java/org/apache/gravitino/audit/AuditLog.java

@@ -19,6 +19,8 @@
 
 package org.apache.gravitino.audit;
 
+import com.google.common.collect.ImmutableMap;
+import java.util.Map;
 import org.apache.gravitino.listener.api.event.AlterCatalogEvent;
 import org.apache.gravitino.listener.api.event.AlterCatalogFailureEvent;
 import org.apache.gravitino.listener.api.event.AlterFilesetEvent;
@@ -56,6 +58,7 @@
 import org.apache.gravitino.listener.api.event.DropTopicEvent;
 import org.apache.gravitino.listener.api.event.DropTopicFailureEvent;
 import org.apache.gravitino.listener.api.event.Event;
+import org.apache.gravitino.listener.api.event.EventSource;
 import org.apache.gravitino.listener.api.event.GetFileLocationEvent;
 import org.apache.gravitino.listener.api.event.GetFileLocationFailureEvent;
 import org.apache.gravitino.listener.api.event.GetPartitionEvent;
@@ -86,6 +89,8 @@
 import org.apache.gravitino.listener.api.event.LoadTableFailureEvent;
 import org.apache.gravitino.listener.api.event.LoadTopicEvent;
 import org.apache.gravitino.listener.api.event.LoadTopicFailureEvent;
+import org.apache.gravitino.listener.api.event.OperationStatus;
+import org.apache.gravitino.listener.api.event.OperationType;
 import org.apache.gravitino.listener.api.event.PartitionExistsEvent;
 import org.apache.gravitino.listener.api.event.PartitionExistsFailureEvent;
 import org.apache.gravitino.listener.api.event.PurgePartitionEvent;
@@ -106,7 +111,9 @@ public interface AuditLog {
    * The operation name.
    *
    * @return operation name.
+   * @deprecated use {@code #operationType()} instead.
    */
+  @Deprecated
   Operation operation();
 
   /**
@@ -127,9 +134,61 @@ public interface AuditLog {
    * The status of the operation.
    *
    * @return operation status.
+   * @deprecated use {@link #operationStatus()} instead.
    */
+  @Deprecated
   Status status();
 
+  /**
+   * The remote address of the operation.
+   *
+   * @return The remote address string.
+   * @since 0.8.0
+   */
+  default String remoteAddress() {
+    return "unknown";
+  }
+
+  /**
+   * The status of the operation.
+   *
+   * @return The operation status.
+   * @since 0.8.0
+   */
+  default OperationStatus operationStatus() {
+    return OperationStatus.UNKNOWN;
+  }
+
+  /**
+   * The type of the operation.
+   *
+   * @return The operation status.
+   * @since 0.8.0
+   */
+  default OperationType operationType() {
+    return OperationType.UNKNOWN;
+  }
+
+  /**
+   * The custom information.
+   *
+   * @return the custom information.
+   * @since 0.8.0
+   */
+  default Map<String, String> customInfo() {
+    return ImmutableMap.of();
+  }
+
+  /**
+   * The event source.
+   *
+   * @return the event source.
+   * @since 0.8.0
+   */
+  default EventSource eventSource() {
+    return EventSource.GRAVITINO_SERVER;
+  }
+
   /** Define user metadata operation. */
   enum Operation {
     CREATE_METALAKE,
@@ -294,6 +353,7 @@ public static Operation fromEvent(Event event) {
 
   enum Status {
     SUCCESS,
-    FAILURE
+    FAILURE,
+    UNKNOWN
   }
 }

core/src/main/java/org/apache/gravitino/audit/SimpleAuditLog.java

@@ -23,8 +23,14 @@
 import javax.annotation.Nullable;
 import lombok.Builder;
 
-/** The default implementation of the audit log. */
+/**
+ * The first version of implementation of the audit log.
+ *
+ * @deprecated since 0.8.0
+ */
 @Builder
+@Deprecated
+@SuppressWarnings("deprecation")
 public class SimpleAuditLog implements AuditLog {
 
   private String user;

core/src/main/java/org/apache/gravitino/audit/SimpleFormatter.java

@@ -23,10 +23,16 @@
 import org.apache.gravitino.listener.api.event.Event;
 import org.apache.gravitino.listener.api.event.FailureEvent;
 
-/** The default formatter implementation of the audit log. */
+/**
+ * The first version of formatter implementation of the audit log.
+ *
+ * @deprecated since 0.8.0, please use {@link org.apache.gravitino.audit.v2.SimpleFormatterV2}
+ */
+@Deprecated
 public class SimpleFormatter implements Formatter {
 
   @Override
+  @SuppressWarnings("deprecation")
   public SimpleAuditLog format(Event event) {
     Status status = event instanceof FailureEvent ? Status.FAILURE : Status.SUCCESS;
     return SimpleAuditLog.builder()

core/src/main/java/org/apache/gravitino/audit/v2/CompatibilityUtils.java

@@ -0,0 +1,100 @@
+/*
+ *  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.gravitino.audit.v2;
+
+import com.google.common.collect.ImmutableMap;
+import org.apache.gravitino.audit.AuditLog.Operation;
+import org.apache.gravitino.audit.AuditLog.Status;
+import org.apache.gravitino.listener.api.event.OperationStatus;
+import org.apache.gravitino.listener.api.event.OperationType;
+
+public class CompatibilityUtils {
+
+  private static ImmutableMap<OperationType, Operation> operationTypeMap =
+      ImmutableMap.<OperationType, Operation>builder()
+          // Metalake operation
+          .put(OperationType.CREATE_METALAKE, Operation.CREATE_METALAKE)
+          .put(OperationType.ALTER_METALAKE, Operation.ALTER_METALAKE)
+          .put(OperationType.DROP_METALAKE, Operation.DROP_METALAKE)
+          .put(OperationType.LOAD_METALAKE, Operation.LOAD_METALAKE)
+          .put(OperationType.LIST_METALAKE, Operation.LIST_METALAKE)
+
+          // Catalog operation
+          .put(OperationType.CREATE_CATALOG, Operation.CREATE_CATALOG)
+          .put(OperationType.ALTER_CATALOG, Operation.ALTER_CATALOG)
+          .put(OperationType.DROP_CATALOG, Operation.DROP_CATALOG)
+          .put(OperationType.LOAD_CATALOG, Operation.LOAD_CATALOG)
+          .put(OperationType.LIST_CATALOG, Operation.LIST_CATALOG)
+
+          // Schema operation
+          .put(OperationType.CREATE_SCHEMA, Operation.CREATE_SCHEMA)
+          .put(OperationType.ALTER_SCHEMA, Operation.ALTER_SCHEMA)
+          .put(OperationType.DROP_SCHEMA, Operation.DROP_SCHEMA)
+          .put(OperationType.LOAD_SCHEMA, Operation.LOAD_SCHEMA)
+          .put(OperationType.LIST_SCHEMA, Operation.LIST_SCHEMA)
+
+          // Table operation
+          .put(OperationType.CREATE_TABLE, Operation.CREATE_TABLE)
+          .put(OperationType.ALTER_TABLE, Operation.ALTER_TABLE)
+          .put(OperationType.DROP_TABLE, Operation.DROP_TABLE)
+          .put(OperationType.PURGE_TABLE, Operation.PURGE_TABLE)
+          .put(OperationType.LOAD_TABLE, Operation.LOAD_TABLE)
+          .put(OperationType.TABLE_EXISTS, Operation.UNKNOWN_OPERATION)
+          .put(OperationType.LIST_TABLE, Operation.LIST_TABLE)
+
+          // Partition operation
+          .put(OperationType.ADD_PARTITION, Operation.UNKNOWN_OPERATION)
+          .put(OperationType.DROP_PARTITION, Operation.UNKNOWN_OPERATION)
+          .put(OperationType.PURGE_PARTITION, Operation.PURGE_PARTITION)
+          .put(OperationType.LOAD_PARTITION, Operation.GET_PARTITION)
+          .put(OperationType.PARTITION_EXISTS, Operation.PARTITION_EXIST)
+          .put(OperationType.LIST_PARTITION, Operation.LIST_PARTITION)
+          .put(OperationType.LIST_PARTITION_NAMES, Operation.LIST_PARTITION)
+
+          // Fileset operation
+          .put(OperationType.CREATE_FILESET, Operation.CREATE_FILESET)
+          .put(OperationType.ALTER_FILESET, Operation.ALTER_FILESET)
+          .put(OperationType.DROP_FILESET, Operation.DROP_FILESET)
+          .put(OperationType.LOAD_FILESET, Operation.LOAD_FILESET)
+          .put(OperationType.LIST_FILESET, Operation.LIST_FILESET)
+          .put(OperationType.GET_FILESET_LOCATION, Operation.GET_FILE_LOCATION)
+
+          // Topic operation
+          .put(OperationType.CREATE_TOPIC, Operation.CREATE_TOPIC)
+          .put(OperationType.ALTER_TOPIC, Operation.ALTER_TOPIC)
+          .put(OperationType.DROP_TOPIC, Operation.DROP_TOPIC)
+          .put(OperationType.LOAD_TOPIC, Operation.LOAD_TOPIC)
+          .put(OperationType.LIST_TOPIC, Operation.LIST_TOPIC)
+          .build();
+
+  static Operation toAuditLogOperation(OperationType operationType) {
+    return operationTypeMap.getOrDefault(operationType, Operation.UNKNOWN_OPERATION);
+  }
+
+  static Status toAuditLogStatus(OperationStatus operationStatus) {
+    if (operationStatus.equals(OperationStatus.SUCCESS)) {
+      return Status.SUCCESS;
+    } else if (operationStatus.equals(OperationStatus.FAILURE)) {
+      return Status.FAILURE;
+    } else {
+      return Status.UNKNOWN;
+    }
+  }
+}

core/src/main/java/org/apache/gravitino/audit/v2/SimpleAuditLogV2.java

@@ -0,0 +1,108 @@
+/*
+ *  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.gravitino.audit.v2;
+
+import java.text.SimpleDateFormat;
+import java.util.Map;
+import java.util.Optional;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.audit.AuditLog;
+import org.apache.gravitino.listener.api.event.BaseEvent;
+import org.apache.gravitino.listener.api.event.EventSource;
+import org.apache.gravitino.listener.api.event.OperationStatus;
+import org.apache.gravitino.listener.api.event.OperationType;
+
+/**
+ * Compared to {@link org.apache.gravitino.audit.SimpleAuditLog}, adds audit log for Iceberg REST
+ * server, add eventSource and remoteAddress to audit log.
+ */
+public class SimpleAuditLogV2 implements AuditLog {
+
+  private final BaseEvent event;
+
+  public SimpleAuditLogV2(BaseEvent event) {
+    this.event = event;
+  }
+
+  @Override
+  public String user() {
+    return event.user();
+  }
+
+  @Override
+  @SuppressWarnings("deprecation")
+  public Operation operation() {
+    return CompatibilityUtils.toAuditLogOperation(event.operationType());
+  }
+
+  @Override
+  public String identifier() {
+    return Optional.ofNullable(event.identifier()).map(NameIdentifier::toString).orElse(null);
+  }
+
+  @Override
+  public long timestamp() {
+    return event.eventTime();
+  }
+
+  @Override
+  @SuppressWarnings("deprecation")
+  public Status status() {
+    return CompatibilityUtils.toAuditLogStatus(event.operationStatus());
+  }
+
+  @Override
+  public String remoteAddress() {
+    return event.remoteAddress();
+  }
+
+  @Override
+  public OperationStatus operationStatus() {
+    return event.operationStatus();
+  }
+
+  @Override
+  public OperationType operationType() {
+    return event.operationType();
+  }
+
+  @Override
+  public EventSource eventSource() {
+    return event.eventSource();
+  }
+
+  @Override
+  public Map<String, String> customInfo() {
+    return event.customInfo();
+  }
+
+  @Override
+  public String toString() {
+    return String.format(
+        "[%s]\t%s\t%s\t%s\t%s\t%s\t%s",
+        new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(timestamp()),
+        user(),
+        operationType(),
+        identifier(),
+        operationStatus(),
+        eventSource(),
+        remoteAddress());
+  }
+}

core/src/main/java/org/apache/gravitino/audit/v2/SimpleFormatterV2.java

@@ -0,0 +1,32 @@
+/*
+ *  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.gravitino.audit.v2;
+
+import org.apache.gravitino.audit.Formatter;
+import org.apache.gravitino.listener.api.event.Event;
+
+/** The default formatter implementation of the audit log. */
+public class SimpleFormatterV2 implements Formatter {
+
+  @Override
+  public SimpleAuditLogV2 format(Event event) {
+    return new SimpleAuditLogV2(event);
+  }
+}