feat(s3): attribute S3 GET and PUT metrics to WAL sources

core/src/main/java/kafka/automq/zerozone/DefaultRouterChannelProvider.java

@@ -24,6 +24,7 @@
 import org.apache.kafka.image.MetadataDelta;
 import org.apache.kafka.image.MetadataImage;
 
+import com.automq.stream.s3.metrics.operations.S3RequestSource;
 import com.automq.stream.s3.network.GlobalNetworkBandwidthLimiters;
 import com.automq.stream.s3.operator.BucketURI;
 import com.automq.stream.s3.operator.ObjectStorage;
@@ -103,6 +104,7 @@ private RouterChannel newChannel(int ownerNodeId, boolean readOnly) {
                 .withBucketId(bucket.bucketId())
                 .withOpenMode(bucketReadOnly ? OpenMode.READ_ONLY : OpenMode.READ_WRITE)
                 .withType(WAL_TYPE)
+                .withSource(S3RequestSource.ROUTER_WAL)
                 .withManualMode(true);
             if (!bucketReadOnly) {
                 builder.withEpoch(nodeEpoch);

core/src/main/scala/kafka/log/stream/s3/wal/DefaultWalFactory.java

@@ -20,6 +20,7 @@
 package kafka.log.stream.s3.wal;
 
 import com.automq.shell.AutoMQApplication;
+import com.automq.stream.s3.metrics.operations.S3RequestSource;
 import com.automq.stream.s3.network.NetworkBandwidthLimiter;
 import com.automq.stream.s3.operator.BucketURI;
 import com.automq.stream.s3.operator.ObjectStorage;
@@ -56,6 +57,7 @@ public WriteAheadLog build(IdURI uri, BuildOptions options) {
         switch (uri.protocol().toUpperCase(Locale.ENGLISH)) {
             case "S3":
                 ObjectWALConfig.Builder configBuilder = ObjectWALConfig.builder().withURI(uri)
+                    .withSource(S3RequestSource.DELTA_WAL)
                     .withClusterId(AutoMQApplication.getClusterId())
                     .withNodeId(nodeId)
                     .withEpoch(options.nodeEpoch())
@@ -68,7 +70,8 @@ public WriteAheadLog build(IdURI uri, BuildOptions options) {
                     .outboundLimiter(networkOutboundLimiter)
                     .build();
 
-                ReservationService reservationService = new ObjectReservationService(AutoMQApplication.getClusterId(), walObjectStorage, walObjectStorage.bucketId());
+                ReservationService reservationService = new ObjectReservationService(AutoMQApplication.getClusterId(),
+                    walObjectStorage, walObjectStorage.bucketId(), S3RequestSource.DELTA_WAL);
                 configBuilder.withReservationService(reservationService);
                 // This factory creates a dedicated ObjectStorage, so the WAL owns its lifecycle.
                 return new ObjectWALService(Time.SYSTEM, walObjectStorage, configBuilder.build(), true);

core/src/test/java/kafka/automq/zerozone/DefaultRouterChannelProviderTest.java

@@ -19,6 +19,7 @@
 
 package kafka.automq.zerozone;
 
+import com.automq.stream.s3.metrics.operations.S3RequestSource;
 import com.automq.stream.s3.operator.BucketURI;
 import com.automq.stream.s3.operator.ObjectStorage;
 import com.automq.stream.s3.wal.OpenMode;
@@ -68,13 +69,15 @@ public void testReadWriteChannelUsesURIOptions() {
         assertEquals(123L, writer.epoch());
         assertEquals("cluster", writer.clusterId());
         assertEquals("rc", writer.type());
+        assertEquals(S3RequestSource.ROUTER_WAL, writer.source());
         assertEquals(OpenMode.READ_WRITE, writer.openMode());
         assertEquals("test-secret", IdURI.parse(writer.uri()).extensionString("secretKey"));
         assertFalse(writer.toString().contains("test-secret"));
         assertEquals(1, writer.version());
 
         ObjectWALConfig reader = configs.get(1);
         assertEquals(OpenMode.READ_ONLY, reader.openMode());
+        assertEquals(S3RequestSource.ROUTER_WAL, reader.source());
         assertEquals(43, reader.nodeId());
         assertEquals(writer.uri(), reader.uri());
         assertEquals(1, reader.version());
@@ -118,6 +121,7 @@ public void testMultipleBuckets() {
             provider.channel();
             provider.readOnlyChannel(43);
             assertEquals(4, wals.constructed().size());
+            configs.forEach(config -> assertEquals(S3RequestSource.ROUTER_WAL, config.source()));
             assertEquals(100, configs.get(0).batchInterval());
             assertEquals(200, configs.get(1).batchInterval());
             assertEquals(0, configs.get(0).bucketId());

core/src/test/java/kafka/log/stream/s3/wal/DefaultWalFactoryTest.java

@@ -19,11 +19,13 @@
 
 package kafka.log.stream.s3.wal;
 
+import com.automq.stream.s3.metrics.operations.S3RequestSource;
 import com.automq.stream.s3.network.NetworkBandwidthLimiter;
 import com.automq.stream.s3.operator.ObjectStorage;
 import com.automq.stream.s3.operator.ObjectStorageFactory;
 import com.automq.stream.s3.wal.OpenMode;
 import com.automq.stream.s3.wal.WalFactory;
+import com.automq.stream.s3.wal.impl.object.ObjectReservationService;
 import com.automq.stream.s3.wal.impl.object.ObjectWALConfig;
 import com.automq.stream.s3.wal.impl.object.ObjectWALService;
 import com.automq.stream.utils.IdURI;
@@ -58,6 +60,8 @@ public void testHistoricalURIRespectsLayoutVersion(OpenMode mode) {
         List<ObjectWALConfig> configs = new ArrayList<>();
         try (MockedConstruction<ObjectStorageFactory.Builder> builders = mockConstruction(ObjectStorageFactory.Builder.class,
             withSettings().defaultAnswer(RETURNS_SELF), (builder, context) -> when(builder.build()).thenReturn(mock(ObjectStorage.class)));
+            MockedConstruction<ObjectReservationService> reservations = mockConstruction(ObjectReservationService.class,
+                (reservation, context) -> assertEquals(S3RequestSource.DELTA_WAL, context.arguments().get(3)));
             MockedConstruction<ObjectWALService> wals = mockConstruction(ObjectWALService.class,
                 (wal, context) -> configs.add((ObjectWALConfig) context.arguments().get(2)))) {
             WalFactory.BuildOptions options = WalFactory.BuildOptions.builder().nodeEpoch(42).openMode(mode).build();
@@ -69,7 +73,9 @@ public void testHistoricalURIRespectsLayoutVersion(OpenMode mode) {
             assertEquals(0, configs.get(2).version());
             assertEquals(3, builders.constructed().size());
             assertEquals(3, wals.constructed().size());
+            assertEquals(3, reservations.constructed().size());
             configs.forEach(config -> assertEquals(mode, config.openMode()));
+            configs.forEach(config -> assertEquals(S3RequestSource.DELTA_WAL, config.source()));
         }
     }
 

s3stream/src/main/java/com/automq/stream/s3/metrics/operations/S3RequestSource.java

@@ -0,0 +1,33 @@
+/*
+ * Copyright 2026, AutoMQ HK Limited.
+ *
+ * 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 com.automq.stream.s3.metrics.operations;
+
+/**
+ * Bounded ownership labels for S3 GET/PUT observations. Values are stable metric labels,
+ * not per-producer context; shared WAL helpers can also perform replay or maintenance I/O.
+ */
+public enum S3RequestSource {
+    /** GET/PUT callers that have not supplied source attribution. */
+    UNCLASSIFIED,
+    ROUTER_WAL,
+    DELTA_WAL,
+    /** One merged GET serving child reads with different source labels. */
+    MIXED
+}

s3stream/src/main/java/com/automq/stream/s3/metrics/stats/OperationLatencyMetrics.java

@@ -22,6 +22,7 @@
 import com.automq.stream.s3.metrics.Metrics;
 import com.automq.stream.s3.metrics.MetricsLevel;
 import com.automq.stream.s3.metrics.operations.S3Operation;
+import com.automq.stream.s3.metrics.operations.S3RequestSource;
 import com.automq.stream.s3.metrics.operations.S3Stage;
 import com.automq.stream.s3.metrics.wrapper.DeltaHistogram;
 
@@ -66,6 +67,15 @@ public static DeltaHistogram operation(MetricsLevel metricsLevel, String operati
             operationAttributes(operationType, operationName, status, sizeLabelName));
     }
 
+    /** Registers a sized S3 request histogram with one bounded initiating source. */
+    public static DeltaHistogram request(MetricsLevel metricsLevel, S3Operation operation, String status,
+        String sizeLabelName, S3RequestSource source) {
+        return OPERATION_LATENCY.histogram(metricsLevel, Attributes.builder()
+            .putAll(operationAttributes(operation.getType().getName(), operation.getName(), status, sizeLabelName))
+            .put("source", source.name())
+            .build());
+    }
+
     public static DeltaHistogram stage(MetricsLevel metricsLevel, S3Stage stage) {
         return OPERATION_LATENCY.histogram(metricsLevel, Attributes.builder()
             .putAll(operationAttributes(stage.getOperation().getType().getName(), stage.getOperation().getName()))

s3stream/src/main/java/com/automq/stream/s3/operator/AbstractObjectStorage.java

@@ -22,6 +22,7 @@
 import com.automq.stream.s3.exceptions.ObjectNotExistException;
 import com.automq.stream.s3.metrics.TimerUtil;
 import com.automq.stream.s3.metrics.operations.S3Operation;
+import com.automq.stream.s3.metrics.operations.S3RequestSource;
 import com.automq.stream.s3.network.NetworkBandwidthLimiter;
 import com.automq.stream.s3.network.ThrottleStrategy;
 import com.automq.stream.s3.objects.ObjectAttributes;
@@ -203,8 +204,9 @@ public Writer writer(WriteOptions options, String objectPath) {
 
     @Override
     public CompletableFuture<ByteBuf> rangeRead(ReadOptions options, String objectPath, long start, long end) {
+        ReadOptions requestOptions = options.copy();
         CompletableFuture<ByteBuf> cf = new CompletableFuture<>();
-        if (!bucketCheck(options.bucket(), cf)) {
+        if (!bucketCheck(requestOptions.bucket(), cf)) {
             return cf;
         }
         if (end != RANGE_READ_TO_END && start > end) {
@@ -234,12 +236,12 @@ public CompletableFuture<ByteBuf> rangeRead(ReadOptions options, String objectPa
             });
         }
 
-        networkInboundBandwidthLimiterFunction.apply(options.throttleStrategy(), acquiredSize).whenComplete((v, ex) -> {
+        networkInboundBandwidthLimiterFunction.apply(requestOptions.throttleStrategy(), acquiredSize).whenComplete((v, ex) -> {
             if (ex != null) {
                 cf.completeExceptionally(ex);
             } else {
                 synchronized (waitingReadTasks) {
-                    waitingReadTasks.add(new AbstractObjectStorage.ReadTask(options, objectPath, start, end, cf));
+                    waitingReadTasks.add(new AbstractObjectStorage.ReadTask(requestOptions, objectPath, start, end, cf));
                 }
             }
         });
@@ -249,25 +251,26 @@ public CompletableFuture<ByteBuf> rangeRead(ReadOptions options, String objectPa
 
     @Override
     public CompletableFuture<WriteResult> write(WriteOptions options, String objectPath, ByteBuf data) {
+        WriteOptions requestOptions = options.copy();
         CompletableFuture<Void> cf = new CompletableFuture<>();
         CompletableFuture<WriteResult> retCf = acquireWritePermit(cf).thenApply(nil -> new WriteResult(bucketURI.bucketId()));
         if (retCf.isDone()) {
             data.release();
             return retCf;
         }
         networkOutboundBandwidthLimiter
-            .consume(options.throttleStrategy(), data.readableBytes())
+            .consume(requestOptions.throttleStrategy(), data.readableBytes())
             .whenCompleteAsync((v, ex) -> {
                 if (ex != null) {
                     cf.completeExceptionally(ex);
                     data.release();
                     return;
                 }
-                if (checkTimeout(options, cf)) {
+                if (checkTimeout(requestOptions, cf)) {
                     data.release();
                     return;
                 }
-                queuedWrite0(options, objectPath, data, cf);
+                queuedWrite0(requestOptions, objectPath, data, cf);
             }, writeLimiterCallbackExecutor);
         return retCf;
     }
@@ -309,10 +312,10 @@ private void copy0(String sourceBucket, String sourcePath, String destinationPat
         });
     }
 
-    private void recordWriteStats(String path, long objectSize, TimerUtil timerUtil) {
+    private void recordWriteStats(WriteOptions options, String path, long objectSize, TimerUtil timerUtil) {
         s3LatencyCalculator.record(objectSize, timerUtil.elapsedAs(TimeUnit.MILLISECONDS));
         ObjectStorageMetrics.recordUploadSize(objectSize);
-        ObjectStorageMetrics.recordPutObject(objectSize, true, timerUtil.elapsedAs(TimeUnit.NANOSECONDS));
+        ObjectStorageMetrics.recordPutObject(options.source(), objectSize, true, timerUtil.elapsedAs(TimeUnit.NANOSECONDS));
         successWriteMonitor.record(objectSize);
         if (logger.isDebugEnabled()) {
             logger.debug("put object {} with size {}, cost {}ms", path, objectSize, timerUtil.elapsedAs(TimeUnit.NANOSECONDS));
@@ -355,7 +358,7 @@ private void write0(WriteOptions options, String path, ByteBuf data, Completable
         }
 
         writeCf.thenAccept(nil -> {
-            recordWriteStats(path, objectSize, timerUtil);
+            recordWriteStats(options, path, objectSize, timerUtil);
             data.release();
             finalCf.complete(null);
         }).exceptionally(ex -> {
@@ -368,7 +371,7 @@ private void write0(WriteOptions options, String path, ByteBuf data, Completable
                 return null;
             }
 
-            ObjectStorageMetrics.recordPutObject(objectSize, false, timerUtil.elapsedAs(TimeUnit.NANOSECONDS));
+            ObjectStorageMetrics.recordPutObject(options.source(), objectSize, false, timerUtil.elapsedAs(TimeUnit.NANOSECONDS));
             Pair<RetryStrategy, Throwable> strategyAndCause = toRetryStrategyAndCause(ex, S3Operation.PUT_OBJECT);
             RetryStrategy retryStrategy = strategyAndCause.getLeft();
             Throwable cause = strategyAndCause.getRight();
@@ -407,9 +410,9 @@ private CompletableFuture<Void> doFastRetryWrite(WriteOptions options, String pa
         }
         return cf.whenComplete((nil, ex) -> {
             if (ex == null) {
-                recordWriteStats(path, objectSize, timerUtil);
+                recordWriteStats(options, path, objectSize, timerUtil);
             } else if (!finalCf.isDone()) {
-                ObjectStorageMetrics.recordPutObject(objectSize, false, timerUtil.elapsedAs(TimeUnit.NANOSECONDS));
+                ObjectStorageMetrics.recordPutObject(options.source(), objectSize, false, timerUtil.elapsedAs(TimeUnit.NANOSECONDS));
             }
         });
     }
@@ -822,7 +825,7 @@ private void tryMergeRead0() {
                         path, mergedReadTask.start, mergedReadTask.end,
                         mergedReadTask.end - mergedReadTask.start, mergedReadTask.dataSparsityRate);
                 }
-                mergedRangeRead(mergedReadTask.readTasks.get(0).options, path, mergedReadTask.start, mergedReadTask.end)
+                mergedRangeRead(mergedReadTask.options(), path, mergedReadTask.start, mergedReadTask.end)
                     .whenComplete((rst, ex) -> FutureUtil.suppress(() -> mergedReadTask.handleReadCompleted(rst, ex), logger));
             }
         );
@@ -853,15 +856,15 @@ CompletableFuture<ByteBuf> mergedRangeRead(ReadOptions options, String path, lon
         return retCf;
     }
 
-    private void recordReadStats(String path, long start, long end, ByteBuf buf, TimerUtil timerUtil) {
+    private void recordReadStats(ReadOptions options, String path, long start, long end, ByteBuf buf, TimerUtil timerUtil) {
         long dataSize = buf.readableBytes();
         readLatencyCalculator.record(dataSize, timerUtil.elapsedAs(TimeUnit.MILLISECONDS));
         if (logger.isDebugEnabled()) {
             logger.debug("GetObject for object {} [{}, {}), size: {}, cost: {} ms",
                 path, start, end, dataSize, timerUtil.elapsedAs(TimeUnit.MILLISECONDS));
         }
         ObjectStorageMetrics.recordDownloadSize(dataSize);
-        ObjectStorageMetrics.recordGetObject(dataSize, true, timerUtil.elapsedAs(TimeUnit.NANOSECONDS));
+        ObjectStorageMetrics.recordGetObject(options.source(), dataSize, true, timerUtil.elapsedAs(TimeUnit.NANOSECONDS));
     }
 
     private CompletableFuture<ByteBuf> doFastRetryRead(ReadOptions options, String path, long start, long end,
@@ -876,13 +879,13 @@ private CompletableFuture<ByteBuf> doFastRetryRead(ReadOptions options, String p
         return cf.whenComplete((buf, ex) -> {
             if (ex == null) {
                 try {
-                    recordReadStats(path, start, end, buf, timerUtil);
+                    recordReadStats(options, path, start, end, buf, timerUtil);
                 } catch (Throwable e) {
                     buf.release();
                     throw e;
                 }
             } else if (!finalCf.isDone()) {
-                ObjectStorageMetrics.recordGetObject(end - start, false, timerUtil.elapsedAs(TimeUnit.NANOSECONDS));
+                ObjectStorageMetrics.recordGetObject(options.source(), end - start, false, timerUtil.elapsedAs(TimeUnit.NANOSECONDS));
             }
         });
     }
@@ -898,7 +901,7 @@ private CompletableFuture<ByteBuf> mergedRangeRead0(ReadOptions options, String
         readCf.thenAccept(buf -> {
             boolean transferred = false;
             try {
-                recordReadStats(path, start, end, buf, timerUtil);
+                recordReadStats(options, path, start, end, buf, timerUtil);
                 transferred = cf.complete(buf);
             } finally {
                 if (!transferred) {
@@ -922,7 +925,7 @@ private CompletableFuture<ByteBuf> mergedRangeRead0(ReadOptions options, String
                 logger.warn("GetObject for object {} [{}, {}) fail, retry in {}ms", path, start, end, delay, cause);
                 scheduler.schedule(() -> mergedRangeRead0(options, path, start, end, cf), delay, TimeUnit.MILLISECONDS);
             }
-            ObjectStorageMetrics.recordGetObject(size, false, timerUtil.elapsedAs(TimeUnit.NANOSECONDS));
+            ObjectStorageMetrics.recordGetObject(options.source(), size, false, timerUtil.elapsedAs(TimeUnit.NANOSECONDS));
             return null;
         });
         return readCf;
@@ -1102,6 +1105,18 @@ static class MergedReadTask {
             this.maxMergeReadSparsityRate = maxMergeReadSparsityRate;
         }
 
+        // Preserve the existing first-task settings, but do not assign heterogeneous work to its first caller.
+        ReadOptions options() {
+            ReadOptions options = readTasks.get(0).options.copy();
+            for (ReadTask task : readTasks) {
+                if (task.options.source() != options.source()) {
+                    options.source(S3RequestSource.MIXED);
+                    break;
+                }
+            }
+            return options;
+        }
+
         boolean tryMerge(AbstractObjectStorage.ReadTask readTask) {
             if (!canMerge(readTask)) {
                 return false;