WIP

linkedin · Feb 8, 2025 · 544e5e8 · 544e5e8
1 parent 2696aa9
commit 544e5e8
Show file tree

Hide file tree

Showing 9 changed files with 122 additions and 62 deletions.
diff --git a/ambry-api/src/main/java/com/github/ambry/store/FileInfo.java b/ambry-api/src/main/java/com/github/ambry/store/FileInfo.java
@@ -18,7 +18,9 @@ public class FileInfo {
   private final String fileName;
   private final long fileSize;
 
-  public FileInfo(String fileName, Long fileSize) {
+  public FileInfo(
+      String fileName,
+      Long fileSize) {
     this.fileName = fileName;
     this.fileSize = fileSize;
   }
@@ -30,4 +32,12 @@ public String getFileName() {
   public Long getFileSize() {
     return fileSize;
   }
-}
+
+  @Override
+  public String toString() {
+    return "FileInfo{" +
+        "fileName='" + fileName + '\'' +
+        ", fileSize=" + fileSize +
+        '}';
+  }
+}
diff --git a/ambry-api/src/main/java/com/github/ambry/store/LogInfo.java b/ambry-api/src/main/java/com/github/ambry/store/LogInfo.java
@@ -22,7 +22,10 @@ public class LogInfo {
   private final List<FileInfo> indexSegments;
   private final List<FileInfo> bloomFilters;
 
-  public LogInfo(FileInfo logSegment, List<FileInfo> indexSegments, List<FileInfo> bloomFilters) {
+  public LogInfo(
+      FileInfo logSegment,
+      List<FileInfo> indexSegments,
+      List<FileInfo> bloomFilters) {
     this.logSegment = logSegment;
     this.indexSegments = indexSegments;
     this.bloomFilters = bloomFilters;
@@ -42,4 +45,13 @@ public List<FileInfo> getIndexSegments() {
   public List<FileInfo> getBloomFilters() {
     return Collections.unmodifiableList(bloomFilters);
   }
+
+  @Override
+  public String toString() {
+    return "LogInfo{" +
+        "logSegment=" + logSegment +
+        ", indexSegments=" + indexSegments +
+        ", bloomFilters=" + bloomFilters +
+        '}';
+  }
 }
diff --git a/ambry-commons/src/main/java/com/github/ambry/commons/ServerMetrics.java b/ambry-commons/src/main/java/com/github/ambry/commons/ServerMetrics.java
@@ -101,6 +101,15 @@ public class ServerMetrics {
   public final Histogram deleteBlobProcessingTimeInMs;
   public final Histogram deleteBlobResponseQueueTimeInMs;
 
+  public final Histogram fileCopyGetMetadataRequestQueueTimeInMs;
+  public final Histogram fileCopyGetMetadataProcessingTimeInMs;
+  public final Histogram fileCopyGetMetadataResponseQueueTimeInMs;
+  public final Histogram fileCopyGetMetadataSendTimeInMs;
+  public final Histogram fileCopyGetMetadataTotalTimeInMs;
+  public final Meter fileCopyGetMetadataRequestRate;
+  public final Meter fileCopyGetMetadataDroppedRate;
+
+
   public final Histogram batchDeleteBlobRequestQueueTimeInMs;
   public final Histogram batchDeleteBlobProcessingTimeInMs;
   public final Histogram batchDeleteBlobResponseQueueTimeInMs;
@@ -419,6 +428,22 @@ public ServerMetrics(MetricRegistry registry, Class<?> requestClass, Class<?> se
     deleteBlobProcessingTimeInMs = registry.histogram(MetricRegistry.name(requestClass, "DeleteBlobProcessingTime"));
     deleteBlobResponseQueueTimeInMs =
         registry.histogram(MetricRegistry.name(requestClass, "DeleteBlobResponseQueueTime"));
+
+    fileCopyGetMetadataRequestQueueTimeInMs =
+        registry.histogram(MetricRegistry.name(requestClass, "FileCopyGetMetadataRequestQueueTimeInMs"));
+    fileCopyGetMetadataProcessingTimeInMs =
+        registry.histogram(MetricRegistry.name(requestClass, "FileCopyGetMetadataProcessingTimeInMs"));
+    fileCopyGetMetadataResponseQueueTimeInMs =
+        registry.histogram(MetricRegistry.name(requestClass, "FileCopyGetMetadataResponseQueueTimeInMs"));
+    fileCopyGetMetadataSendTimeInMs =
+        registry.histogram(MetricRegistry.name(requestClass, "FileCopyGetMetadataSendTimeInMs"));
+    fileCopyGetMetadataTotalTimeInMs =
+        registry.histogram(MetricRegistry.name(requestClass, "FileCopyGetMetadataTotalTimeInMs"));
+    fileCopyGetMetadataRequestRate =
+        registry.meter(MetricRegistry.name(requestClass, "FileCopyGetMetadataRequestRate"));
+    fileCopyGetMetadataDroppedRate =
+        registry.meter(MetricRegistry.name(requestClass, "FileCopyGetMetadataDroppedRate"));
+
     batchDeleteBlobRequestQueueTimeInMs =
         registry.histogram(MetricRegistry.name(requestClass, "BatchDeleteBlobRequestQueueTimeInMs"));
     batchDeleteBlobProcessingTimeInMs = registry.histogram(MetricRegistry.name(requestClass, "BatchDeleteBlobProcessingTimeInMs"));

diff --git a/ambry-protocol/src/main/java/com/github/ambry/protocol/FileCopyGetMetaDataRequest.java b/ambry-protocol/src/main/java/com/github/ambry/protocol/FileCopyGetMetaDataRequest.java
@@ -22,7 +22,7 @@
 import javax.annotation.Nonnull;
 
 
-public class FileCopyGetMetaDataRequest extends RequestOrResponse{
+public class FileCopyGetMetaDataRequest extends RequestOrResponse {
   private final PartitionId partitionId;
   private final String hostName;
   public static final short File_Metadata_Request_Version_V1 = 1;
@@ -56,10 +56,12 @@ public static FileCopyGetMetaDataRequest readFrom(
       @Nonnull ClusterMap clusterMap) throws IOException {
     short versionId = stream.readShort();
     validateVersion(versionId);
+
     int correlationId = stream.readInt();
     String clientId = Utils.readIntString(stream);
     String hostName = Utils.readIntString(stream);
     PartitionId partitionId = clusterMap.getPartitionIdFromStream(stream);
+
     return new FileCopyGetMetaDataRequest(versionId, correlationId, clientId, partitionId, hostName);
   }
 
@@ -73,10 +75,17 @@ public String toString() {
     return sb.toString();
   }
 
+  @Override
+  public void accept(RequestVisitor visitor) {
+    visitor.visit(this);
+  }
+
+  @Override
   public long sizeInBytes() {
     return super.sizeInBytes() + HostName_Field_Size_In_Bytes + hostName.length() + partitionId.getBytes().length;
   }
 
+  @Override
   protected void prepareBuffer() {
     super.prepareBuffer();
     Utils.serializeString(bufferToSend, hostName, Charset.defaultCharset());

diff --git a/ambry-protocol/src/main/java/com/github/ambry/protocol/FileCopyGetMetaDataResponse.java b/ambry-protocol/src/main/java/com/github/ambry/protocol/FileCopyGetMetaDataResponse.java
@@ -32,7 +32,7 @@ public class FileCopyGetMetaDataResponse extends Response {
   static short CURRENT_VERSION = File_Copy_Protocol_Metadata_Response_Version_V1;
 
   public FileCopyGetMetaDataResponse(short versionId, int correlationId, String clientId, int numberOfLogfiles,
-      List<LogInfo> logInfos, ServerErrorCode errorCode) {
+      @Nonnull List<LogInfo> logInfos, @Nonnull ServerErrorCode errorCode) {
     super(RequestOrResponseType.FileCopyGetMetaDataResponse, versionId, correlationId, clientId, errorCode);
 
     validateVersion(versionId);
@@ -60,14 +60,15 @@ public static FileCopyGetMetaDataResponse readFrom(
       //Setting the number of logfiles to 0 as there are no logfiles to be read.
       return new FileCopyGetMetaDataResponse(versionId, correlationId, clientId, 0, new ArrayList<>(), errorCode);
     }
-
     int numberOfLogfiles = stream.readInt();
     List<LogInfo> logInfos = new ArrayList<>();
     for (int i = 0; i < numberOfLogfiles; i++) {
       logInfos.add(LogInfo.readFrom(stream));
     }
     return new FileCopyGetMetaDataResponse(versionId, correlationId, clientId, numberOfLogfiles, logInfos, errorCode);
   }
+
+  @Override
   protected void prepareBuffer() {
     super.prepareBuffer();
     bufferToSend.writeInt(numberOfLogfiles);
@@ -76,6 +77,7 @@ protected void prepareBuffer() {
     }
   }
 
+  @Override
   public long sizeInBytes() {
     return super.sizeInBytes() + Integer.BYTES + logInfos.stream().mapToLong(LogInfo::sizeInBytes).sum();
   }

diff --git a/ambry-protocol/src/main/java/com/github/ambry/protocol/FileInfo.java b/ambry-protocol/src/main/java/com/github/ambry/protocol/FileInfo.java
@@ -34,7 +34,7 @@ public class FileInfo {
 
   public FileInfo(
       @Nonnull String fileName,
-      @Nonnull long fileSize) {
+      long fileSize) {
     this.fileName = fileName;
     this.fileSizeInBytes = fileSize;
   }
@@ -56,8 +56,10 @@ public void writeTo(@Nonnull ByteBuf buf) {
 
   public String toString() {
     StringBuilder sb = new StringBuilder();
-    sb.append("FileInfo[").append("FileName=").append(fileName).append(", FileSizeInBytes=").append(fileSizeInBytes)
-        .append("]");
+    sb.append("FileInfo[")
+        .append("FileName=").append(fileName)
+        .append(", FileSizeInBytes=").append(fileSizeInBytes)
+      .append("]");
     return sb.toString();
   }
 

diff --git a/ambry-protocol/src/main/java/com/github/ambry/protocol/RequestVisitor.java b/ambry-protocol/src/main/java/com/github/ambry/protocol/RequestVisitor.java
@@ -48,6 +48,12 @@ public interface RequestVisitor {
    */
   void visit(BatchDeleteRequest deleteRequest);
 
+  /**
+   * Performs any actions related to Batch Delete request.
+   * @param fileCopyGetMetaDataRequest to visit.
+   */
+  void visit(FileCopyGetMetaDataRequest fileCopyGetMetaDataRequest);
+
   /**
    * Performs any actions related to Un-delete request.
    * @param undeleteRequest to visit.

diff --git a/ambry-server/src/main/java/com/github/ambry/server/AmbryRequests.java b/ambry-server/src/main/java/com/github/ambry/server/AmbryRequests.java
@@ -1684,65 +1684,57 @@ public void handleUndeleteRequest(NetworkRequest request) throws IOException, In
    * Handler for FileMetadataRequest
    */
   void handleFileCopyGetMetaDataRequest(NetworkRequest request) throws InterruptedException, IOException {
+    long requestQueueTime = SystemTime.getInstance().milliseconds() - request.getStartTimeInMs();
+    long totalTimeSpent = requestQueueTime;
+    long startTime = SystemTime.getInstance().milliseconds();
+
     FileCopyGetMetaDataRequest fileCopyGetMetaDataRequest =
         FileCopyGetMetaDataRequest.readFrom(new DataInputStream(request.getInputStream()), clusterMap);
-
+    FileCopyGetMetaDataResponse response = null;
     List<LogInfo> logInfos;
     try {
-      List<com.github.ambry.store.LogInfo> logSegments =
-          storeManager.getLogSegmentMetadataFiles(fileCopyGetMetaDataRequest.getPartitionId(), true);
-
+      List<com.github.ambry.store.LogInfo> logSegments = storeManager.getLogSegmentMetadataFiles(
+          fileCopyGetMetaDataRequest.getPartitionId(), true);
       logInfos = convertStoreToProtocolLogInfo(logSegments);
+
+      response = new FileCopyGetMetaDataResponse(
+          FileCopyGetMetaDataResponse.File_Copy_Protocol_Metadata_Response_Version_V1,
+          fileCopyGetMetaDataRequest.getCorrelationId(), fileCopyGetMetaDataRequest.getClientId(),
+          logInfos.size(), logInfos, ServerErrorCode.No_Error);
     } catch (Exception e) {
       logger.error("Error while getting log segment metadata for partition {}",
           fileCopyGetMetaDataRequest.getPartitionId().getId(), e);
-      FileCopyGetMetaDataResponse response = new FileCopyGetMetaDataResponse(
+
+      response = new FileCopyGetMetaDataResponse(
           FileCopyGetMetaDataResponse.File_Copy_Protocol_Metadata_Response_Version_V1,
           fileCopyGetMetaDataRequest.getCorrelationId(), fileCopyGetMetaDataRequest.getClientId(),
-          0, null, ServerErrorCode.Unknown_Error);
-      requestResponseChannel.sendResponse(response, request, null);
-      return;
-    }
-    FileCopyGetMetaDataResponse response = new FileCopyGetMetaDataResponse(
-        FileCopyGetMetaDataResponse.File_Copy_Protocol_Metadata_Response_Version_V1,
-        fileCopyGetMetaDataRequest.getCorrelationId(), fileCopyGetMetaDataRequest.getClientId(),
-        logInfos.size(), logInfos, ServerErrorCode.No_Error);
-
-    // TODO: Add metrics for this operation
-    Histogram dummyHistogram = new Histogram(new Reservoir() {
-      @Override
-      public int size() {
-        return 0;
-      }
-
-      @Override
-      public void update(long value) {
-      }
-
-      @Override
-      public Snapshot getSnapshot() {
-        return null;
-      }
-    });
-    ServerNetworkResponseMetrics serverNetworkResponseMetrics = new ServerNetworkResponseMetrics(dummyHistogram,
-        dummyHistogram, dummyHistogram, null, null, 0);
+          0, new ArrayList<>(), ServerErrorCode.Unknown_Error);
+    } finally {
+      long processingTime = SystemTime.getInstance().milliseconds() - startTime;
+      totalTimeSpent += processingTime;
+      publicAccessLogger.info("{} {} processingTime {}", fileCopyGetMetaDataRequest, response, processingTime);
 
-    logger.info("Dw: Api response, partition-" + fileCopyGetMetaDataRequest.getPartitionId().getId() + " " + response);
-    requestResponseChannel.sendResponse(response, request, serverNetworkResponseMetrics);
+      // Update request metrics.
+      RequestMetricsUpdater metricsUpdater = new RequestMetricsUpdater(
+          requestQueueTime, processingTime, 0, 0, false);
+      fileCopyGetMetaDataRequest.accept(metricsUpdater);
+    }
+    requestResponseChannel.sendResponse(response, request,
+        new ServerNetworkResponseMetrics(metrics.fileCopyGetMetadataResponseQueueTimeInMs,
+            metrics.fileCopyGetMetadataSendTimeInMs, metrics.fileCopyGetMetadataTotalTimeInMs,
+            null, null, totalTimeSpent));
   }
 
   private List<LogInfo> convertStoreToProtocolLogInfo(List<com.github.ambry.store.LogInfo> logSegments) {
     List<LogInfo> logInfos = new ArrayList<>();
     for (com.github.ambry.store.LogInfo logSegment : logSegments) {
       List<com.github.ambry.protocol.FileInfo> indexSegments = new ArrayList<>();
-      logSegment.getIndexSegments().forEach(indexSegment -> {
-        indexSegments.add(new com.github.ambry.protocol.FileInfo(indexSegment.getFileName(), indexSegment.getFileSize()));
-      });
+      logSegment.getIndexSegments().forEach(indexSegment ->
+        indexSegments.add(new com.github.ambry.protocol.FileInfo(indexSegment.getFileName(), indexSegment.getFileSize())));
 
       List<com.github.ambry.protocol.FileInfo> bloomFilters = new ArrayList<>();
-      logSegment.getBloomFilters().forEach(bloomFilter -> {
-        bloomFilters.add(new com.github.ambry.protocol.FileInfo(bloomFilter.getFileName(), bloomFilter.getFileSize()));
-      });
+      logSegment.getBloomFilters().forEach(bloomFilter ->
+        bloomFilters.add(new com.github.ambry.protocol.FileInfo(bloomFilter.getFileName(), bloomFilter.getFileSize())));
 
       logInfos.add(new LogInfo(
         logSegment.getLogSegment().getFileName(), logSegment.getLogSegment().getFileSize(),
@@ -2128,6 +2120,20 @@ public void visit(BatchDeleteRequest deleteRequest) {
 
     }
 
+    @Override
+    public void visit(FileCopyGetMetaDataRequest fileCopyGetMetaDataRequest) {
+      metrics.fileCopyGetMetadataRequestQueueTimeInMs.update(requestQueueTime);
+      metrics.fileCopyGetMetadataRequestRate.mark();
+      metrics.fileCopyGetMetadataProcessingTimeInMs.update(requestProcessingTime);
+      responseQueueTimeHistogram = metrics.fileCopyGetMetadataResponseQueueTimeInMs;
+      responseSendTimeHistogram = metrics.fileCopyGetMetadataSendTimeInMs;
+      requestTotalTimeHistogram = metrics.fileCopyGetMetadataTotalTimeInMs;
+      if (isRequestDropped) {
+        metrics.fileCopyGetMetadataDroppedRate.mark();
+        metrics.totalRequestDroppedRate.mark();
+      }
+    }
+
     @Override
     public void visit(UndeleteRequest undeleteRequest) {
       metrics.undeleteBlobRequestQueueTimeInMs.update(requestQueueTime);

diff --git a/ambry-store/src/main/java/com/github/ambry/store/BlobStore.java b/ambry-store/src/main/java/com/github/ambry/store/BlobStore.java
@@ -1330,22 +1330,10 @@ List<LogInfo> getLogSegmentMetadataFiles(boolean includeActiveLogSegment) {
     List<FileInfo> sealedLogsAndMetaDataFiles = getLogSegments(includeActiveLogSegment);
     if (null != sealedLogsAndMetaDataFiles) {
       for (FileInfo E : sealedLogsAndMetaDataFiles) {
-        logger.info("[Dw] LS file: {} size: {}", E.getFileName(), E.getFileSize());
-
         LogSegmentName logSegmentName = LogSegmentName.fromFilename(E.getFileName() + LogSegmentName.SUFFIX);
-
         List<FileInfo> allIndexSegmentsForLogSegment = getAllIndexSegmentsForLogSegment(dataDir, logSegmentName);
-        if (null != allIndexSegmentsForLogSegment) {
-          for (FileInfo is : allIndexSegmentsForLogSegment) {
-            logger.info("[Dw] IS file: {} size: {}", is.getFileName(), is.getFileSize());
-          }
-        }
         List<FileInfo> bloomFiltersForLogSegment = getAllBloomFiltersForLogSegment(dataDir, logSegmentName);
-        if (null != bloomFiltersForLogSegment) {
-          for (FileInfo bf : bloomFiltersForLogSegment) {
-            logger.info("[Dw] BF file: {} size: {}", bf.getFileName(), bf.getFileSize());
-          }
-        }
+
         result.add(new LogInfo(E, allIndexSegmentsForLogSegment, bloomFiltersForLogSegment));
       }
     }