[client] RemoteLogDownloader should increment remoteFetchBytesPerSecond of ScannerMetricGroup (apache#858)

Author: SteNicholas

fluss-client/src/main/java/com/alibaba/fluss/client/metrics/ScannerMetricGroup.java

@@ -47,7 +47,7 @@ public class ScannerMetricGroup extends AbstractMetricGroup {
     private final Histogram bytesPerRequest;
 
     // remote log
-    private final Counter remoteFetchBytesCount;
+    private final Counter remoteFetchBytes;
     private final Counter remoteFetchRequestCount;
     private final Counter remoteFetchErrorCount;
 
@@ -64,8 +64,8 @@ public ScannerMetricGroup(ClientMetricGroup parent, TablePath tablePath) {
         fetchRequestCount = new ThreadSafeSimpleCounter();
         meter(MetricNames.SCANNER_FETCH_RATE, new MeterView(fetchRequestCount));
 
-        remoteFetchBytesCount = new ThreadSafeSimpleCounter();
-        meter(MetricNames.SCANNER_REMOTE_FETCH_BYTES_RATE, new MeterView(remoteFetchBytesCount));
+        remoteFetchBytes = new ThreadSafeSimpleCounter();
+        meter(MetricNames.SCANNER_REMOTE_FETCH_BYTES_RATE, new MeterView(remoteFetchBytes));
         remoteFetchRequestCount = new ThreadSafeSimpleCounter();
         meter(MetricNames.SCANNER_REMOTE_FETCH_RATE, new MeterView(remoteFetchRequestCount));
         remoteFetchErrorCount = new ThreadSafeSimpleCounter();
@@ -90,8 +90,8 @@ public Histogram bytesPerRequest() {
         return bytesPerRequest;
     }
 
-    public Counter remoteFetchBytesCount() {
-        return remoteFetchBytesCount;
+    public Counter remoteFetchBytes() {
+        return remoteFetchBytes;
     }
 
     public Counter remoteFetchRequestCount() {

fluss-client/src/main/java/com/alibaba/fluss/client/table/scanner/log/RemoteLogDownloader.java

@@ -163,6 +163,7 @@ void fetchOnce() throws Exception {
                     segmentPath,
                     new CloseableRegistry());
             File localFile = new File(segmentPath.toFile(), fsPathAndFileName.getFileName());
+            scannerMetricGroup.remoteFetchBytes().inc(localFile.length());
             String segmentId = request.segment.remoteLogSegmentId().toString();
             fetchedFiles.put(segmentId, segmentPath);
             request.future.complete(localFile);
@@ -229,6 +230,18 @@ Path getLocalLogDir() {
         return localLogDir;
     }
 
+    protected static FsPathAndFileName getFsPathAndFileName(
+            FsPath remoteLogTabletDir, RemoteLogSegment segment) {
+        FsPath remotePath =
+                remoteLogSegmentFile(
+                        remoteLogSegmentDir(remoteLogTabletDir, segment.remoteLogSegmentId()),
+                        segment.remoteLogStartOffset());
+        return new FsPathAndFileName(
+                remotePath,
+                FlussPaths.filenamePrefixFromOffset(segment.remoteLogStartOffset())
+                        + LOG_FILE_SUFFIX);
+    }
+
     /**
      * Thread to download remote log files to local. The thread will keep fetching remote log files
      * until it is interrupted.
@@ -257,14 +270,7 @@ public RemoteLogDownloadRequest(RemoteLogSegment segment, FsPath remoteLogTablet
         }
 
         public FsPathAndFileName getFsPathAndFileName() {
-            FsPath remotePath =
-                    remoteLogSegmentFile(
-                            remoteLogSegmentDir(remoteLogTabletDir, segment.remoteLogSegmentId()),
-                            segment.remoteLogStartOffset());
-            return new FsPathAndFileName(
-                    remotePath,
-                    FlussPaths.filenamePrefixFromOffset(segment.remoteLogStartOffset())
-                            + LOG_FILE_SUFFIX);
+            return RemoteLogDownloader.getFsPathAndFileName(remoteLogTabletDir, segment);
         }
     }
 }

fluss-client/src/test/java/com/alibaba/fluss/client/table/scanner/log/RemoteLogDownloaderTest.java

@@ -16,6 +16,7 @@
 
 package com.alibaba.fluss.client.table.scanner.log;
 
+import com.alibaba.fluss.client.metrics.ScannerMetricGroup;
 import com.alibaba.fluss.client.metrics.TestingScannerMetricGroup;
 import com.alibaba.fluss.client.table.scanner.RemoteFileDownloader;
 import com.alibaba.fluss.config.ConfigOptions;
@@ -56,6 +57,7 @@ class RemoteLogDownloaderTest {
     private FsPath remoteLogDir;
     private Configuration conf;
     private RemoteFileDownloader remoteFileDownloader;
+    private ScannerMetricGroup scannerMetricGroup;
     private RemoteLogDownloader remoteLogDownloader;
 
     @BeforeEach
@@ -66,12 +68,13 @@ void beforeEach() {
         conf.set(ConfigOptions.CLIENT_SCANNER_REMOTE_LOG_PREFETCH_NUM, 4);
         remoteLogDir = remoteLogDir(conf);
         remoteFileDownloader = new RemoteFileDownloader(1);
+        scannerMetricGroup = TestingScannerMetricGroup.newInstance();
         remoteLogDownloader =
                 new RemoteLogDownloader(
                         DATA1_TABLE_PATH,
                         conf,
                         remoteFileDownloader,
-                        TestingScannerMetricGroup.newInstance(),
+                        scannerMetricGroup,
                         // use a short timout for faster testing
                         10L);
     }
@@ -106,12 +109,18 @@ void testPrefetchNum() throws Exception {
                 });
 
         assertThat(FileUtils.listDirectory(localLogDir).length).isEqualTo(4);
+        assertThat(scannerMetricGroup.remoteFetchRequestCount().getCount()).isEqualTo(4);
+        assertThat(scannerMetricGroup.remoteFetchBytes().getCount())
+                .isEqualTo(remoteLogSegmentFilesLength(remoteLogSegments, remoteLogTabletDir, 4));
         assertThat(remoteLogDownloader.getPrefetchSemaphore().availablePermits()).isEqualTo(0);
 
         futures.get(0).getRecycleCallback().run();
         // the 5th segment should success.
         retry(Duration.ofMinutes(1), () -> assertThat(futures.get(4).isDone()).isTrue());
         assertThat(FileUtils.listDirectory(localLogDir).length).isEqualTo(4);
+        assertThat(scannerMetricGroup.remoteFetchRequestCount().getCount()).isEqualTo(5);
+        assertThat(scannerMetricGroup.remoteFetchBytes().getCount())
+                .isEqualTo(remoteLogSegmentFilesLength(remoteLogSegments, remoteLogTabletDir, 5));
         assertThat(remoteLogDownloader.getPrefetchSemaphore().availablePermits()).isEqualTo(0);
 
         futures.get(1).getRecycleCallback().run();
@@ -163,4 +172,19 @@ private static List<RemoteLogSegment> buildRemoteLogSegmentList(
         }
         return remoteLogSegmentList;
     }
+
+    private static Long remoteLogSegmentFilesLength(
+            List<RemoteLogSegment> remoteLogSegments, FsPath remoteLogTabletDir, int segmentNum) {
+        return remoteLogSegments.stream()
+                .limit(segmentNum)
+                .mapToLong(
+                        segment ->
+                                new File(
+                                                RemoteLogDownloader.getFsPathAndFileName(
+                                                                remoteLogTabletDir, segment)
+                                                        .getPath()
+                                                        .getPath())
+                                        .length())
+                .sum();
+    }
 }