Count invalid server-generated upload IDs

ambry-api/src/main/java/com/github/ambry/rest/RestUtils.java

@@ -514,6 +514,11 @@ public static final class InternalKeys {
      */
     public static final String REQUEST_PATH = KEY_PREFIX + "request-path";
 
+    /**
+     * Boolean context for consuming a server-generated upload ID. Cleared after successful conversion or TTL promotion.
+     */
+    public static final String BLOB_ID_IS_SERVER_GENERATED = KEY_PREFIX + "blob-id-is-server-generated";
+
     /**
      * The internal header to determine if the delete request is coming from a dataset deletion.
      */

ambry-frontend/src/main/java/com/github/ambry/frontend/FrontendMetrics.java

@@ -255,6 +255,7 @@ public class FrontendMetrics {
   // Errors
   // FrontendRestRequestService
   public final Counter responseSubmissionError;
+  public final Counter generatedBlobIdError;
   public final Counter resourceReleaseError;
   public final Counter ttlTooLargeError;
   public final Counter ttlNotCompliantError;
@@ -707,6 +708,8 @@ public FrontendMetrics(MetricRegistry metricRegistry, FrontendConfig frontendCon
     // FrontendRestRequestService
     responseSubmissionError =
         metricRegistry.counter(MetricRegistry.name(FrontendRestRequestService.class, "ResponseSubmissionError"));
+    generatedBlobIdError =
+        metricRegistry.counter(MetricRegistry.name(FrontendRestRequestService.class, "GeneratedBlobIdError"));
     resourceReleaseError =
         metricRegistry.counter(MetricRegistry.name(FrontendRestRequestService.class, "ResourceReleaseError"));
     // DeleteCallback

ambry-frontend/src/main/java/com/github/ambry/frontend/FrontendRestRequestService.java

@@ -561,6 +561,7 @@ public void handleOptions(RestRequest restRequest, RestResponseChannel restRespo
   /**
    * Submits the response and {@code responseBody} (and any {@code exception})for the {@code restRequest} to the
    * {@code responseHandler}.
+   * Counts invalid IDs encountered while consuming server-generated upload results.
    * @param restRequest the {@link RestRequest} for which a response is ready.
    * @param restResponseChannel the {@link RestResponseChannel} over which the response can be sent.
    * @param responseBody the body of the response in the form of a {@link ReadableStreamChannel}.
@@ -575,8 +576,12 @@ void submitResponse(RestRequest restRequest, RestResponseChannel restResponseCha
         restResponseChannel.setHeader(TrackingHeaders.FRONTEND_NAME, hostname);
       }
       if (exception instanceof RouterException) {
-        exception = new RestServiceException(exception,
-            RestServiceErrorCode.getRestServiceErrorCode(((RouterException) exception).getErrorCode()));
+        RouterErrorCode routerErrorCode = ((RouterException) exception).getErrorCode();
+        if (routerErrorCode == RouterErrorCode.InvalidBlobId
+            && Boolean.TRUE.equals(restRequest.getArgs().get(BLOB_ID_IS_SERVER_GENERATED))) {
+          frontendMetrics.generatedBlobIdError.inc();
+        }
+        exception = new RestServiceException(exception, RestServiceErrorCode.getRestServiceErrorCode(routerErrorCode));
       }
       responseHandler.handleResponse(restRequest, restResponseChannel, responseBody, exception);
     } catch (Exception e) {

ambry-frontend/src/main/java/com/github/ambry/frontend/NamedBlobPutHandler.java

@@ -244,6 +244,7 @@ private Callback<String> routerPutBlobCallback(BlobInfo blobInfo) {
         restRequest.getMetricsTracker().setBytesTransferred(restRequest.getBytesReceived());
         restResponseChannel.setHeader(RestUtils.Headers.BLOB_SIZE, restRequest.getBlobBytesReceived());
         restResponseChannel.setHeader(RestUtils.Headers.LOCATION, blobId);
+        restRequest.setArg(BLOB_ID_IS_SERVER_GENERATED, true);
         String blobIdClean = stripPrefixAndExtension(blobId);
         if (blobInfo.getBlobProperties().getTimeToLiveInSeconds() == Utils.Infinite_Time) {
           // Do ttl update with retryExecutor. Use the blob ID returned from the router instead of the converted ID
@@ -254,6 +255,7 @@ private Callback<String> routerPutBlobCallback(BlobInfo blobInfo) {
                   QuotaUtils.buildQuotaChargeCallback(restRequest, quotaManager, false)), this::isRetriable,
               routerTtlUpdateCallbackForPut(blobInfo));
         } else {
+          restRequest.removeArg(BLOB_ID_IS_SERVER_GENERATED);
           if (RestUtils.isDatasetVersionQueryEnabled(restRequest.getArgs())) {
             //Make sure to process response after delete finished
             updateVersionStateAndDeleteDatasetVersionOutOfRetentionCount(
@@ -291,6 +293,7 @@ private Callback<Long> fetchStitchRequestBodyCallback(RetainingAsyncWritableChan
      */
     private Callback<String> routerStitchBlobCallback(BlobInfo blobInfo, BlobProperties propertiesForRouterUpload) {
       return buildCallback(frontendMetrics.putRouterStitchBlobMetrics, convertedBlobId -> {
+        restRequest.setArg(BLOB_ID_IS_SERVER_GENERATED, true);
         // The actual blob size is now present in the instance of BlobProperties passed to the router.stitchBlob().
         // Update it in the BlobInfo so that IdConverter can add it to the named blob DB
         blobInfo.getBlobProperties().setBlobSize(propertiesForRouterUpload.getBlobSize());
@@ -305,6 +308,7 @@ private Callback<String> routerStitchBlobCallback(BlobInfo blobInfo, BlobPropert
                   QuotaUtils.buildQuotaChargeCallback(restRequest, quotaManager, false)), this::isRetriable,
               routerTtlUpdateCallbackForStitch(blobInfo));
         } else {
+          restRequest.removeArg(BLOB_ID_IS_SERVER_GENERATED);
           if (RestUtils.isDatasetVersionQueryEnabled(restRequest.getArgs())) {
             //Make sure to process response after delete finished
             updateVersionStateAndDeleteDatasetVersionOutOfRetentionCount(
@@ -335,6 +339,7 @@ private boolean isRetriable(Throwable throwable) {
      */
     private Callback<Void> routerTtlUpdateCallbackForPut(BlobInfo blobInfo) {
       return buildCallback(frontendMetrics.updateBlobTtlRouterMetrics, convertedBlobId -> {
+        restRequest.removeArg(BLOB_ID_IS_SERVER_GENERATED);
         if (RestUtils.isDatasetVersionQueryEnabled(restRequest.getArgs())) {
           //Make sure to process response after delete finished
           updateVersionStateAndDeleteDatasetVersionOutOfRetentionCount(
@@ -354,6 +359,7 @@ private Callback<Void> routerTtlUpdateCallbackForPut(BlobInfo blobInfo) {
      */
     private Callback<Void> routerTtlUpdateCallbackForStitch(BlobInfo blobInfo) {
       return buildCallback(frontendMetrics.updateBlobTtlRouterMetrics, convertedBlobId -> {
+        restRequest.removeArg(BLOB_ID_IS_SERVER_GENERATED);
         if (RestUtils.isDatasetVersionQueryEnabled(restRequest.getArgs())) {
           //Make sure to process response after delete finished
           updateVersionStateAndDeleteDatasetVersionOutOfRetentionCount(

ambry-frontend/src/main/java/com/github/ambry/frontend/PostBlobHandler.java

@@ -223,9 +223,10 @@ private Callback<Long> fetchStitchRequestBodyCallback(RetainingAsyncWritableChan
      * @return a {@link Callback} to be used with {@link Router#putBlob}.
      */
     private Callback<String> routerStitchBlobCallback(BlobInfo blobInfo) {
-      return buildCallback(frontendMetrics.postRouterStitchBlobMetrics,
-          blobId -> idConverter.convert(restRequest, blobId, blobInfo.getBlobProperties(), idConverterCallback(blobInfo)), uri, LOGGER,
-          finalCallback);
+      return buildCallback(frontendMetrics.postRouterStitchBlobMetrics, blobId -> {
+        restRequest.setArg(RestUtils.InternalKeys.BLOB_ID_IS_SERVER_GENERATED, true);
+        idConverter.convert(restRequest, blobId, blobInfo.getBlobProperties(), idConverterCallback(blobInfo));
+      }, uri, LOGGER, finalCallback);
     }
 
     /**
@@ -238,6 +239,7 @@ private Callback<String> routerPutBlobCallback(BlobInfo blobInfo) {
       return buildCallback(frontendMetrics.postRouterPutBlobMetrics, blobId -> {
         setSignedIdMetadataAndBlobSize(blobInfo.getBlobProperties());
         restRequest.getMetricsTracker().setBytesTransferred(restRequest.getBytesReceived());
+        restRequest.setArg(RestUtils.InternalKeys.BLOB_ID_IS_SERVER_GENERATED, true);
         idConverter.convert(restRequest, blobId, idConverterCallback(blobInfo));
       }, uri, LOGGER, finalCallback);
     }
@@ -250,6 +252,7 @@ private Callback<String> routerPutBlobCallback(BlobInfo blobInfo) {
      */
     private Callback<String> idConverterCallback(BlobInfo blobInfo) {
       return buildCallback(frontendMetrics.postIdConversionMetrics, convertedBlobId -> {
+        restRequest.removeArg(RestUtils.InternalKeys.BLOB_ID_IS_SERVER_GENERATED);
         restResponseChannel.setHeader(RestUtils.Headers.LOCATION, convertedBlobId);
         securityService.processResponse(restRequest, restResponseChannel, blobInfo, securityProcessResponseCallback());
       }, uri, LOGGER, finalCallback);

ambry-frontend/src/main/java/com/github/ambry/frontend/s3/S3MultipartCompleteUploadHandler.java

@@ -241,6 +241,7 @@ private Callback<Long> fetchStitchRequestBodyCallback(RetainingAsyncWritableChan
     private Callback<String> routerStitchBlobCallback(BlobInfo blobInfo,
         BlobProperties propertiesPassedInRouterUpload) {
       return buildCallback(frontendMetrics.putRouterStitchBlobMetrics, blobId -> {
+        restRequest.setArg(BLOB_ID_IS_SERVER_GENERATED, true);
         // The actual blob size is now present in the instance of BlobProperties passed to the router.stitchBlob().
         // Update it in the BlobInfo so that IdConverter can add it to the named blob DB
         blobInfo.getBlobProperties().setBlobSize(propertiesPassedInRouterUpload.getBlobSize());
@@ -277,6 +278,7 @@ private Callback<String> idConverterCallback(BlobInfo blobInfo, String blobId) {
      */
     private Callback<Void> routerTtlUpdateCallback(BlobInfo blobInfo, String blobId) {
       return buildCallback(frontendMetrics.updateBlobTtlRouterMetrics, convertedBlobId -> {
+        restRequest.removeArg(BLOB_ID_IS_SERVER_GENERATED);
         // Set the named blob state to be 'READY' after the Ttl update succeed
         if (!restRequest.getArgs().containsKey(NAMED_BLOB_VERSION)) {
           throw new RestServiceException(

ambry-frontend/src/main/java/com/github/ambry/frontend/s3/S3MultipartUploadPartHandler.java

@@ -188,6 +188,7 @@ private Callback<Void> securityPostProcessRequestCallback(BlobInfo blobInfo) {
      */
     private Callback<String> routerPutBlobCallback(BlobInfo blobInfo) {
       return buildCallback(frontendMetrics.putRouterPutBlobMetrics, blobId -> {
+        restRequest.setArg(BLOB_ID_IS_SERVER_GENERATED, true);
         restResponseChannel.setHeader(RestUtils.Headers.BLOB_SIZE, restRequest.getBlobBytesReceived());
         // TODO [S3] Make changes to sign ETags. Currently they are sent as shown below.
         //  ETag: {"chunks":[{"blob":"AAYQAQBlAAgAAQAAAAAAAAAAw6UGCoNgS8KGgV-SGXAMdQ","size":4194304},
@@ -196,6 +197,7 @@ private Callback<String> routerPutBlobCallback(BlobInfo blobInfo) {
         S3MultipartETag etag = new S3MultipartETag(putBlobMetaInfo.getOrderedChunkIdSizeList());
         String eTagStr = S3MultipartETag.serialize(etag);
         restResponseChannel.setHeader(RestUtils.Headers.LOCATION, eTagStr);
+        restRequest.removeArg(BLOB_ID_IS_SERVER_GENERATED);
         securityService.processResponse(restRequest, restResponseChannel, blobInfo, securityProcessResponseCallback());
       }, uri, logger, finalCallback);
     }