>From Savyasach Reddy <[email protected]>:
Savyasach Reddy has uploaded this change for review. (
https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/18645 )
Change subject: Add Guardian and Request Limits for GCS
......................................................................
Add Guardian and Request Limits for GCS
Change-Id: Id639afcadb1d88b4630e12dde40dbaae94e15f23
---
A
asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/google/gcs/GCSRequestRateLimiter.java
M
asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/google/gcs/GCSCloudClient.java
M asterixdb/asterix-app/src/test/resources/cc-cloud-storage-gcs.conf
M
asterixdb/asterix-app/src/test/java/org/apache/asterix/test/cloud_storage/CloudStorageGCSTest.java
M
asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/google/gcs/GCSClientConfig.java
M
asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/google/gcs/GCSWriter.java
6 files changed, 110 insertions(+), 8 deletions(-)
git pull ssh://asterix-gerrit.ics.uci.edu:29418/asterixdb
refs/changes/45/18645/1
diff --git
a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/cloud_storage/CloudStorageGCSTest.java
b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/cloud_storage/CloudStorageGCSTest.java
index 3a03445..89a4781 100644
---
a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/cloud_storage/CloudStorageGCSTest.java
+++
b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/cloud_storage/CloudStorageGCSTest.java
@@ -36,7 +36,6 @@
import org.junit.Assume;
import org.junit.BeforeClass;
import org.junit.FixMethodOrder;
-import org.junit.Ignore;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.MethodSorters;
@@ -55,7 +54,6 @@
*/
@RunWith(Parameterized.class)
@FixMethodOrder(MethodSorters.NAME_ASCENDING)
-@Ignore
public class CloudStorageGCSTest {
private static final Logger LOGGER = LogManager.getLogger();
diff --git a/asterixdb/asterix-app/src/test/resources/cc-cloud-storage-gcs.conf
b/asterixdb/asterix-app/src/test/resources/cc-cloud-storage-gcs.conf
index 3c883a8..cbaeb6e 100644
--- a/asterixdb/asterix-app/src/test/resources/cc-cloud-storage-gcs.conf
+++ b/asterixdb/asterix-app/src/test/resources/cc-cloud-storage-gcs.conf
@@ -70,3 +70,5 @@
cloud.storage.endpoint=http://127.0.0.1:4443
cloud.storage.anonymous.auth=true
cloud.storage.cache.policy=selective
+cloud.max.write.requests.per.second=1000
+cloud.max.read.requests.per.second=5000
diff --git
a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/google/gcs/GCSClientConfig.java
b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/google/gcs/GCSClientConfig.java
index 4edb7a7..a034c38 100644
---
a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/google/gcs/GCSClientConfig.java
+++
b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/google/gcs/GCSClientConfig.java
@@ -40,20 +40,33 @@
private final String prefix;
private final boolean anonymousAuth;
private final long profilerLogInterval;
+ private final long tokenAcquireTimeout;
+ private final int readMaxRequestsPerSeconds;
+ private final int writeMaxRequestsPerSeconds;
- public GCSClientConfig(String region, String endpoint, String prefix,
boolean anonymousAuth,
- long profilerLogInterval) {
+ private GCSClientConfig(String region, String endpoint, String prefix,
boolean anonymousAuth,
+ long profilerLogInterval, long tokenAcquireTimeout, int
writeMaxRequestsPerSeconds,
+ int readMaxRequestsPerSeconds) {
this.region = region;
this.endpoint = endpoint;
this.prefix = prefix;
this.anonymousAuth = anonymousAuth;
this.profilerLogInterval = profilerLogInterval;
+ this.tokenAcquireTimeout = tokenAcquireTimeout;
+ this.writeMaxRequestsPerSeconds = writeMaxRequestsPerSeconds;
+ this.readMaxRequestsPerSeconds = readMaxRequestsPerSeconds;
+ }
+
+ public GCSClientConfig(String region, String endpoint, String prefix,
boolean anonymousAuth,
+ long profilerLogInterval) {
+ this(region, endpoint, prefix, anonymousAuth, profilerLogInterval, 1,
0, 0);
}
public static GCSClientConfig of(CloudProperties cloudProperties) {
return new GCSClientConfig(cloudProperties.getStorageRegion(),
cloudProperties.getStorageEndpoint(),
cloudProperties.getStoragePrefix(),
cloudProperties.isStorageAnonymousAuth(),
- cloudProperties.getProfilerLogInterval());
+ cloudProperties.getProfilerLogInterval(),
cloudProperties.getTokenAcquireTimeout(),
+ cloudProperties.getWriteMaxRequestsPerSecond(),
cloudProperties.getReadMaxRequestsPerSecond());
}
public static GCSClientConfig of(Map<String, String> configuration) {
@@ -94,4 +107,16 @@
throw HyracksDataException.create(e);
}
}
+
+ public long getTokenAcquireTimeout() {
+ return tokenAcquireTimeout;
+ }
+
+ public int getWriteMaxRequestsPerSeconds() {
+ return writeMaxRequestsPerSeconds;
+ }
+
+ public int getReadMaxRequestsPerSeconds() {
+ return readMaxRequestsPerSeconds;
+ }
}
diff --git
a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/google/gcs/GCSCloudClient.java
b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/google/gcs/GCSCloudClient.java
index de242bd..59eb1bf 100644
---
a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/google/gcs/GCSCloudClient.java
+++
b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/google/gcs/GCSCloudClient.java
@@ -98,7 +98,7 @@
@Override
public ICloudWriter createWriter(String bucket, String path,
IWriteBufferProvider bufferProvider) {
- return new GCSWriter(bucket, path, gcsClient, profilerLimiter);
+ return new GCSWriter(bucket, path, gcsClient, profilerLimiter,
guardian);
}
@Override
@@ -119,6 +119,7 @@
@Override
public int read(String bucket, String path, long offset, ByteBuffer
buffer) throws HyracksDataException {
+ guardian.checkReadAccess(bucket, path);
profilerLimiter.objectGet();
BlobId blobId = BlobId.of(bucket, path);
long readTo = offset + buffer.remaining();
@@ -140,6 +141,7 @@
@Override
public byte[] readAllBytes(String bucket, String path) {
+ guardian.checkReadAccess(bucket, path);
profilerLimiter.objectGet();
BlobId blobId = BlobId.of(bucket, path);
try {
@@ -151,6 +153,7 @@
@Override
public InputStream getObjectStream(String bucket, String path, long
offset, long length) {
+ guardian.checkReadAccess(bucket, path);
profilerLimiter.objectGet();
try (ReadChannel reader = gcsClient.reader(bucket, path).limit(offset
+ length)) {
reader.seek(offset);
@@ -170,8 +173,9 @@
@Override
public void copy(String bucket, String srcPath, FileReference destPath) {
- Page<Blob> blobs = gcsClient.list(bucket,
BlobListOption.prefix(srcPath));
+ guardian.checkReadAccess(bucket, srcPath);
profilerLimiter.objectsList();
+ Page<Blob> blobs = gcsClient.list(bucket,
BlobListOption.prefix(srcPath));
for (Blob blob : blobs.iterateAll()) {
profilerLimiter.objectCopy();
BlobId source = blob.getBlobId();
diff --git
a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/google/gcs/GCSRequestRateLimiter.java
b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/google/gcs/GCSRequestRateLimiter.java
new file mode 100644
index 0000000..94e0486
--- /dev/null
+++
b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/google/gcs/GCSRequestRateLimiter.java
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.asterix.cloud.clients.google.gcs;
+
+import org.apache.asterix.cloud.clients.profiler.limiter.IRateLimiter;
+import org.apache.asterix.cloud.clients.profiler.limiter.IRequestRateLimiter;
+import org.apache.asterix.cloud.clients.profiler.limiter.NoOpRateLimiter;
+import org.apache.asterix.cloud.clients.profiler.limiter.TokenBasedRateLimiter;
+
+public class GCSRequestRateLimiter implements IRequestRateLimiter {
+ private final IRateLimiter writeLimiter;
+ private final IRateLimiter readLimiter;
+
+ public GCSRequestRateLimiter(GCSClientConfig config) {
+ long tokenAcquireTimeout = config.getTokenAcquireTimeout();
+ this.writeLimiter =
createLimiter(config.getWriteMaxRequestsPerSeconds(), tokenAcquireTimeout);
+ this.readLimiter =
createLimiter(config.getReadMaxRequestsPerSeconds(), tokenAcquireTimeout);
+ }
+
+ @Override
+ public void writeRequest() {
+ writeLimiter.acquire();
+ }
+
+ @Override
+ public void readRequest() {
+ readLimiter.acquire();
+ }
+
+ @Override
+ public void listRequest() {
+ writeLimiter.acquire();
+ }
+
+ private static IRateLimiter createLimiter(int maxRequestsPerSecond, long
tokeAcquireTimeout) {
+ if (maxRequestsPerSecond > 0) {
+ return new TokenBasedRateLimiter(maxRequestsPerSecond,
tokeAcquireTimeout);
+ }
+ return NoOpRateLimiter.INSTANCE;
+ }
+}
diff --git
a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/google/gcs/GCSWriter.java
b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/google/gcs/GCSWriter.java
index 41d1a71..d9c916b 100644
---
a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/google/gcs/GCSWriter.java
+++
b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/clients/google/gcs/GCSWriter.java
@@ -23,6 +23,7 @@
import java.io.IOException;
import java.nio.ByteBuffer;
+import org.apache.asterix.cloud.clients.ICloudGuardian;
import org.apache.asterix.cloud.clients.ICloudWriter;
import org.apache.asterix.cloud.clients.profiler.IRequestProfilerLimiter;
import org.apache.hyracks.api.exceptions.HyracksDataException;
@@ -40,14 +41,18 @@
private final String path;
private final IRequestProfilerLimiter profiler;
private final Storage gcsClient;
+ private final ICloudGuardian guardian;
+
private WriteChannel writer = null;
private long writtenBytes;
- public GCSWriter(String bucket, String path, Storage gcsClient,
IRequestProfilerLimiter profiler) {
+ public GCSWriter(String bucket, String path, Storage gcsClient,
IRequestProfilerLimiter profiler,
+ ICloudGuardian guardian) {
this.bucket = bucket;
this.path = path;
this.profiler = profiler;
this.gcsClient = gcsClient;
+ this.guardian = guardian;
writtenBytes = 0;
}
@@ -58,6 +63,7 @@
@Override
public int write(ByteBuffer page) throws HyracksDataException {
+ guardian.checkIsolatedWriteAccess(bucket, path);
profiler.objectMultipartUpload();
setUploadId();
int written = 0;
@@ -93,6 +99,7 @@
@Override
public void finish() throws HyracksDataException {
+ guardian.checkWriteAccess(bucket, path);
setUploadId();
profiler.objectMultipartUpload();
try {
--
To view, visit https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/18645
To unsubscribe, or for help writing mail filters, visit
https://asterix-gerrit.ics.uci.edu/settings
Gerrit-Project: asterixdb
Gerrit-Branch: master
Gerrit-Change-Id: Id639afcadb1d88b4630e12dde40dbaae94e15f23
Gerrit-Change-Number: 18645
Gerrit-PatchSet: 1
Gerrit-Owner: Savyasach Reddy <[email protected]>
Gerrit-MessageType: newchange
