Merge pull request #15782 from cdapio/feature/stackdriver-integration

[CDAP-21104] Add CompositeLogAppender with SPI Integration and Guice Configuration for LogPublisher

cdap-common/src/main/java/io/cdap/cdap/common/conf/Constants.java

@@ -1505,6 +1505,12 @@ public static final class Logging {
     public static final String LOG_APPENDER_EXT_DIR = "app.program.log.appender.extensions.dir";
     public static final String LOG_APPENDER_PROPERTY_PREFIX = "app.program.log.appender.system.properties.";
 
+    // Log publisher configs.
+    public static final String LOG_PUBLISHER_PROVIDER = "log.publisher.provider";
+    public static final String LOG_PUBLISHER_ENABLED = "log.publisher.enabled";
+    public static final String LOG_PUBLISHER_EXT_DIR = "log.publisher.extensions.dir";
+    public static final String LOG_PUBLISHER_PREFIX = "log.publisher";
+
     // Property key in the logger context to indicate it is performing pipeline validation
     public static final String PIPELINE_VALIDATION = "log.pipeline.validation";
 
@@ -2557,7 +2563,7 @@ public static final class Operation {
 
 
   /**
-   * Constants for Data Plane Audit Logging
+   * Constants for Data Plane Audit Logging.
    */
   public static final class AuditLogging {
     public static final String AUDIT_LOG_PUBLISH_INTERVAL_SECONDS = "auditlog.publish.interval.seconds";

cdap-common/src/main/resources/cdap-default.xml

@@ -2804,6 +2804,16 @@
     </description>
   </property>
 
+  <property>
+    <name>log.publisher.enabled</name>
+    <value>false</value>
+  </property>
+
+  <property>
+    <name>log.publisher.extensions.dir</name>
+    <value>/opt/cdap/master/ext/log-publisher</value>
+  </property>
+
   <!-- Metrics Configuration -->
 
   <property>

cdap-log-publisher-spi/pom.xml

@@ -0,0 +1,37 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  Copyright © 2025 Cask Data, Inc.
+  Licensed 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.
+  -->
+
+<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+  xmlns="http://maven.apache.org/POM/4.0.0"
+  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+
+  <modelVersion>4.0.0</modelVersion>
+
+  <parent>
+    <artifactId>cdap</artifactId>
+    <groupId>io.cdap.cdap</groupId>
+    <version>6.11.0-SNAPSHOT</version>
+  </parent>
+
+  <artifactId>cdap-log-publisher-spi</artifactId>
+  <name>CDAP Log Publisher SPI</name>
+  <packaging>jar</packaging>
+
+  <dependencies>
+    <dependency>
+      <artifactId>logback-classic</artifactId>
+      <groupId>ch.qos.logback</groupId>
+    </dependency>
+  </dependencies>
+</project>

cdap-log-publisher-spi/src/main/java/io/cdap/cdap/spi/logs/LogPublisher.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright © 2025 Cask Data, Inc.
+ *
+ * Licensed 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 io.cdap.cdap.spi.logs;
+
+import ch.qos.logback.classic.spi.ILoggingEvent;
+
+/**
+ * Publishes log events to a destination.
+ */
+public interface LogPublisher extends AutoCloseable {
+
+  /**
+   * Returns the name of the log publisher.
+   *
+   * @return the name of the publisher.
+   */
+  String getName();
+
+  /**
+   * Publishes a logging event.
+   *
+   * @param event the logging event to publish.
+   */
+  void publish(ILoggingEvent event);
+
+  /**
+   * Initializes the log publisher with the required context.
+   *
+   * @param context the context to initialize the publisher with.
+   */
+  void initialize(LogPublisherContext context);
+
+  /**
+   * Closes the log publisher and releases associated resources.
+   */
+  @Override
+  void close();
+}

cdap-log-publisher-spi/src/main/java/io/cdap/cdap/spi/logs/LogPublisherContext.java

@@ -0,0 +1,33 @@
+/*
+ * Copyright © 2025 Cask Data, Inc.
+ *
+ * Licensed 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 io.cdap.cdap.spi.logs;
+
+import java.util.Map;
+
+/**
+ * Provides context information for {@link LogPublisher} initialization.
+ */
+public interface LogPublisherContext {
+
+  /**
+   * Properties are derived from the CDAP configuration. Configuration file path will be added as an
+   * entry in the  properties.
+   *
+   * @return unmodifiable properties for the log publisher.
+   */
+  Map<String, String> getProperties();
+}

cdap-watchdog/pom.xml

@@ -44,6 +44,11 @@
       <type>test-jar</type>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>io.cdap.cdap</groupId>
+      <artifactId>cdap-log-publisher-spi</artifactId>
+      <version>${project.version}</version>
+    </dependency>
     <dependency>
       <groupId>io.cdap.cdap</groupId>
       <artifactId>cdap-watchdog-api</artifactId>
@@ -138,6 +143,14 @@
       <artifactId>log4j-over-slf4j</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.powermock</groupId>
+      <artifactId>powermock-api-mockito2</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.powermock</groupId>
+      <artifactId>powermock-module-junit4</artifactId>
+    </dependency>
   </dependencies>
 
 </project>

cdap-watchdog/src/main/java/io/cdap/cdap/logging/appender/CompositeLogAppender.java

@@ -0,0 +1,95 @@
+/*
+ * Copyright © 2025 Cask Data, Inc.
+ *
+ * Licensed 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 io.cdap.cdap.logging.appender;
+
+import ch.qos.logback.core.Context;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.function.Consumer;
+
+/**
+ * A log appender that delegates logging events to a list of other log appenders.
+ */
+public class CompositeLogAppender extends LogAppender {
+
+  private final List<LogAppender> appenders;
+
+  /**
+   * Constructs a CompositeLogAppender with the given list of appenders.
+   *
+   * @param appenders the list of {@link LogAppender} instances to delegate logging events to
+   */
+  public CompositeLogAppender(List<LogAppender> appenders) {
+    if (appenders == null) {
+      throw new IllegalArgumentException("Appenders list cannot be null");
+    }
+
+    // Make the appenders list unmodifiable to ensure thread safety and prevent external
+    // modifications.
+    this.appenders = Collections.unmodifiableList(new ArrayList<>(appenders));
+    setName(getClass().getName());
+  }
+
+  @Override
+  public void start() {
+    executeOnAppenders("start", LogAppender::start);
+    super.start();
+  }
+
+  @Override
+  public void stop() {
+    super.stop();
+    executeOnAppenders("stop", LogAppender::stop);
+  }
+
+  @Override
+  protected void appendEvent(LogMessage logMessage) {
+    executeOnAppenders("appendEvent", appender -> appender.appendEvent(logMessage));
+  }
+
+  @Override
+  public void setContext(Context context) {
+    executeOnAppenders("setContext", appender -> appender.setContext(context));
+    super.setContext(context);
+  }
+
+  /**
+   * Executes a specified action on all appenders and handles any exceptions.
+   *
+   * @param actionName the name of the action being performed (for logging purposes)
+   * @param action     the action to execute on each appender
+   * @throws RuntimeException if one or more appenders fail to perform the action
+   */
+  private void executeOnAppenders(String actionName, Consumer<LogAppender> action) {
+    RuntimeException exceptions = null;
+    for (LogAppender appender : appenders) {
+      try {
+        action.accept(appender);
+      } catch (Exception e) {
+        if (exceptions == null) {
+          exceptions = new RuntimeException("One or more appenders failed to " + actionName);
+        }
+        exceptions.addSuppressed(e);
+      }
+    }
+
+    if (exceptions != null) {
+      throw exceptions;
+    }
+  }
+}