BP-69: Allow caller to pass a parent slog Logger to created/opened ledgers

bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerCreateOp.java

@@ -22,6 +22,7 @@
 package org.apache.bookkeeper.client;
 
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+import io.github.merlimat.slog.Logger;
 import java.security.GeneralSecurityException;
 import java.util.Collections;
 import java.util.EnumSet;
@@ -73,6 +74,7 @@ class LedgerCreateOp {
     final long startTime;
     final OpStatsLogger createOpLogger;
     final BookKeeperClientStats clientStats;
+    final Logger parentLogger;
     boolean adv = false;
     boolean generateLedgerId = true;
 
@@ -104,6 +106,16 @@ class LedgerCreateOp {
             byte[] passwd, CreateCallback cb, Object ctx, final Map<String, byte[]> customMetadata,
             EnumSet<WriteFlag> writeFlags,
             BookKeeperClientStats clientStats) {
+        this(bk, ensembleSize, writeQuorumSize, ackQuorumSize, digestType, passwd, cb, ctx,
+                customMetadata, writeFlags, clientStats, null);
+    }
+
+    LedgerCreateOp(
+            BookKeeper bk, int ensembleSize, int writeQuorumSize, int ackQuorumSize, DigestType digestType,
+            byte[] passwd, CreateCallback cb, Object ctx, final Map<String, byte[]> customMetadata,
+            EnumSet<WriteFlag> writeFlags,
+            BookKeeperClientStats clientStats,
+            Logger parentLogger) {
         this.bk = bk;
         this.metadataFormatVersion = bk.getConf().getLedgerMetadataFormatVersion();
         this.ensembleSize = ensembleSize;
@@ -118,6 +130,7 @@ class LedgerCreateOp {
         this.startTime = MathUtils.nowInNano();
         this.createOpLogger = clientStats.getCreateOpLogger();
         this.clientStats = clientStats;
+        this.parentLogger = parentLogger;
     }
 
     /**
@@ -242,9 +255,10 @@ private void metadataCallback(Versioned<LedgerMetadata> writtenMetadata,
             try {
                 if (adv) {
                     lh = new LedgerHandleAdv(bk.getClientCtx(), ledgerId, writtenMetadata,
-                                             digestType, passwd, writeFlags);
+                                             digestType, passwd, writeFlags, parentLogger);
                 } else {
-                    lh = new LedgerHandle(bk.getClientCtx(), ledgerId, writtenMetadata, digestType, passwd, writeFlags);
+                    lh = new LedgerHandle(bk.getClientCtx(), ledgerId, writtenMetadata, digestType, passwd, writeFlags,
+                                          parentLogger);
                 }
             } catch (GeneralSecurityException e) {
                 log.error()
@@ -302,6 +316,7 @@ public static class CreateBuilderImpl implements CreateBuilder {
         private org.apache.bookkeeper.client.api.DigestType builderDigestType =
             org.apache.bookkeeper.client.api.DigestType.CRC32;
         private Map<String, byte[]> builderCustomMetadata = Collections.emptyMap();
+        private Logger builderParentLogger;
 
         CreateBuilderImpl(BookKeeper bk) {
             this.bk = bk;
@@ -350,6 +365,12 @@ public CreateBuilder withDigestType(org.apache.bookkeeper.client.api.DigestType
             return this;
         }
 
+        @Override
+        public CreateBuilder withLoggerContext(Logger parentLogger) {
+            this.builderParentLogger = parentLogger;
+            return this;
+        }
+
         @Override
         public CreateAdvBuilder makeAdv() {
             return new CreateAdvBuilderImpl(this);
@@ -416,7 +437,7 @@ private void create(CreateCallback cb) {
             LedgerCreateOp op = new LedgerCreateOp(bk, builderEnsembleSize,
                 builderWriteQuorumSize, builderAckQuorumSize, DigestType.fromApiDigestType(builderDigestType),
                 builderPassword, cb, null, builderCustomMetadata, builderWriteFlags,
-                bk.getClientCtx().getClientStats());
+                bk.getClientCtx().getClientStats(), builderParentLogger);
             ReentrantReadWriteLock closeLock = bk.getCloseLock();
             closeLock.readLock().lock();
             try {
@@ -477,7 +498,8 @@ private void create(CreateCallback cb) {
                     DigestType.fromApiDigestType(parent.builderDigestType),
                     parent.builderPassword, cb, null, parent.builderCustomMetadata,
                     parent.builderWriteFlags,
-                    parent.bk.getClientCtx().getClientStats());
+                    parent.bk.getClientCtx().getClientStats(),
+                    parent.builderParentLogger);
             ReentrantReadWriteLock closeLock = parent.bk.getCloseLock();
             closeLock.readLock().lock();
             try {

bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java

@@ -32,6 +32,7 @@
 import com.google.common.collect.Sets;
 import com.google.common.util.concurrent.RateLimiter;
 import io.github.merlimat.slog.Logger;
+import io.github.merlimat.slog.LoggerBuilder;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
 import java.security.GeneralSecurityException;
@@ -180,7 +181,20 @@ private enum HandleState {
                  BookKeeper.DigestType digestType, byte[] password,
                  EnumSet<WriteFlag> writeFlags)
             throws GeneralSecurityException, NumberFormatException {
-        this.log = Logger.get(LedgerHandle.class).with().attr("ledgerId", ledgerId).build();
+        this(clientCtx, ledgerId, versionedMetadata, digestType, password, writeFlags, null);
+    }
+
+    LedgerHandle(ClientContext clientCtx,
+                 long ledgerId, Versioned<LedgerMetadata> versionedMetadata,
+                 BookKeeper.DigestType digestType, byte[] password,
+                 EnumSet<WriteFlag> writeFlags,
+                 Logger parentLogger)
+            throws GeneralSecurityException, NumberFormatException {
+        LoggerBuilder builder = Logger.get(LedgerHandle.class).with();
+        if (parentLogger != null) {
+            builder = builder.ctx(parentLogger);
+        }
+        this.log = builder.attr("ledgerId", ledgerId).build();
         this.clientCtx = clientCtx;
 
         this.versionedMetadata = versionedMetadata;

bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandleAdv.java

@@ -21,6 +21,7 @@
 
 package org.apache.bookkeeper.client;
 
+import io.github.merlimat.slog.Logger;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
 import java.io.Serializable;
@@ -56,7 +57,15 @@ public int compare(PendingAddOp o1, PendingAddOp o2) {
                     long ledgerId, Versioned<LedgerMetadata> metadata,
                     BookKeeper.DigestType digestType, byte[] password, EnumSet<WriteFlag> writeFlags)
             throws GeneralSecurityException, NumberFormatException {
-        super(clientCtx, ledgerId, metadata, digestType, password, writeFlags);
+        this(clientCtx, ledgerId, metadata, digestType, password, writeFlags, null);
+    }
+
+    LedgerHandleAdv(ClientContext clientCtx,
+                    long ledgerId, Versioned<LedgerMetadata> metadata,
+                    BookKeeper.DigestType digestType, byte[] password, EnumSet<WriteFlag> writeFlags,
+                    Logger parentLogger)
+            throws GeneralSecurityException, NumberFormatException {
+        super(clientCtx, ledgerId, metadata, digestType, password, writeFlags, parentLogger);
         pendingAddOps = new PriorityBlockingQueue<PendingAddOp>(10, new PendingOpsComparator());
     }
 

bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerOpenOp.java

@@ -24,6 +24,7 @@
 import static org.apache.bookkeeper.client.BookKeeper.DigestType.fromApiDigestType;
 
 import io.github.merlimat.slog.Logger;
+import io.github.merlimat.slog.LoggerBuilder;
 import java.security.GeneralSecurityException;
 import java.util.Arrays;
 import java.util.concurrent.CompletableFuture;
@@ -75,6 +76,7 @@ class LedgerOpenOp {
 
     final DigestType suggestedDigestType;
     final boolean enableDigestAutodetection;
+    final Logger parentLogger;
 
     /**
      * Constructor.
@@ -89,7 +91,17 @@ class LedgerOpenOp {
     public LedgerOpenOp(BookKeeper bk, BookKeeperClientStats clientStats,
                         long ledgerId, DigestType digestType, byte[] passwd,
                         OpenCallback cb, Object ctx) {
-        this.log = Logger.get(LedgerOpenOp.class).with().attr("ledgerId", ledgerId).build();
+        this(bk, clientStats, ledgerId, digestType, passwd, cb, ctx, null);
+    }
+
+    public LedgerOpenOp(BookKeeper bk, BookKeeperClientStats clientStats,
+                        long ledgerId, DigestType digestType, byte[] passwd,
+                        OpenCallback cb, Object ctx, Logger parentLogger) {
+        LoggerBuilder builder = Logger.get(LedgerOpenOp.class).with();
+        if (parentLogger != null) {
+            builder = builder.ctx(parentLogger);
+        }
+        this.log = builder.attr("ledgerId", ledgerId).build();
         this.bk = bk;
         this.ledgerId = ledgerId;
         this.passwd = passwd;
@@ -98,6 +110,7 @@ public LedgerOpenOp(BookKeeper bk, BookKeeperClientStats clientStats,
         this.enableDigestAutodetection = bk.getConf().getEnableDigestTypeAutodetection();
         this.suggestedDigestType = digestType;
         this.openOpLogger = clientStats.getOpenOpLogger();
+        this.parentLogger = parentLogger;
     }
 
     public LedgerOpenOp(BookKeeper bk, BookKeeperClientStats clientStats,
@@ -113,6 +126,7 @@ public LedgerOpenOp(BookKeeper bk, BookKeeperClientStats clientStats,
         this.enableDigestAutodetection = false;
         this.suggestedDigestType = bk.conf.getBookieRecoveryDigestType();
         this.openOpLogger = clientStats.getOpenOpLogger();
+        this.parentLogger = null;
     }
 
     /**
@@ -215,7 +229,7 @@ private void openWithMetadata(Versioned<LedgerMetadata> versionedMetadata) {
             // Therefore, if a user needs to the feature that update metadata automatically, he will set
             // "keepUpdateMetadata" to "true",
             lh = new ReadOnlyLedgerHandle(bk.getClientCtx(), ledgerId, versionedMetadata, digestType,
-                                          passwd, watchImmediately);
+                                          passwd, watchImmediately, parentLogger);
         } catch (GeneralSecurityException e) {
             log.error().exception(e).attr("ledgerId", ledgerId).log("Security exception while opening ledger");
             openComplete(BKException.Code.DigestNotInitializedException, null);
@@ -335,7 +349,7 @@ private void open(OpenCallback cb) {
 
             LedgerOpenOp op = new LedgerOpenOp(bk, bk.getClientCtx().getClientStats(),
                                                ledgerId, fromApiDigestType(digestType),
-                                               password, cb, null);
+                                               password, cb, null, parentLogger);
             ReentrantReadWriteLock closeLock = bk.getCloseLock();
             closeLock.readLock().lock();
             try {

bookkeeper-server/src/main/java/org/apache/bookkeeper/client/ReadOnlyLedgerHandle.java

@@ -23,6 +23,7 @@
 import static com.google.common.base.Preconditions.checkState;
 
 import com.google.common.annotations.VisibleForTesting;
+import io.github.merlimat.slog.Logger;
 import java.security.GeneralSecurityException;
 import java.util.List;
 import java.util.Map;
@@ -95,7 +96,16 @@ public String toString() {
                          BookKeeper.DigestType digestType, byte[] password,
                          boolean watchImmediately)
             throws GeneralSecurityException, NumberFormatException {
-        super(clientCtx, ledgerId, metadata, digestType, password, WriteFlag.NONE);
+        this(clientCtx, ledgerId, metadata, digestType, password, watchImmediately, null);
+    }
+
+    ReadOnlyLedgerHandle(ClientContext clientCtx,
+                         long ledgerId, Versioned<LedgerMetadata> metadata,
+                         BookKeeper.DigestType digestType, byte[] password,
+                         boolean watchImmediately,
+                         Logger parentLogger)
+            throws GeneralSecurityException, NumberFormatException {
+        super(clientCtx, ledgerId, metadata, digestType, password, WriteFlag.NONE, parentLogger);
         if (watchImmediately) {
             registerLedgerMetadataListener();
         }

bookkeeper-server/src/main/java/org/apache/bookkeeper/client/api/CreateBuilder.java

@@ -20,6 +20,7 @@
  */
 package org.apache.bookkeeper.client.api;
 
+import io.github.merlimat.slog.Logger;
 import java.util.Arrays;
 import java.util.EnumSet;
 import java.util.Map;
@@ -122,4 +123,22 @@ default CreateBuilder withWriteFlags(WriteFlag ... writeFlags) {
      */
     CreateAdvBuilder makeAdv();
 
+    /**
+     * Inherit the context attributes of the given slog {@link Logger} on the logger bound to
+     * the resulting {@link WriteHandle}. Every log statement emitted by the handle (and by the create-time machinery
+     * that produces it) will carry the parent logger's context attributes, in addition to the {@code ledgerId}
+     * attribute that is always added by the client.
+     *
+     * <p>Useful for correlating bookkeeper-client log output with the application's own request / tenant / trace
+     * identifiers — typically the application has built a per-request logger via
+     * {@code Logger.get(...).with().attr(...)...build()} and passes it here.
+     *
+     * @param parentLogger logger whose context attributes to inherit; {@code null} is treated as no extra context
+     *
+     * @return the builder itself
+     */
+    default CreateBuilder withLoggerContext(Logger parentLogger) {
+        return this;
+    }
+
 }

bookkeeper-server/src/main/java/org/apache/bookkeeper/client/api/OpenBuilder.java

@@ -20,6 +20,7 @@
  */
 package org.apache.bookkeeper.client.api;
 
+import io.github.merlimat.slog.Logger;
 import org.apache.bookkeeper.common.annotation.InterfaceAudience.Public;
 import org.apache.bookkeeper.common.annotation.InterfaceStability.Unstable;
 import org.apache.bookkeeper.conf.ClientConfiguration;
@@ -72,4 +73,22 @@ public interface OpenBuilder extends OpBuilder<ReadHandle> {
      */
     OpenBuilder withDigestType(DigestType digestType);
 
+    /**
+     * Inherit the context attributes of the given slog {@link Logger} on the logger bound to
+     * the resulting {@link ReadHandle}. Every log statement emitted by the handle (and by the open-time machinery
+     * that produces it) will carry the parent logger's context attributes, in addition to the {@code ledgerId}
+     * attribute that is always added by the client.
+     *
+     * <p>Useful for correlating bookkeeper-client log output with the application's own request / tenant / trace
+     * identifiers — typically the application has built a per-request logger via
+     * {@code Logger.get(...).with().attr(...)...build()} and passes it here.
+     *
+     * @param parentLogger logger whose context attributes to inherit; {@code null} is treated as no extra context
+     *
+     * @return the builder itself
+     */
+    default OpenBuilder withLoggerContext(Logger parentLogger) {
+        return this;
+    }
+
 }

bookkeeper-server/src/main/java/org/apache/bookkeeper/client/impl/OpenBuilderBase.java

@@ -18,6 +18,7 @@
 
 package org.apache.bookkeeper.client.impl;
 
+import io.github.merlimat.slog.Logger;
 import java.util.Arrays;
 import lombok.CustomLog;
 import org.apache.bookkeeper.client.LedgerHandle;
@@ -35,6 +36,7 @@ public abstract class OpenBuilderBase implements OpenBuilder {
     protected long ledgerId = LedgerHandle.INVALID_LEDGER_ID;
     protected byte[] password;
     protected DigestType digestType = DigestType.CRC32;
+    protected Logger parentLogger;
 
     @Override
     public OpenBuilder withLedgerId(long ledgerId) {
@@ -60,6 +62,12 @@ public OpenBuilder withDigestType(DigestType digestType) {
         return this;
     }
 
+    @Override
+    public OpenBuilder withLoggerContext(Logger parentLogger) {
+        this.parentLogger = parentLogger;
+        return this;
+    }
+
     protected int validate() {
         if (ledgerId < 0) {
             log.error().attr("ledgerId", ledgerId).log("invalid ledgerId < 0");