Fix cacheMissPercentage metric

### What changes are proposed in this pull request? Please outline the changes and how this PR fixes the issue. modify the calculation method for cacheMissPercentage metric, ensuring it comprehensively accounts for the influence of job worker read operations. Alluxio#16945 ### Why are the changes needed? Because of the incorrect calculation results of this metric, it may produce exception value. ### Does this PR introduce any user facing changes? No pr-link: Alluxio#18208 change-id: cid-d24a6f324f7148cab4a30fbbf819510a1a314ebc
maobaolong · Dec 21, 2023 · f07e66f · f07e66f
1 parent 59b1ce6
commit f07e66f
Show file tree

Hide file tree

Showing 12 changed files with 188 additions and 13 deletions.
diff --git a/core/common/src/main/java/alluxio/metrics/MetricKey.java b/core/common/src/main/java/alluxio/metrics/MetricKey.java
@@ -1674,6 +1674,13 @@ public static String getSyncMetricName(long mountId) {
           .setDescription("Bytes read per minute throughput from all Alluxio UFSes by all workers")
           .setMetricType(MetricType.GAUGE)
           .build();
+
+  public static final MetricKey CLUSTER_BYTES_READ_CACHE =
+          new Builder("Cluster.BytesReadCache")
+                  .setDescription("Total number of bytes read from all worker's cache")
+                  .setMetricType(MetricType.COUNTER)
+                  .build();
+
   public static final MetricKey CLUSTER_BYTES_WRITTEN_REMOTE =
       new Builder("Cluster.BytesWrittenRemote")
           .setDescription("Total number of bytes written to workers via network (RPC). "
@@ -1974,6 +1981,14 @@ public static String getSyncMetricName(long mountId) {
           .setMetricType(MetricType.METER)
           .setIsClusterAggregated(false)
           .build();
+
+  public static final MetricKey WORKER_BYTES_READ_CACHE =
+          new Builder("Worker.BytesReadCache")
+                  .setDescription("Total number of bytes read from the worker's cache")
+                  .setMetricType(MetricType.COUNTER)
+                  .setIsClusterAggregated(true)
+                  .build();
+
   public static final MetricKey WORKER_BYTES_WRITTEN_DIRECT =
       new Builder("Worker.BytesWrittenDirect")
           .setDescription("Total number of bytes written to this worker "

diff --git a/core/server/master/src/main/java/alluxio/master/meta/AlluxioMasterRestServiceHandler.java b/core/server/master/src/main/java/alluxio/master/meta/AlluxioMasterRestServiceHandler.java
@@ -868,8 +868,11 @@ public Response getWebUIMetrics() {
           .setTotalBytesReadRemote(FormatUtils.getSizeFromBytes(bytesReadRemote))
           .setTotalBytesReadUfs(FormatUtils.getSizeFromBytes(bytesReadUfs));
 
+      Long bytesReadCache = counters.get(
+              MetricKey.CLUSTER_BYTES_READ_CACHE.getName()).getCount();
+
       // cluster cache hit and miss
-      long bytesReadTotal = bytesReadLocal + bytesReadRemote + bytesReadDomainSocket;
+      long bytesReadTotal = bytesReadLocal + bytesReadCache + bytesReadUfs;
       double cacheHitLocalPercentage =
           (bytesReadTotal > 0)
               ? (100D * (bytesReadLocal + bytesReadDomainSocket) / bytesReadTotal) : 0;

diff --git a/core/server/master/src/main/java/alluxio/master/metrics/MetricsStore.java b/core/server/master/src/main/java/alluxio/master/metrics/MetricsStore.java
@@ -233,6 +233,9 @@ public void initMetricKeys() {
       mClusterCounters.putIfAbsent(new ClusterCounterKey(InstanceType.WORKER,
           MetricKey.WORKER_BYTES_READ_DOMAIN.getMetricName()),
           MetricsSystem.counter(MetricKey.CLUSTER_BYTES_READ_DOMAIN.getName()));
+      mClusterCounters.putIfAbsent(new ClusterCounterKey(InstanceType.WORKER,
+                      MetricKey.WORKER_BYTES_READ_CACHE.getMetricName()),
+              MetricsSystem.counter(MetricKey.CLUSTER_BYTES_READ_CACHE.getName()));
       mClusterCounters.putIfAbsent(new ClusterCounterKey(InstanceType.WORKER,
           MetricKey.WORKER_BYTES_WRITTEN_REMOTE.getMetricName()),
           MetricsSystem.counter(MetricKey.CLUSTER_BYTES_WRITTEN_REMOTE.getName()));

diff --git a/core/server/worker/src/main/java/alluxio/worker/block/BlockMasterSyncHelper.java b/core/server/worker/src/main/java/alluxio/worker/block/BlockMasterSyncHelper.java
@@ -203,7 +203,6 @@ boolean heartbeat(
     // Send the heartbeat and execute the response
     Command cmdFromMaster = null;
     List<alluxio.grpc.Metric> metrics = MetricsSystem.reportWorkerMetrics();
-
     try {
       cmdFromMaster = mMasterClient.heartbeat(workerId, storeMeta.getCapacityBytesOnTiers(),
           storeMeta.getUsedBytesOnTiers(), blockReport.getRemovedBlocks(),

diff --git a/core/server/worker/src/main/java/alluxio/worker/block/TieredBlockStore.java b/core/server/worker/src/main/java/alluxio/worker/block/TieredBlockStore.java
@@ -31,6 +31,8 @@
 import alluxio.worker.block.io.BlockReader;
 import alluxio.worker.block.io.BlockWriter;
 import alluxio.worker.block.io.DelegatingBlockReader;
+import alluxio.worker.block.io.LocalFileBlockReader;
+import alluxio.worker.block.io.MetricAccountingBlockReader;
 import alluxio.worker.block.io.FakeBlockReader;
 import alluxio.worker.block.io.FakeBlockWriter;
 import alluxio.worker.block.io.StoreBlockReader;
@@ -242,10 +244,11 @@ public BlockReader createBlockReader(long sessionId, long blockId, long offset)
     }
 
     try {
-      BlockReader reader = new StoreBlockReader(sessionId, block);
+      LocalFileBlockReader reader = new StoreBlockReader(sessionId, block);
       ((FileChannel) reader.getChannel()).position(offset);
       accessBlock(sessionId, blockId);
-      return new DelegatingBlockReader(reader, blockLock);
+      BlockReader mareader = new MetricAccountingBlockReader(reader);
+      return new DelegatingBlockReader(mareader, blockLock);
     } catch (Exception e) {
       blockLock.close();
       throw new IOException(format("Failed to get local block reader, sessionId=%d, "

diff --git a/core/server/worker/src/main/java/alluxio/worker/block/UnderFileSystemBlockStore.java b/core/server/worker/src/main/java/alluxio/worker/block/UnderFileSystemBlockStore.java
@@ -393,11 +393,19 @@ public String toString() {
     }
   }
 
-  private static class BytesReadMetricKey {
-    private final AlluxioURI mUri;
-    private final String mUser;
+  /**
+   * create an BytesReadMetricKey.
+   */
+  public static class BytesReadMetricKey {
+    public final AlluxioURI mUri;
+    public final String mUser;
 
-    BytesReadMetricKey(AlluxioURI uri, String user) {
+    /**
+     * create an instance of the key class.
+     * @param uri
+     * @param user
+     */
+    public BytesReadMetricKey(AlluxioURI uri, String user) {
       mUri = uri;
       mUser = user;
     }

diff --git a/core/server/worker/src/main/java/alluxio/worker/block/io/MetricAccountingBlockReader.java b/core/server/worker/src/main/java/alluxio/worker/block/io/MetricAccountingBlockReader.java
@@ -0,0 +1,103 @@
+/*
+ * The Alluxio Open Foundation licenses this work under the Apache License, version 2.0
+ * (the "License"). You may not use this work except in compliance with the License, which is
+ * available at www.apache.org/licenses/LICENSE-2.0
+ *
+ * This software is distributed on an "AS IS" basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
+ * either express or implied, as more fully set forth in the License.
+ *
+ * See the NOTICE file distributed with this work for information regarding copyright ownership.
+ */
+
+package alluxio.worker.block.io;
+
+import alluxio.metrics.MetricKey;
+import alluxio.metrics.MetricsSystem;
+
+import io.netty.buffer.ByteBuf;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.ReadableByteChannel;
+
+/**
+ * An reader class with metrics.
+ */
+public class MetricAccountingBlockReader extends BlockReader {
+  private final LocalFileBlockReader mBlockReader;
+
+  /**
+   * A decorator of BlockReader.
+   * @param mblockReader block reader
+   */
+  public MetricAccountingBlockReader(LocalFileBlockReader mblockReader) {
+    mBlockReader = mblockReader;
+  }
+
+  @Override
+  public ByteBuffer read(long offset, long length) throws IOException {
+    ByteBuffer buffer = mBlockReader.read(offset, length);
+    int bytesReadFromCache = buffer.limit() - buffer.position();
+    MetricsSystem.counter(MetricKey.WORKER_BYTES_READ_CACHE.getName()).inc(bytesReadFromCache);
+    return buffer;
+  }
+
+  @Override
+  public long getLength() {
+    return mBlockReader.getLength();
+  }
+
+  @Override
+  public ReadableByteChannel getChannel() {
+    return new ReadableByteChannel() {
+      private final ReadableByteChannel mDelegate = mBlockReader.getChannel();
+      @Override
+      public int read(ByteBuffer dst) throws IOException {
+        int bytesRead = mDelegate.read(dst);
+        if (bytesRead != -1) {
+          MetricsSystem.counter(MetricKey.WORKER_BYTES_READ_CACHE.getName()).inc(bytesRead);
+        }
+        return bytesRead;
+      }
+
+      @Override
+      public boolean isOpen() {
+        return mDelegate.isOpen();
+      }
+
+      @Override
+      public void close() throws IOException {
+        mDelegate.close();
+      }
+    };
+  }
+
+  @Override
+  public int transferTo(ByteBuf buf) throws IOException {
+    int bytesReadFromCache = mBlockReader.transferTo(buf);
+    if (bytesReadFromCache != -1) {
+      MetricsSystem.counter(MetricKey.WORKER_BYTES_READ_CACHE.getName()).inc(bytesReadFromCache);
+    }
+    return bytesReadFromCache;
+  }
+
+  @Override
+  public boolean isClosed() {
+    return mBlockReader.isClosed();
+  }
+
+  @Override
+  public String getLocation() {
+    return mBlockReader.getLocation();
+  }
+
+  @Override
+  public String toString() {
+    return mBlockReader.toString();
+  }
+
+  @Override
+  public void close() throws IOException {
+    mBlockReader.close();
+  }
+}
diff --git a/core/server/worker/src/main/java/alluxio/worker/page/PagedBlockReader.java b/core/server/worker/src/main/java/alluxio/worker/page/PagedBlockReader.java
@@ -39,7 +39,6 @@
  */
 @NotThreadSafe
 public class PagedBlockReader extends BlockReader {
-
   private static final ByteBuffer EMPTY_BYTE_BUFFER = ByteBuffer.allocate(0);
   private final long mPageSize;
   private final CacheManager mCacheManager;
@@ -122,6 +121,7 @@ private long read(ByteBuf byteBuf, long offset, long length) throws IOException
         bytesRead += bytesReadFromCache;
         MetricsSystem.meter(MetricKey.CLIENT_CACHE_BYTES_READ_CACHE.getName()).mark(bytesRead);
         mReadFromLocalCache = true;
+        MetricsSystem.counter(MetricKey.WORKER_BYTES_READ_CACHE.getName()).inc(bytesReadFromCache);
       } else {
         if (!mUfsBlockReader.isPresent()) {
           throw new AlluxioRuntimeException(

diff --git a/core/server/worker/src/main/java/alluxio/worker/page/PagedBlockStore.java b/core/server/worker/src/main/java/alluxio/worker/page/PagedBlockStore.java
@@ -353,7 +353,7 @@ private BlockReader getBlockReader(PagedBlockMeta blockMeta, long offset,
     }
     final Optional<PagedUfsBlockReader> ufsBlockReader =
         readOptions.map(opt -> new PagedUfsBlockReader(
-            mUfsManager, mUfsInStreamCache, blockMeta, offset, opt, mPageSize));
+                mUfsManager, mUfsInStreamCache, blockMeta, offset, opt, mPageSize));
     return new PagedBlockReader(mCacheManager, blockMeta, offset, ufsBlockReader, mPageSize);
   }
 

diff --git a/core/server/worker/src/main/java/alluxio/worker/page/PagedUfsBlockReader.java b/core/server/worker/src/main/java/alluxio/worker/page/PagedUfsBlockReader.java
@@ -12,16 +12,24 @@
 package alluxio.worker.page;
 
 import alluxio.conf.PropertyKey;
+import alluxio.exception.runtime.AlluxioRuntimeException;
+import alluxio.exception.status.NotFoundException;
+import alluxio.exception.status.UnavailableException;
+import alluxio.metrics.MetricInfo;
+import alluxio.metrics.MetricKey;
+import alluxio.metrics.MetricsSystem;
 import alluxio.network.protocol.databuffer.NioDirectBufferPool;
 import alluxio.resource.CloseableResource;
 import alluxio.underfs.UfsManager;
 import alluxio.underfs.UnderFileSystem;
 import alluxio.underfs.options.OpenOptions;
 import alluxio.util.IdUtils;
 import alluxio.worker.block.UfsInputStreamCache;
+import alluxio.worker.block.UnderFileSystemBlockStore.BytesReadMetricKey;
 import alluxio.worker.block.io.BlockReader;
 import alluxio.worker.block.meta.BlockMeta;
 
+import com.codahale.metrics.Counter;
 import com.google.common.base.Preconditions;
 import io.netty.buffer.ByteBuf;
 
@@ -31,6 +39,8 @@
 import java.nio.channels.Channels;
 import java.nio.channels.ClosedChannelException;
 import java.nio.channels.ReadableByteChannel;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
 
 /**
  * Block reader that reads from UFS.
@@ -47,6 +57,9 @@ public class PagedUfsBlockReader extends BlockReader {
   private long mLastPageIndex = -1;
   private boolean mClosed = false;
   private long mPosition;
+  private final ConcurrentMap<BytesReadMetricKey, Counter> mUfsBytesReadMetrics =
+          new ConcurrentHashMap<>();
+  private final Counter mUfsBytesRead;
 
   /**
    * @param ufsManager
@@ -70,6 +83,23 @@ public PagedUfsBlockReader(UfsManager ufsManager,
     mInitialOffset = offset;
     mLastPage = ByteBuffer.allocateDirect((int) mPageSize);
     mPosition = offset;
+    try {
+      UfsManager.UfsClient ufsClient = mUfsManager.get(mUfsBlockOptions.getMountId());
+      mUfsBytesRead = mUfsBytesReadMetrics.computeIfAbsent(
+              new BytesReadMetricKey(ufsClient.getUfsMountPointUri(), mUfsBlockOptions.getUser()),
+              key -> key.mUser == null
+                      ? MetricsSystem.counterWithTags(
+                      MetricKey.WORKER_BYTES_READ_UFS.getName(),
+                      MetricKey.WORKER_BYTES_READ_UFS.isClusterAggregated(),
+                      MetricInfo.TAG_UFS, MetricsSystem.escape(key.mUri))
+                      : MetricsSystem.counterWithTags(
+                      MetricKey.WORKER_BYTES_READ_UFS.getName(),
+                      MetricKey.WORKER_BYTES_READ_UFS.isClusterAggregated(),
+                      MetricInfo.TAG_UFS, MetricsSystem.escape(key.mUri),
+                      MetricInfo.TAG_USER, key.mUser));
+    } catch (UnavailableException | NotFoundException e) {
+      throw AlluxioRuntimeException.from(e);
+    }
   }
 
   @Override
@@ -145,6 +175,7 @@ public int readPageAtIndex(ByteBuffer buffer, long pageIndex) throws IOException
     mLastPage.flip();
     mLastPageIndex = pageIndex;
     fillWithCachedPage(buffer, pageIndex * mPageSize, totalBytesRead);
+    mUfsBytesRead.inc(totalBytesRead);
     return totalBytesRead;
   }
 

diff --git a/core/server/worker/src/main/java/alluxio/worker/page/UfsBlockReadOptions.java b/core/server/worker/src/main/java/alluxio/worker/page/UfsBlockReadOptions.java
@@ -16,6 +16,7 @@
 import com.google.common.base.Preconditions;
 
 import java.util.Objects;
+import javax.annotation.Nullable;
 
 /**
  * Options for reading a block from UFS.
@@ -26,12 +27,15 @@ public final class UfsBlockReadOptions {
   private final long mOffsetInFile;
   private final String mUfsPath;
   private final boolean mCacheIntoAlluxio;
+  @Nullable private final String mUser;
 
-  UfsBlockReadOptions(long mountId, long offsetInFile, String ufsPath, boolean cacheIntoAlluxio) {
+  UfsBlockReadOptions(long mountId, long offsetInFile, String ufsPath, boolean cacheIntoAlluxio,
+                      @Nullable String user) {
     mMountId = mountId;
     mOffsetInFile = offsetInFile;
     mUfsPath = ufsPath;
     mCacheIntoAlluxio = cacheIntoAlluxio;
+    mUser = user;
   }
 
   /**
@@ -47,7 +51,7 @@ public static UfsBlockReadOptions fromProto(Protocol.OpenUfsBlockOptions options
         "missing offset in file for UFS block read");
     Preconditions.checkArgument(options.hasUfsPath(), "missing UFS path for UFS block read");
     return new UfsBlockReadOptions(options.getMountId(),
-        options.getOffsetInFile(), options.getUfsPath(), !options.getNoCache());
+        options.getOffsetInFile(), options.getUfsPath(), !options.getNoCache(), options.getUser());
   }
 
   /**
@@ -71,6 +75,12 @@ public String getUfsPath() {
     return mUfsPath;
   }
 
+  /**
+   *
+   * @return user
+   */
+  public String getUser() { return mUser; }
+
   /**
    * @return whether the UFS block should be cached into Alluxio
    */

diff --git a/core/server/worker/src/test/java/alluxio/worker/page/PagedBlockReaderTest.java b/core/server/worker/src/test/java/alluxio/worker/page/PagedBlockReaderTest.java
@@ -239,7 +239,7 @@ public void sequentialTransferMultipleTimes() throws Exception {
   }
 
   private static UfsBlockReadOptions createUfsBlockOptions(String ufsPath) {
-    return new UfsBlockReadOptions(MOUNT_ID, OFFSET_IN_FILE, ufsPath, true);
+    return new UfsBlockReadOptions(MOUNT_ID, OFFSET_IN_FILE, ufsPath, true, null);
   }
 
   private static void createTempUfsBlock(Path destPath, long blockSize) throws Exception {