feat: update Storage.createFrom(BlobInfo, Path) to have 150% higher t…

…hroughput (#2059)

When uploading a file where we are able to rewind to an arbitrary offset, we can be more optimistic in the way we send requests to GCS.

Add new code middleware to allow PUTing an entire file to GCS in a single request, and using query resumable session to recover from the specific offset in the case of retryable error.


### Benchmark Results

#### Methodology

Generate a random file on disk of size `128KiB..2GiB` from `/dev/urandom`, then upload the generated file using `Storage.createFrom(BlobInfo, Path)`.

Perform each 4096 times.

Run on a c2-standard-60 instance is us-central1 against a regional bucket located in us-central1.

#### Results

The following summary of throughput in MiB/s as observed between the existing implementation, and the new implementation proposed in this PR.

```
                                count     mean     std    min      50%      75%      90%      99%      max
runId                 ApiName                                                                             
createFrom - existing JSON     4096.0   66.754  10.988  3.249   67.317   73.476   78.961   91.197  107.247
createFrom - new      JSON     4096.0  158.769  67.105  4.600  170.680  218.618  240.992  266.297  305.205
```

#### Comparison

When comparing the new implementation to the existing implementation we get the following improvement to throughput (higher is better):
```
stat      pct
mean  137.841
 50%  153.547
 90%  205.204
 99%  192.003
```

google-cloud-storage/src/main/java/com/google/cloud/storage/ByteRangeSpec.java

@@ -16,6 +16,9 @@
 
 package com.google.cloud.storage;
 
+import static com.google.api.client.util.Preconditions.checkNotNull;
+import static com.google.common.base.Preconditions.checkArgument;
+
 import com.google.api.core.InternalApi;
 import com.google.common.base.MoreObjects;
 import com.google.common.base.MoreObjects.ToStringHelper;
@@ -124,6 +127,20 @@ static ByteRangeSpec explicitClosed(
     return create(beginOffset, endOffsetInclusive, LeftClosedRightClosedByteRangeSpec::new);
   }
 
+  static ByteRangeSpec parse(String string) {
+    checkNotNull(string, "Range header is null");
+    checkArgument(string.startsWith("bytes="), "malformed Range header value: %s", string);
+
+    int i = string.indexOf('-');
+    String minS = string.substring(6, i);
+    String maxS = string.substring(i + 1);
+
+    long min = Long.parseLong(minS);
+    long max = Long.parseLong(maxS);
+
+    return explicitClosed(min, max);
+  }
+
   private static ByteRangeSpec create(
       @Nullable Long beginOffset,
       @Nullable Long length,

google-cloud-storage/src/main/java/com/google/cloud/storage/ByteSizeConstants.java

@@ -23,10 +23,16 @@ final class ByteSizeConstants {
   static final int _256KiB = 256 * _1KiB;
   static final int _384KiB = 384 * _1KiB;
   static final int _512KiB = 512 * _1KiB;
+  static final int _768KiB = 768 * _1KiB;
   static final int _1MiB = 1024 * _1KiB;
   static final int _2MiB = 2 * _1MiB;
   static final int _16MiB = 16 * _1MiB;
   static final int _32MiB = 32 * _1MiB;
 
+  static final long _128KiBL = 131072L;
+  static final long _256KiBL = 262144L;
+  static final long _512KiBL = 524288L;
+  static final long _768KiBL = 786432L;
+
   private ByteSizeConstants() {}
 }

google-cloud-storage/src/main/java/com/google/cloud/storage/HttpClientContext.java

@@ -0,0 +1,85 @@
+/*
+ * Copyright 2023 Google LLC
+ *
+ * Licensed 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.google.cloud.storage;
+
+import com.google.api.client.http.HttpHeaders;
+import com.google.api.client.http.HttpRequestFactory;
+import com.google.api.client.json.JsonObjectParser;
+import com.google.api.client.util.ObjectParser;
+import com.google.cloud.storage.spi.v1.StorageRpc;
+import io.opencensus.trace.Span;
+import io.opencensus.trace.Tracer;
+import io.opencensus.trace.Tracing;
+import java.util.List;
+import org.checkerframework.checker.nullness.qual.NonNull;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+final class HttpClientContext {
+
+  private final HttpRequestFactory requestFactory;
+  private final ObjectParser objectParser;
+  private final Tracer tracer;
+
+  private HttpClientContext(
+      HttpRequestFactory requestFactory, ObjectParser objectParser, Tracer tracer) {
+    this.requestFactory = requestFactory;
+    this.objectParser = objectParser;
+    this.tracer = tracer;
+  }
+
+  @SuppressWarnings({"unchecked", "SameParameterValue"})
+  static @Nullable String firstHeaderValue(
+      @NonNull HttpHeaders headers, @NonNull String headerName) {
+    Object v = headers.get(headerName);
+    // HttpHeaders doesn't type its get method, so we have to jump through hoops here
+    if (v instanceof List) {
+      List<String> list = (List<String>) v;
+      return list.get(0);
+    } else {
+      return null;
+    }
+  }
+
+  public HttpRequestFactory getRequestFactory() {
+    return requestFactory;
+  }
+
+  public ObjectParser getObjectParser() {
+    return objectParser;
+  }
+
+  public Tracer getTracer() {
+    return tracer;
+  }
+
+  public Span startSpan(String name) {
+    // record events is hardcoded to true in HttpStorageRpc, preserve it here
+    return tracer.spanBuilder(name).setRecordEvents(true).startSpan();
+  }
+
+  static HttpClientContext from(StorageRpc storageRpc) {
+    return new HttpClientContext(
+        storageRpc.getStorage().getRequestFactory(),
+        storageRpc.getStorage().getObjectParser(),
+        Tracing.getTracer());
+  }
+
+  public static HttpClientContext of(
+      HttpRequestFactory requestFactory, JsonObjectParser jsonObjectParser) {
+    return new HttpClientContext(requestFactory, jsonObjectParser, Tracing.getTracer());
+  }
+}

google-cloud-storage/src/main/java/com/google/cloud/storage/HttpContentRange.java

@@ -0,0 +1,249 @@
+/*
+ * Copyright 2023 Google LLC
+ *
+ * Licensed 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.google.cloud.storage;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+import com.google.common.base.MoreObjects;
+import java.util.Objects;
+import java.util.function.UnaryOperator;
+
+abstract class HttpContentRange {
+
+  private final boolean finalizing;
+
+  private HttpContentRange(boolean finalizing) {
+    this.finalizing = finalizing;
+  }
+
+  public abstract String getHeaderValue();
+
+  public boolean isFinalizing() {
+    return finalizing;
+  }
+
+  static Total of(ByteRangeSpec spec, long size) {
+    checkArgument(size >= 0, "size must be >= 0");
+    checkArgument(size >= spec.endOffsetInclusive(), "size must be >= end");
+    return new Total(spec, size);
+  }
+
+  static Incomplete of(ByteRangeSpec spec) {
+    return new Incomplete(spec);
+  }
+
+  static Size of(long size) {
+    checkArgument(size >= 0, "size must be >= 0");
+    return new Size(size);
+  }
+
+  static Query query() {
+    return Query.INSTANCE;
+  }
+
+  static HttpContentRange parse(String string) {
+    if ("bytes */*".equals(string)) {
+      return HttpContentRange.query();
+    } else if (string.startsWith("bytes */")) {
+      return HttpContentRange.of(Long.parseLong(string.substring(8)));
+    } else {
+      int idxDash = string.indexOf('-');
+      int idxSlash = string.indexOf('/');
+
+      String beginS = string.substring(6, idxDash);
+      String endS = string.substring(idxDash + 1, idxSlash);
+      long begin = Long.parseLong(beginS);
+      long end = Long.parseLong(endS);
+      if (string.endsWith("/*")) {
+        return HttpContentRange.of(ByteRangeSpec.explicitClosed(begin, end));
+      } else {
+        String sizeS = string.substring(idxSlash + 1);
+        long size = Long.parseLong(sizeS);
+        return HttpContentRange.of(ByteRangeSpec.explicitClosed(begin, end), size);
+      }
+    }
+  }
+
+  static final class Incomplete extends HttpContentRange implements HasRange<Incomplete> {
+
+    private final ByteRangeSpec spec;
+
+    private Incomplete(ByteRangeSpec spec) {
+      super(false);
+      this.spec = spec;
+    }
+
+    @Override
+    public String getHeaderValue() {
+      return String.format("bytes %d-%d/*", spec.beginOffset(), spec.endOffsetInclusive());
+    }
+
+    @Override
+    public ByteRangeSpec range() {
+      return spec;
+    }
+
+    @Override
+    public Incomplete map(UnaryOperator<ByteRangeSpec> f) {
+      return new Incomplete(f.apply(spec));
+    }
+
+    @Override
+    public boolean equals(Object o) {
+      if (this == o) {
+        return true;
+      }
+      if (!(o instanceof Incomplete)) {
+        return false;
+      }
+      Incomplete that = (Incomplete) o;
+      return Objects.equals(spec, that.spec);
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hash(spec);
+    }
+
+    @Override
+    public String toString() {
+      return MoreObjects.toStringHelper(this).add("spec", spec).toString();
+    }
+  }
+
+  static final class Total extends HttpContentRange implements HasRange<Total>, HasSize {
+
+    private final ByteRangeSpec spec;
+    private final long size;
+
+    private Total(ByteRangeSpec spec, long size) {
+      super(true);
+      this.spec = spec;
+      this.size = size;
+    }
+
+    @Override
+    public String getHeaderValue() {
+      return String.format("bytes %d-%d/%d", spec.beginOffset(), spec.endOffsetInclusive(), size);
+    }
+
+    @Override
+    public long getSize() {
+      return size;
+    }
+
+    @Override
+    public ByteRangeSpec range() {
+      return spec;
+    }
+
+    @Override
+    public Total map(UnaryOperator<ByteRangeSpec> f) {
+      return new Total(f.apply(spec), size);
+    }
+
+    @Override
+    public boolean equals(Object o) {
+      if (this == o) {
+        return true;
+      }
+      if (!(o instanceof Total)) {
+        return false;
+      }
+      Total total = (Total) o;
+      return size == total.size && Objects.equals(spec, total.spec);
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hash(spec, size);
+    }
+
+    @Override
+    public String toString() {
+      return MoreObjects.toStringHelper(this).add("spec", spec).add("size", size).toString();
+    }
+  }
+
+  static final class Size extends HttpContentRange implements HasSize {
+
+    private final long size;
+
+    private Size(long size) {
+      super(true);
+      this.size = size;
+    }
+
+    @Override
+    public String getHeaderValue() {
+      return String.format("bytes */%d", size);
+    }
+
+    @Override
+    public long getSize() {
+      return size;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+      if (this == o) {
+        return true;
+      }
+      if (!(o instanceof Size)) {
+        return false;
+      }
+      Size size1 = (Size) o;
+      return size == size1.size;
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hash(size);
+    }
+
+    @Override
+    public String toString() {
+      return MoreObjects.toStringHelper(this).add("size", size).toString();
+    }
+  }
+
+  static final class Query extends HttpContentRange {
+
+    private static final Query INSTANCE = new Query();
+
+    private Query() {
+      super(false);
+    }
+
+    @Override
+    public String getHeaderValue() {
+      return "bytes */*";
+    }
+  }
+
+  interface HasRange<T extends HttpContentRange> {
+
+    ByteRangeSpec range();
+
+    T map(UnaryOperator<ByteRangeSpec> f);
+  }
+
+  interface HasSize {
+
+    long getSize();
+  }
+}