You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by yc...@apache.org on 2021/09/29 23:12:18 UTC
[cassandra-sidecar] branch trunk updated (1012c0d -> 4d88af6)
This is an automated email from the ASF dual-hosted git repository.
ycai pushed a change to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra-sidecar.git.
discard 1012c0d Avoid having sidecar's health response code depend on Cassandra's health information
omit 3f6dc1f Add Stream SSTable API to Sidecar to stream SSTable components through zero copy streaming
new 4e4b8ea Add Stream SSTable API to Sidecar to stream SSTable components through zero copy streaming
new 4d88af6 Avoid having sidecar's health response code depend on Cassandra's health information
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (1012c0d)
\
N -- N -- N refs/heads/trunk (4d88af6)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
[cassandra-sidecar] 01/02: Add Stream SSTable API to Sidecar to
stream SSTable components through zero copy streaming
Posted by yc...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ycai pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra-sidecar.git
commit 4e4b8eac699606ad09e2f9debc8ef71285d61af3
Author: Saranya Krishnakumar <sa...@apple.com>
AuthorDate: Tue Sep 28 15:28:29 2021 -0700
Add Stream SSTable API to Sidecar to stream SSTable components through zero copy streaming
Patch by Saranya Krishnakumar; reviewed by Dinesh Joshi, Yifan Cai for CASSANDRASC-28
---
.gitignore | 1 -
README.md | 5 +
build.gradle | 3 +-
src/main/dist/conf/sidecar.yaml | 5 +
.../common/util/concurrent/SidecarRateLimiter.java | 36 +++
.../apache/cassandra/sidecar/Configuration.java | 91 +++++-
.../org/apache/cassandra/sidecar/MainModule.java | 27 +-
.../sidecar/exceptions/RangeException.java | 12 +
.../cassandra/sidecar/models/HttpResponse.java | 88 ++++++
.../org/apache/cassandra/sidecar/models/Range.java | 182 ++++++++++++
.../sidecar/routes/StreamSSTableComponent.java | 99 +++++++
.../sidecar/utils/CachedFilePathBuilder.java | 219 +++++++++++++++
.../cassandra/sidecar/utils/FilePathBuilder.java | 145 ++++++++++
.../cassandra/sidecar/utils/FileStreamer.java | 113 ++++++++
.../cassandra/sidecar/FilePathBuilderTest.java | 126 +++++++++
.../org/apache/cassandra/sidecar/RangeTest.java | 86 ++++++
.../sidecar/StreamSSTableComponentTest.java | 308 +++++++++++++++++++++
.../org/apache/cassandra/sidecar/TestModule.java | 8 +-
.../apache/cassandra/sidecar/TestSslModule.java | 3 +
.../org/apache/cassandra/sidecar/ThrottleTest.java | 109 ++++++++
...le-54ea95ce-bba2-4e0a-a9be-e428e5d7160b-Data.db | 1 +
21 files changed, 1653 insertions(+), 14 deletions(-)
diff --git a/.gitignore b/.gitignore
index b00da53..f17cb17 100644
--- a/.gitignore
+++ b/.gitignore
@@ -5,7 +5,6 @@ build/
src/gen-java/
src/resources/org/apache/cassandra/config/
logs/
-data/
conf/hotspot_compiler
doc/cql3/CQL.html
diff --git a/README.md b/README.md
index 17187a6..f7643f6 100644
--- a/README.md
+++ b/README.md
@@ -19,6 +19,11 @@ Apache Cassandra running on the host & port specified in `conf/sidecar.yaml`.
$ ./gradlew run
+Configuring Cassandra Instance
+------------------------------
+
+While setting up cassandra instance, make sure the data directories of cassandra are in the path stored in sidecar.yaml file, else modify data directories path to point to the correct directories for stream APIs to work.
+
Testing
---------
diff --git a/build.gradle b/build.gradle
index 8bfc590..836c4dd 100644
--- a/build.gradle
+++ b/build.gradle
@@ -152,7 +152,7 @@ dependencies {
compile 'io.vertx:vertx-web-client:3.8.5'
compile 'io.swagger.core.v3:swagger-jaxrs2:2.1.0'
- compile 'org.jboss.resteasy:resteasy-vertx:3.1.0.Final'
+ compile 'org.jboss.resteasy:resteasy-vertx:3.1.2.Final'
compile group: 'org.jboss.spec.javax.servlet', name: 'jboss-servlet-api_4.0_spec', version: '2.0.0.Final'
// Trying to be exactly compatible with Cassandra's deps
@@ -170,6 +170,7 @@ dependencies {
testCompile "org.junit.jupiter:junit-jupiter-api:${project.junitVersion}"
testCompile "org.junit.jupiter:junit-jupiter-params:${project.junitVersion}"
+ testCompile "org.assertj:assertj-core:3.14.0"
testRuntimeOnly "org.junit.jupiter:junit-jupiter-engine:${project.junitVersion}"
testCompile group: 'org.cassandraunit', name: 'cassandra-unit-shaded', version: '3.11.2.0'
diff --git a/src/main/dist/conf/sidecar.yaml b/src/main/dist/conf/sidecar.yaml
index f7e1ce3..39dc0a4 100644
--- a/src/main/dist/conf/sidecar.yaml
+++ b/src/main/dist/conf/sidecar.yaml
@@ -5,10 +5,15 @@
cassandra:
- host: localhost
- port: 9042
+ - data_dirs: /cassandra/d1/data, /cassandra/d2/data
sidecar:
- host: 0.0.0.0
- port: 9043
+ - throttle:
+ - stream_requests_per_sec: 5000
+ - delay_sec: 5
+ - timeout_sec: 10
#
# Enable SSL configuration (Disabled by default)
#
diff --git a/src/main/java/com/google/common/util/concurrent/SidecarRateLimiter.java b/src/main/java/com/google/common/util/concurrent/SidecarRateLimiter.java
new file mode 100644
index 0000000..4a1bd79
--- /dev/null
+++ b/src/main/java/com/google/common/util/concurrent/SidecarRateLimiter.java
@@ -0,0 +1,36 @@
+package com.google.common.util.concurrent;
+
+/**
+ * Wrapper class over guava Rate Limiter, uses SmoothBursty Ratelimiter. This class mainly exists to expose
+ * package protected method queryEarliestAvailable of guava RateLimiter
+ */
+public class SidecarRateLimiter
+{
+ private final RateLimiter rateLimiter;
+
+ private SidecarRateLimiter(final double permitsPerSecond)
+ {
+ this.rateLimiter = RateLimiter.create(permitsPerSecond);
+ }
+
+ public static SidecarRateLimiter create(final double permitsPerSecond)
+ {
+ return new SidecarRateLimiter(permitsPerSecond);
+ }
+
+ /**
+ * Returns earliest time permits will become available
+ */
+ public long queryEarliestAvailable(final long nowMicros)
+ {
+ return this.rateLimiter.queryEarliestAvailable(nowMicros);
+ }
+
+ /**
+ * Tries to reserve 1 permit, if not available immediately returns false
+ */
+ public boolean tryAcquire()
+ {
+ return this.rateLimiter.tryAcquire();
+ }
+}
diff --git a/src/main/java/org/apache/cassandra/sidecar/Configuration.java b/src/main/java/org/apache/cassandra/sidecar/Configuration.java
index 5f289c5..088f956 100644
--- a/src/main/java/org/apache/cassandra/sidecar/Configuration.java
+++ b/src/main/java/org/apache/cassandra/sidecar/Configuration.java
@@ -18,6 +18,8 @@
package org.apache.cassandra.sidecar;
+import java.util.Collection;
+import java.util.List;
import javax.annotation.Nullable;
/**
@@ -31,6 +33,9 @@ public class Configuration
/* Cassandra Port */
private final Integer cassandraPort;
+ /* Cassandra Data Dirs */
+ private Collection<String> cassandraDataDirs;
+
/* Sidecar's HTTP REST API port */
private final Integer port;
@@ -55,15 +60,22 @@ public class Configuration
private final boolean isSslEnabled;
- public Configuration(String cassandraHost, Integer cassandraPort, String host, Integer port,
- Integer healthCheckFrequencyMillis, boolean isSslEnabled,
- @Nullable String keyStorePath,
- @Nullable String keyStorePassword,
- @Nullable String trustStorePath,
- @Nullable String trustStorePassword)
+ private final long rateLimitStreamRequestsPerSecond;
+
+ private final long throttleTimeoutInSeconds;
+
+ private final long throttleDelayInSeconds;
+
+ public Configuration(String cassandraHost, Integer cassandraPort, List<String> cassandraDataDirs, String host,
+ Integer port, Integer healthCheckFrequencyMillis, boolean isSslEnabled,
+ @Nullable String keyStorePath, @Nullable String keyStorePassword,
+ @Nullable String trustStorePath, @Nullable String trustStorePassword,
+ long rateLimitStreamRequestsPerSecond, long throttleTimeoutInSeconds,
+ long throttleDelayInSeconds)
{
this.cassandraHost = cassandraHost;
this.cassandraPort = cassandraPort;
+ this.cassandraDataDirs = cassandraDataDirs;
this.host = host;
this.port = port;
this.healthCheckFrequencyMillis = healthCheckFrequencyMillis;
@@ -73,6 +85,9 @@ public class Configuration
this.trustStorePath = trustStorePath;
this.trustStorePassword = trustStorePassword;
this.isSslEnabled = isSslEnabled;
+ this.rateLimitStreamRequestsPerSecond = rateLimitStreamRequestsPerSecond;
+ this.throttleTimeoutInSeconds = throttleTimeoutInSeconds;
+ this.throttleDelayInSeconds = throttleDelayInSeconds;
}
/**
@@ -96,6 +111,16 @@ public class Configuration
}
/**
+ * Get Cassandra data dirs
+ *
+ * @return
+ */
+ public Collection<String> getCassandraDataDirs()
+ {
+ return cassandraDataDirs;
+ }
+
+ /**
* Sidecar's listen address
*
* @return
@@ -180,12 +205,33 @@ public class Configuration
}
/**
+ * Get number of stream requests accepted per second
+ *
+ * @return
+ */
+ public long getRateLimitStreamRequestsPerSecond()
+ {
+ return rateLimitStreamRequestsPerSecond;
+ }
+
+ public long getThrottleTimeoutInSeconds()
+ {
+ return throttleTimeoutInSeconds;
+ }
+
+ public long getThrottleDelayInSeconds()
+ {
+ return throttleDelayInSeconds;
+ }
+
+ /**
* Configuration Builder
*/
public static class Builder
{
private String cassandraHost;
private Integer cassandraPort;
+ private List<String> cassandraDataDirs;
private String host;
private Integer port;
private Integer healthCheckFrequencyMillis;
@@ -194,6 +240,9 @@ public class Configuration
private String trustStorePath;
private String trustStorePassword;
private boolean isSslEnabled;
+ private long rateLimitStreamRequestsPerSecond;
+ private long throttleTimeoutInSeconds;
+ private long throttleDelayInSeconds;
public Builder setCassandraHost(String host)
{
@@ -207,6 +256,12 @@ public class Configuration
return this;
}
+ public Builder setCassandraDataDirs(List<String> dataDirs)
+ {
+ this.cassandraDataDirs = dataDirs;
+ return this;
+ }
+
public Builder setHost(String host)
{
this.host = host;
@@ -255,10 +310,30 @@ public class Configuration
return this;
}
+ public Builder setRateLimitStreamRequestsPerSecond(long rateLimitStreamRequestsPerSecond)
+ {
+ this.rateLimitStreamRequestsPerSecond = rateLimitStreamRequestsPerSecond;
+ return this;
+ }
+
+ public Builder setThrottleTimeoutInSeconds(long throttleTimeoutInSeconds)
+ {
+ this.throttleTimeoutInSeconds = throttleTimeoutInSeconds;
+ return this;
+ }
+
+ public Builder setThrottleDelayInSeconds(long throttleDelayInSeconds)
+ {
+ this.throttleDelayInSeconds = throttleDelayInSeconds;
+ return this;
+ }
+
public Configuration build()
{
- return new Configuration(cassandraHost, cassandraPort, host, port, healthCheckFrequencyMillis, isSslEnabled,
- keyStorePath, keyStorePassword, trustStorePath, trustStorePassword);
+ return new Configuration(cassandraHost, cassandraPort, cassandraDataDirs, host, port,
+ healthCheckFrequencyMillis, isSslEnabled, keyStorePath, keyStorePassword,
+ trustStorePath, trustStorePassword, rateLimitStreamRequestsPerSecond,
+ throttleTimeoutInSeconds, throttleDelayInSeconds);
}
}
}
diff --git a/src/main/java/org/apache/cassandra/sidecar/MainModule.java b/src/main/java/org/apache/cassandra/sidecar/MainModule.java
index 27eefd7..223ab97 100644
--- a/src/main/java/org/apache/cassandra/sidecar/MainModule.java
+++ b/src/main/java/org/apache/cassandra/sidecar/MainModule.java
@@ -23,6 +23,7 @@ import java.io.InputStream;
import java.net.MalformedURLException;
import java.net.URL;
+import com.google.common.util.concurrent.SidecarRateLimiter;
import org.apache.commons.configuration2.YAMLConfiguration;
import org.apache.commons.configuration2.ex.ConfigurationException;
import org.slf4j.Logger;
@@ -45,7 +46,10 @@ import org.apache.cassandra.sidecar.common.CQLSession;
import org.apache.cassandra.sidecar.common.CassandraAdapterDelegate;
import org.apache.cassandra.sidecar.common.CassandraVersionProvider;
import org.apache.cassandra.sidecar.routes.HealthService;
+import org.apache.cassandra.sidecar.routes.StreamSSTableComponent;
import org.apache.cassandra.sidecar.routes.SwaggerOpenApiResource;
+import org.apache.cassandra.sidecar.utils.CachedFilePathBuilder;
+import org.apache.cassandra.sidecar.utils.FilePathBuilder;
import org.jboss.resteasy.plugins.server.vertx.VertxRegistry;
import org.jboss.resteasy.plugins.server.vertx.VertxRequestHandler;
import org.jboss.resteasy.plugins.server.vertx.VertxResteasyDeployment;
@@ -96,7 +100,9 @@ public class MainModule extends AbstractModule
@Provides
@Singleton
- private VertxRequestHandler configureServices(Vertx vertx, HealthService healthService)
+ private VertxRequestHandler configureServices(Vertx vertx,
+ HealthService healthService,
+ StreamSSTableComponent ssTableComponent)
{
VertxResteasyDeployment deployment = new VertxResteasyDeployment();
deployment.start();
@@ -104,6 +110,7 @@ public class MainModule extends AbstractModule
r.addPerInstanceResource(SwaggerOpenApiResource.class);
r.addSingletonResource(healthService);
+ r.addSingletonResource(ssTableComponent);
return new VertxRequestHandler(vertx, deployment);
}
@@ -122,7 +129,6 @@ public class MainModule extends AbstractModule
// Docs index.html page
StaticHandler docs = StaticHandler.create("docs");
router.route().path("/docs/*").handler(docs);
-
return router;
}
@@ -143,6 +149,7 @@ public class MainModule extends AbstractModule
return new Configuration.Builder()
.setCassandraHost(yamlConf.get(String.class, "cassandra.host"))
.setCassandraPort(yamlConf.get(Integer.class, "cassandra.port"))
+ .setCassandraDataDirs(yamlConf.getList(String.class, "cassandra.data_dirs"))
.setHost(yamlConf.get(String.class, "sidecar.host"))
.setPort(yamlConf.get(Integer.class, "sidecar.port"))
.setHealthCheckFrequency(yamlConf.get(Integer.class, "healthcheck.poll_freq_millis"))
@@ -151,6 +158,9 @@ public class MainModule extends AbstractModule
.setTrustStorePath(yamlConf.get(String.class, "sidecar.ssl.truststore.path", null))
.setTrustStorePassword(yamlConf.get(String.class, "sidecar.ssl.truststore.password", null))
.setSslEnabled(yamlConf.get(Boolean.class, "sidecar.ssl.enabled", false))
+ .setRateLimitStreamRequestsPerSecond(yamlConf.getLong("sidecar.throttle.stream_requests_per_sec"))
+ .setThrottleTimeoutInSeconds(yamlConf.getLong("sidecar.throttle.timeout_sec"))
+ .setThrottleDelayInSeconds(yamlConf.getLong("sidecar.throttle.delay_sec"))
.build();
}
catch (MalformedURLException e)
@@ -184,4 +194,17 @@ public class MainModule extends AbstractModule
{
return new CassandraAdapterDelegate(provider, session, config.getHealthCheckFrequencyMillis());
}
+
+ @Provides
+ public SidecarRateLimiter rateLimiter(Configuration config)
+ {
+ return SidecarRateLimiter.create(config.getRateLimitStreamRequestsPerSecond());
+ }
+
+ @Provides
+ @Singleton
+ public FilePathBuilder filePathBuilder(Configuration config)
+ {
+ return new CachedFilePathBuilder(config.getCassandraDataDirs());
+ }
}
diff --git a/src/main/java/org/apache/cassandra/sidecar/exceptions/RangeException.java b/src/main/java/org/apache/cassandra/sidecar/exceptions/RangeException.java
new file mode 100644
index 0000000..8ea135f
--- /dev/null
+++ b/src/main/java/org/apache/cassandra/sidecar/exceptions/RangeException.java
@@ -0,0 +1,12 @@
+package org.apache.cassandra.sidecar.exceptions;
+
+/**
+ * Custom exception
+ */
+public class RangeException extends RuntimeException
+{
+ public RangeException(String msg)
+ {
+ super(msg);
+ }
+}
diff --git a/src/main/java/org/apache/cassandra/sidecar/models/HttpResponse.java b/src/main/java/org/apache/cassandra/sidecar/models/HttpResponse.java
new file mode 100644
index 0000000..708221e
--- /dev/null
+++ b/src/main/java/org/apache/cassandra/sidecar/models/HttpResponse.java
@@ -0,0 +1,88 @@
+package org.apache.cassandra.sidecar.models;
+
+import java.io.File;
+
+import io.netty.handler.codec.http.HttpHeaderNames;
+import io.netty.handler.codec.http.HttpHeaderValues;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.vertx.core.http.HttpServerResponse;
+
+import static java.util.concurrent.TimeUnit.MICROSECONDS;
+
+/**
+ * Wrapper around HttpServerResponse
+ */
+public class HttpResponse
+{
+ private final HttpServerResponse response;
+
+ public HttpResponse(HttpServerResponse response)
+ {
+ this.response = response;
+ }
+
+ public void setRetryAfterHeader(long microsToWait)
+ {
+ response.setStatusCode(HttpResponseStatus.TOO_MANY_REQUESTS.code());
+ response.putHeader(HttpHeaderNames.RETRY_AFTER, Long.toString(MICROSECONDS.toSeconds(microsToWait))).end();
+ }
+
+ public void setTooManyRequestsStatus()
+ {
+ response.setStatusCode(HttpResponseStatus.TOO_MANY_REQUESTS.code()).end();
+ }
+
+ public void setRangeNotSatisfiable(String msg)
+ {
+ response.setStatusCode(HttpResponseStatus.REQUESTED_RANGE_NOT_SATISFIABLE.code()).setStatusMessage(msg).end();
+ }
+
+ public void setPartialContentStatus(Range range)
+ {
+ response.setStatusCode(HttpResponseStatus.PARTIAL_CONTENT.code())
+ .putHeader(HttpHeaderNames.CONTENT_RANGE, contentRangeHeader(range));
+ }
+
+ private String contentRangeHeader(Range r)
+ {
+ return "bytes " + r.start() + "-" + r.end() + "/" + r.length();
+ }
+
+ public void setBadRequestStatus(String msg)
+ {
+ response.setStatusCode(HttpResponseStatus.BAD_REQUEST.code()).setStatusMessage(msg).end();
+ }
+
+ public void setNotFoundStatus(String msg)
+ {
+ response.setStatusCode(HttpResponseStatus.NOT_FOUND.code()).setStatusMessage(msg).end();
+ }
+
+ public void setInternalErrorStatus(String msg)
+ {
+ response.setStatusCode(HttpResponseStatus.INTERNAL_SERVER_ERROR.code()).setStatusMessage(msg).end();
+ }
+
+ public void sendFile(File file)
+ {
+ response.putHeader(HttpHeaderNames.CONTENT_TYPE, HttpHeaderValues.APPLICATION_OCTET_STREAM)
+ .putHeader(HttpHeaderNames.CONTENT_LENGTH, Long.toString(file.length()))
+ .sendFile(file.getAbsolutePath());
+ }
+
+ public void sendFile(File file, Range range)
+ {
+ if (range.length() != file.length())
+ {
+ setPartialContentStatus(range);
+ }
+ response.putHeader(HttpHeaderNames.CONTENT_TYPE, HttpHeaderValues.APPLICATION_OCTET_STREAM)
+ .putHeader(HttpHeaderNames.CONTENT_LENGTH, Long.toString(range.length()))
+ .sendFile(file.getAbsolutePath(), range.start(), range.length());
+ }
+
+ public void setForbiddenStatus(String msg)
+ {
+ response.setStatusCode(HttpResponseStatus.FORBIDDEN.code()).setStatusMessage(msg).end();
+ }
+}
diff --git a/src/main/java/org/apache/cassandra/sidecar/models/Range.java b/src/main/java/org/apache/cassandra/sidecar/models/Range.java
new file mode 100644
index 0000000..411a2a6
--- /dev/null
+++ b/src/main/java/org/apache/cassandra/sidecar/models/Range.java
@@ -0,0 +1,182 @@
+package org.apache.cassandra.sidecar.models;
+
+import java.util.Objects;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import javax.validation.constraints.NotNull;
+
+import com.google.common.base.Preconditions;
+
+import org.apache.cassandra.sidecar.exceptions.RangeException;
+
+/**
+ * Accepted Range formats are start-end, start-, -suffix_length
+ * start-end (start = start index of the range, end = end index of the range, both inclusive)
+ * start- (start = start index of the range, end = end of file)
+ * -suffix-length (Requested length from end of file)
+ */
+public class Range
+{
+ private static final Pattern START_END = Pattern.compile("^(\\d+)-(\\d+)$");
+ private static final Pattern PARTIAL = Pattern.compile("^((\\d+)-)$|^(-(\\d+))$");
+ private static final String RANGE_UNIT = "bytes";
+ private final long start;
+ private final long end;
+ private final long length;
+
+ private Range(final long start, final long end)
+ {
+ this.start = start;
+ this.end = end;
+ this.length = length(start, end);
+ }
+
+ public Range(final long start, final long end, final long length)
+ {
+ this.start = start;
+ this.end = end;
+ this.length = length;
+ }
+
+ private long length(final long start, final long end)
+ {
+ if (start == 0 && end == Long.MAX_VALUE) // avoid overflow (extra byte)
+ {
+ return Long.MAX_VALUE;
+ }
+ return end - start + 1;
+ }
+
+ public long start()
+ {
+ return this.start;
+ }
+
+ public long end()
+ {
+ return this.end;
+ }
+
+ public long length()
+ {
+ return this.length;
+ }
+
+ public boolean isValidHttpRange()
+ {
+ return start >= 0 && end >= start && length > 0;
+ }
+
+ public Range intersect(@NotNull final Range range)
+ {
+ if (!(start >= range.start() && start <= range.end()) && !(end >= range.start() && end <= range.end()) &&
+ !(range.start() >= start && range.start() <= end) && !(range.end() >= start && range.end() <= end))
+ {
+ throw new RangeException("Range does not overlap");
+ }
+
+ return new Range(Math.max(start, range.start()), Math.min(end, range.end()));
+ }
+
+ /**
+ * Accepted string formats "start-end", both ends of the range required to be parsed
+ * Sample accepted formats "1-2", "232-2355"
+ */
+ private static Range parseAbsolute(@NotNull String range)
+ {
+ Matcher m = START_END.matcher(range);
+
+ if (!m.matches())
+ {
+ throw new IllegalArgumentException("Supported Range formats are <start>-<end>, <start>-, -<suffix-length>");
+ }
+
+ final long start = Long.parseLong(m.group(1));
+ Preconditions.checkArgument(start >= 0, "Range start can not be negative");
+
+ final long end = Long.parseLong(m.group(2));
+ if (end < start)
+ {
+ throw new RangeException("Range does not satisfy boundary requirements");
+ }
+ return new Range(start, end);
+ }
+
+ public static Range parse(@NotNull String range)
+ {
+ // since file size is not passed, we set it to 0
+ return parse(range, 0);
+ }
+
+ /**
+ * Accepted string formats "1453-3563", "-22344", "5346-"
+ * Sample invalid string formats "8-3", "-", "-0", "a-b"
+ *
+ * @param fileSize - passed in to convert partial range into absolute range
+ */
+ public static Range parse(@NotNull String range, final long fileSize)
+ {
+ Matcher m = PARTIAL.matcher(range);
+ if (!m.matches())
+ {
+ return parseAbsolute(range);
+ }
+
+ Preconditions.checkArgument(fileSize > 0, "Reference file size invalid");
+ if (range.startsWith("-"))
+ {
+ final long length = Long.parseLong(m.group(4));
+ if (length <= 0)
+ {
+ throw new IllegalArgumentException("Suffix length in " + range + " cannot be less than or equal to 0");
+ }
+ Preconditions.checkArgument(length <= fileSize, "Suffix length exceeds");
+ return new Range(fileSize - length, fileSize - 1, length);
+ }
+
+ final long start = Long.parseLong(m.group(2));
+ Preconditions.checkArgument(start >= 0, "Range start can not be negative");
+ Preconditions.checkArgument(start < fileSize, "Range exceeds");
+
+ return new Range(start, fileSize - 1);
+ }
+
+ @Override
+ public boolean equals(Object o)
+ {
+ if (this == o)
+ {
+ return true;
+ }
+ if (!(o instanceof Range))
+ {
+ return false;
+ }
+ Range range = (Range) o;
+ return start == range.start &&
+ end == range.end &&
+ length == range.length;
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Objects.hash(start, end, length);
+ }
+
+ /**
+ * Accepted RangeHeader formats are bytes=start-end, bytes=start-, bytes=-suffix_length
+ */
+ public static Range parseHeader(final String header, final long fileSize)
+ {
+ if (header == null)
+ {
+ return new Range(0, fileSize - 1, fileSize);
+ }
+ if (!header.startsWith(RANGE_UNIT + "="))
+ {
+ throw new UnsupportedOperationException("Unsupported range unit only bytes are allowed");
+ }
+ return Range.parse(header.substring(header.indexOf("=") + 1), fileSize);
+ }
+}
diff --git a/src/main/java/org/apache/cassandra/sidecar/routes/StreamSSTableComponent.java b/src/main/java/org/apache/cassandra/sidecar/routes/StreamSSTableComponent.java
new file mode 100644
index 0000000..7eb3da8
--- /dev/null
+++ b/src/main/java/org/apache/cassandra/sidecar/routes/StreamSSTableComponent.java
@@ -0,0 +1,99 @@
+package org.apache.cassandra.sidecar.routes;
+
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.nio.file.Path;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.regex.Pattern;
+import javax.ws.rs.GET;
+import javax.ws.rs.HeaderParam;
+import javax.ws.rs.PathParam;
+import javax.ws.rs.core.Context;
+
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+import io.vertx.core.http.HttpServerResponse;
+import org.apache.cassandra.sidecar.models.HttpResponse;
+import org.apache.cassandra.sidecar.models.Range;
+import org.apache.cassandra.sidecar.utils.FilePathBuilder;
+import org.apache.cassandra.sidecar.utils.FileStreamer;
+
+/**
+ * Handler for serving SSTable components from snapshot folders
+ */
+@Singleton
+@javax.ws.rs.Path("/api/v1/stream/keyspace/{keyspace}/table/{table}/snapshot/{snapshot}/component/{component}")
+public class StreamSSTableComponent
+{
+ private static final Pattern REGEX_DIR = Pattern.compile("[a-zA-Z0-9_-]+");
+ private static final Pattern REGEX_COMPONENT = Pattern.compile("[a-zA-Z0-9_-]+(.db|.cql|.json|.crc32|TOC.txt)");
+ private static final Set<String> FORBIDDEN_DIRS = new HashSet<>(
+ Arrays.asList("system_schema", "system_traces", "system_distributed", "system", "system_auth"));
+
+ private final FilePathBuilder pathBuilder;
+ private final FileStreamer fileStreamer;
+
+ @Inject
+ public StreamSSTableComponent(final FilePathBuilder pathBuilder, final FileStreamer fileStreamer)
+ {
+ this.pathBuilder = pathBuilder;
+ this.fileStreamer = fileStreamer;
+ }
+
+ @GET
+ public void handle(@PathParam("keyspace") String keyspace, @PathParam("table") String table,
+ @PathParam("snapshot") String snapshot, @PathParam("component") String component,
+ @HeaderParam("Range") String range, @Context HttpServerResponse resp)
+ {
+ final HttpResponse response = new HttpResponse(resp);
+ if (FORBIDDEN_DIRS.contains(keyspace))
+ {
+ response.setForbiddenStatus(keyspace + " keyspace is forbidden");
+ return;
+ }
+ if (!arePathParamsValid(keyspace, table, snapshot, component))
+ {
+ response.setBadRequestStatus("Invalid path params found");
+ return;
+ }
+
+ final Path path;
+ try
+ {
+ path = pathBuilder.build(keyspace, table, snapshot, component);
+ }
+ catch (FileNotFoundException e)
+ {
+ response.setNotFoundStatus(e.getMessage());
+ return;
+ }
+ final File file = path.toFile();
+ final Range r;
+ try
+ {
+ r = parseRangeHeader(range, file.length());
+ }
+ catch (Exception e)
+ {
+ response.setRangeNotSatisfiable(e.getMessage());
+ return;
+ }
+ fileStreamer.stream(response, file, r);
+ }
+
+ private boolean arePathParamsValid(String keyspace, String table, String snapshot, String component)
+ {
+ return REGEX_DIR.matcher(keyspace).matches() && REGEX_DIR.matcher(table).matches()
+ && REGEX_DIR.matcher(snapshot).matches() && REGEX_COMPONENT.matcher(component).matches();
+ }
+
+ private Range parseRangeHeader(String rangeHeader, long fileSize)
+ {
+ final Range fileRange = new Range(0, fileSize - 1, fileSize);
+ // sidecar does not support multiple ranges as of now
+ final Range headerRange = Range.parseHeader(rangeHeader, fileSize);
+ return fileRange.intersect(headerRange);
+ }
+}
diff --git a/src/main/java/org/apache/cassandra/sidecar/utils/CachedFilePathBuilder.java b/src/main/java/org/apache/cassandra/sidecar/utils/CachedFilePathBuilder.java
new file mode 100644
index 0000000..0c3aabd
--- /dev/null
+++ b/src/main/java/org/apache/cassandra/sidecar/utils/CachedFilePathBuilder.java
@@ -0,0 +1,219 @@
+package org.apache.cassandra.sidecar.utils;
+
+import java.io.FileNotFoundException;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.security.KeyException;
+import java.util.Collection;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+
+import com.google.common.cache.CacheBuilder;
+import com.google.common.cache.CacheLoader;
+import com.google.common.cache.LoadingCache;
+import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+
+/**
+ * Path builder that caches intermediate paths
+ */
+@Singleton
+public class CachedFilePathBuilder extends FilePathBuilder
+{
+ private static final Logger logger = LoggerFactory.getLogger(CachedFilePathBuilder.class);
+ private final CacheLoader<Key, String> loader = new PathCacheLoader();
+ private final LoadingCache<Key, String> sstableCache = getCacheBuilder();
+ private final LoadingCache<Key, String> snapshotCache = getCacheBuilder();
+ private final LoadingCache<Key, String> tableCache = getCacheBuilder();
+ private final LoadingCache<Key, String> keyspaceCache = getCacheBuilder();
+
+ private LoadingCache<Key, String> getCacheBuilder()
+ {
+ return CacheBuilder.newBuilder().maximumSize(10000).refreshAfterWrite(5, TimeUnit.MINUTES).build(loader);
+ }
+
+ @Inject
+ public CachedFilePathBuilder(final Collection<String> dataDirs)
+ {
+ super(dataDirs);
+ }
+
+ public Path build(String keyspace, String table, String snapshot, String component) throws FileNotFoundException
+ {
+ try
+ {
+ return Paths.get(sstableCache.get(new Key.Builder().setKeyspace(keyspace).setTable(table)
+ .setSnapshot(snapshot).setComponent(component).build()));
+ }
+ catch (Throwable t)
+ {
+ if (ExceptionUtils.getRootCause(t) instanceof FileNotFoundException)
+ {
+ throw (FileNotFoundException) ExceptionUtils.getRootCause(t);
+ }
+ else
+ {
+ logger.error("Unexpected error while building path ", t);
+ throw new RuntimeException("Error loading value from path cache");
+ }
+ }
+ }
+
+ /**
+ * Cache Loader for guava cache storing path to files
+ */
+ public class PathCacheLoader extends CacheLoader<Key, String>
+ {
+ @Override
+ public String load(Key key) throws FileNotFoundException, KeyException, ExecutionException
+ {
+ switch (key.type())
+ {
+ case KEYSPACE_TABLE_SNAPSHOT_COMPONENT:
+ return addSSTableComponentToPath(key.component(), snapshotCache.get(new Key.Builder()
+ .setKeyspace(key.keyspace()).setTable(key.table()).setSnapshot(key.snapshot()).build()));
+ case KEYSPACE_TABLE_SNAPSHOT:
+ return addSnapshotToPath(key.snapshot(), tableCache.get(new Key.Builder()
+ .setKeyspace(key.keyspace()).setTable(key.table()).build()));
+ case KEYSPACE_TABLE:
+ return addTableToPath(key.table(), keyspaceCache.get(new Key.Builder().setKeyspace(key.keyspace())
+ .build()));
+ case JUST_KEYSPACE:
+ return addKeyspaceToPath(key.keyspace());
+ default:
+ throw new KeyException();
+ }
+ }
+ }
+
+ /**
+ * Key to retrieve path information from cache
+ */
+ public static class Key
+ {
+ private final String keyspace;
+ private final String table;
+ private final String snapshot;
+ private final String component;
+ private final KeyType type;
+
+ private Key(String keyspace, String table, String snapshot, String component, KeyType type)
+ {
+ this.keyspace = keyspace;
+ this.table = table;
+ this.snapshot = snapshot;
+ this.component = component;
+ this.type = type;
+ }
+
+ public String keyspace() throws KeyException
+ {
+ return Optional.ofNullable(keyspace).orElseThrow(KeyException::new);
+ }
+
+ public String table() throws KeyException
+ {
+ return Optional.ofNullable(table).orElseThrow(KeyException::new);
+ }
+
+ public String snapshot() throws KeyException
+ {
+ return Optional.ofNullable(snapshot).orElseThrow(KeyException::new);
+ }
+
+ public String component() throws KeyException
+ {
+ return Optional.ofNullable(component).orElseThrow(KeyException::new);
+ }
+
+ public KeyType type()
+ {
+ return type;
+ }
+
+ @Override
+ public boolean equals(Object o)
+ {
+ if (this == o)
+ {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass())
+ {
+ return false;
+ }
+ Key key = (Key) o;
+ return type == key.type &&
+ Objects.equals(keyspace, key.keyspace) &&
+ Objects.equals(table, key.table) &&
+ Objects.equals(snapshot, key.snapshot) &&
+ Objects.equals(component, key.component);
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Objects.hash(keyspace, table, snapshot, component, type);
+ }
+
+ /**
+ * Builder class for Key
+ */
+ public static class Builder
+ {
+ private String keyspace;
+ private String table;
+ private String snapshot;
+ private String component;
+ private int length;
+
+ public Builder setKeyspace(String keyspace)
+ {
+ length++;
+ this.keyspace = keyspace;
+ return this;
+ }
+
+ public Builder setTable(String table)
+ {
+ length++;
+ this.table = table;
+ return this;
+ }
+
+ public Builder setSnapshot(String snapshot)
+ {
+ length++;
+ this.snapshot = snapshot;
+ return this;
+ }
+
+ public Builder setComponent(String component)
+ {
+ length++;
+ this.component = component;
+ return this;
+ }
+
+ public CachedFilePathBuilder.Key build()
+ {
+ return new CachedFilePathBuilder.Key(keyspace, table, snapshot, component,
+ KeyType.values()[length - 1]);
+ }
+ }
+ }
+
+ /**
+ * Enum to hold types of keys created
+ */
+ public enum KeyType
+ {
+ JUST_KEYSPACE, KEYSPACE_TABLE, KEYSPACE_TABLE_SNAPSHOT, KEYSPACE_TABLE_SNAPSHOT_COMPONENT
+ }
+}
diff --git a/src/main/java/org/apache/cassandra/sidecar/utils/FilePathBuilder.java b/src/main/java/org/apache/cassandra/sidecar/utils/FilePathBuilder.java
new file mode 100644
index 0000000..334f58e
--- /dev/null
+++ b/src/main/java/org/apache/cassandra/sidecar/utils/FilePathBuilder.java
@@ -0,0 +1,145 @@
+package org.apache.cassandra.sidecar.utils;
+
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.nio.file.DirectoryStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.Collection;
+
+import javax.validation.constraints.NotNull;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Builds file path after verifying it exists
+ */
+public abstract class FilePathBuilder
+{
+ private static final Logger logger = LoggerFactory.getLogger(FilePathBuilder.class);
+ private static final String dataSubDir = "/data";
+ private final Collection<String> dataDirs;
+
+ public FilePathBuilder(@NotNull final Collection<String> dataDirs)
+ {
+ this.dataDirs = dataDirs;
+ }
+
+ public abstract Path build(String keyspace, String table, String snapshot, String component)
+ throws FileNotFoundException;
+
+ String addKeyspaceToPath(String keyspace) throws FileNotFoundException
+ {
+ for (String dir : dataDirs)
+ {
+ StringBuilder path = new StringBuilder(dir);
+ if (addFileToPathIfPresent(path, keyspace, true))
+ {
+ return path.toString();
+ }
+
+ if (dir.endsWith(dataSubDir))
+ {
+ continue;
+ }
+
+ // check in "data" sub directory
+ if (addFileToPathIfPresent(path.append(dataSubDir), keyspace, true))
+ {
+ return path.toString();
+ }
+ }
+ throw new FileNotFoundException("Keyspace " + keyspace + " does not exist");
+ }
+
+ String addTableToPath(String table, String path) throws FileNotFoundException
+ {
+ final StringBuilder modifiedPath = new StringBuilder(path);
+ if (addFileToPathIfPresent(modifiedPath, table, false))
+ {
+ return modifiedPath.toString();
+ }
+ throw new FileNotFoundException("Table " + table + " not found, path searched: " + path);
+ }
+
+ String addSnapshotToPath(String snapshot, String path) throws FileNotFoundException
+ {
+ final StringBuilder modifiedPath = new StringBuilder(path);
+ if (addFileToPathIfPresent(modifiedPath.append("/snapshots"), snapshot, true))
+ {
+ return modifiedPath.toString();
+ }
+ throw new FileNotFoundException("Snapshot " + snapshot + " not found, path searched: " + path);
+ }
+
+ String addSSTableComponentToPath(String component, String path) throws FileNotFoundException
+ {
+ final StringBuilder modifiedPath = new StringBuilder(path);
+ if (addFileToPathIfPresent(modifiedPath, component, true))
+ {
+ return modifiedPath.toString();
+ }
+ throw new FileNotFoundException("Component " + component + " not found, path searched: " + path);
+ }
+
+ private boolean addFileToPathIfPresent(StringBuilder path, String file, boolean checkEqual)
+ throws FileNotFoundException
+ {
+ final Path fileDir = Paths.get(path.toString());
+ if (!checkDirExists(fileDir))
+ {
+ throw new FileNotFoundException(fileDir + " directory empty or does not exist!");
+ }
+
+ try
+ {
+ Path finalPath = null;
+ try (final DirectoryStream<Path> dirEntries = Files.newDirectoryStream(fileDir))
+ {
+ for (Path entry : dirEntries)
+ {
+ final Path filePath = entry.getFileName();
+ if (filePath == null)
+ {
+ continue;
+ }
+ final String fileName = filePath.toString();
+ if (fileName.equals(file) || (!checkEqual && fileName.startsWith(file + "-")))
+ {
+ if (finalPath == null
+ || Files.getLastModifiedTime(entry).compareTo(Files.getLastModifiedTime(finalPath)) > 0)
+ {
+ finalPath = entry;
+ }
+ }
+ }
+ if (finalPath != null)
+ {
+ final Path finalFilePath = finalPath.getFileName();
+ if (finalFilePath == null)
+ {
+ return false;
+ }
+ final String finalFileName = finalFilePath.toString();
+ path.append('/').append(finalFileName);
+ return true;
+ }
+ }
+ }
+ catch (IOException e)
+ {
+ logger.error("Error listing files in path {}, could not add file {} to path", path, file, e);
+ throw new RuntimeException("Failed to list files in path " + path);
+ }
+ return false;
+ }
+
+ private boolean checkDirExists(final Path path)
+ {
+ final File file = new File(path.toString());
+ return file.exists() && file.isDirectory();
+ }
+}
diff --git a/src/main/java/org/apache/cassandra/sidecar/utils/FileStreamer.java b/src/main/java/org/apache/cassandra/sidecar/utils/FileStreamer.java
new file mode 100644
index 0000000..b90f231
--- /dev/null
+++ b/src/main/java/org/apache/cassandra/sidecar/utils/FileStreamer.java
@@ -0,0 +1,113 @@
+package org.apache.cassandra.sidecar.utils;
+
+import java.io.File;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import com.google.common.util.concurrent.SidecarRateLimiter;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+import org.apache.cassandra.sidecar.Configuration;
+import org.apache.cassandra.sidecar.models.HttpResponse;
+import org.apache.cassandra.sidecar.models.Range;
+
+import static java.util.concurrent.TimeUnit.MICROSECONDS;
+
+/**
+ * General handler for serving files
+ */
+@Singleton
+public class FileStreamer
+{
+ private static final Logger LOGGER = LoggerFactory.getLogger(FileStreamer.class);
+ private static final ScheduledExecutorService SCHEDULER = Executors.newScheduledThreadPool(1,
+ new ThreadFactoryBuilder().setNameFormat("acquirePermit").setDaemon(true).build());
+ private final Duration delay;
+ private final Duration timeout;
+
+ private final SidecarRateLimiter rateLimiter;
+
+ @Inject
+ public FileStreamer(Configuration config, SidecarRateLimiter rateLimiter)
+ {
+ this.rateLimiter = rateLimiter;
+ this.delay = Duration.ofSeconds(config.getThrottleDelayInSeconds());
+ this.timeout = Duration.ofSeconds(config.getThrottleTimeoutInSeconds());
+ }
+
+ public void stream(final HttpResponse resp, final File file)
+ {
+ stream(resp, file, new Range(0, file.length() - 1, file.length()));
+ }
+
+ public void stream(final HttpResponse resp, final File file, final Range range)
+ {
+ if (!file.exists() || !file.isFile())
+ {
+ resp.setNotFoundStatus("File does not exist or it is not a normal file");
+ return;
+ }
+ if (file.length() == 0)
+ {
+ resp.setBadRequestStatus("File is empty");
+ return;
+ }
+ acquireAndSend(resp, file, range);
+ }
+
+ private void acquireAndSend(HttpResponse response, File file, Range range)
+ {
+ acquireAndSend(response, file, range, Instant.now());
+ }
+
+ /**
+ * If permit becomes available within a short time, retry immediately
+ */
+ private void acquireAndSend(HttpResponse response, File file, Range range, Instant startTime)
+ {
+ while (!rateLimiter.tryAcquire())
+ {
+ if (checkRetriesExhausted(startTime))
+ {
+ LOGGER.error("Retries for acquiring permit exhausted!");
+ response.setTooManyRequestsStatus();
+ return;
+ }
+
+ final long microsToWait = rateLimiter.queryEarliestAvailable(0L);
+ if (microsToWait <= 0) // immediately retry
+ {
+ continue;
+ }
+
+ if (TimeUnit.MICROSECONDS.toNanos(microsToWait) >= delay.getNano())
+ {
+ response.setRetryAfterHeader(microsToWait);
+ }
+ else
+ {
+ retryStreaming(response, file, range, startTime, microsToWait);
+ }
+ return;
+ }
+ LOGGER.info("File {} streamed from path {}", file.getName(), file.getAbsolutePath());
+ response.sendFile(file, range);
+ }
+
+ private boolean checkRetriesExhausted(Instant startTime)
+ {
+ return startTime.plus(timeout).isBefore(Instant.now());
+ }
+
+ private void retryStreaming(HttpResponse response, File file, Range range, Instant startTime, long microsToSleep)
+ {
+ SCHEDULER.schedule(() -> acquireAndSend(response, file, range, startTime), microsToSleep, MICROSECONDS);
+ }
+}
diff --git a/src/test/java/org/apache/cassandra/sidecar/FilePathBuilderTest.java b/src/test/java/org/apache/cassandra/sidecar/FilePathBuilderTest.java
new file mode 100644
index 0000000..734ca22
--- /dev/null
+++ b/src/test/java/org/apache/cassandra/sidecar/FilePathBuilderTest.java
@@ -0,0 +1,126 @@
+package org.apache.cassandra.sidecar;
+
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.Collections;
+
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import com.google.inject.Guice;
+import com.google.inject.Injector;
+import com.google.inject.util.Modules;
+import org.apache.cassandra.sidecar.utils.CachedFilePathBuilder;
+import org.apache.cassandra.sidecar.utils.FilePathBuilder;
+import org.assertj.core.api.Assertions;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * FilePathBuilderTest
+ */
+public class FilePathBuilderTest
+{
+ private static final String expectedFilePath = "src/test/resources/data/TestKeyspace" +
+ "/TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b/snapshots/TestSnapshot" +
+ "/TestKeyspace-TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b-Data.db";
+ private static FilePathBuilder pathBuilder;
+
+ @BeforeAll
+ public static void setUp()
+ {
+ Injector injector = Guice.createInjector(Modules.override(new MainModule()).with(new TestModule()));
+ pathBuilder = injector.getInstance(FilePathBuilder.class);
+ }
+
+ @Test
+ public void testRoute() throws IOException
+ {
+ final String keyspace = "TestKeyspace";
+ final String table = "TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b";
+ final String snapshot = "TestSnapshot";
+ final String component = "TestKeyspace-TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b-Data.db";
+ Path filePath = pathBuilder.build(keyspace, table, snapshot, component);
+ assertEquals(expectedFilePath, filePath.toString());
+ }
+
+ @Test
+ public void testKeyspaceNotFound()
+ {
+ final String keyspace = "random";
+ final String table = "TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b";
+ final String snapshot = "TestSnapshot";
+ final String component = "TestKeyspace-TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b-Data.db";
+ FileNotFoundException thrownException = assertThrows(FileNotFoundException.class, () ->
+ {
+ pathBuilder.build(keyspace, table, snapshot, component);
+ });
+ String msg = "Keyspace random does not exist";
+ assertEquals(msg, thrownException.getMessage());
+ }
+
+ @Test
+ public void testTableNotFound()
+ {
+ final String keyspace = "TestKeyspace";
+ final String table = "random";
+ final String snapshot = "TestSnapshot";
+ final String component = "TestKeyspace-TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b-Data.db";
+ FileNotFoundException thrownException = assertThrows(FileNotFoundException.class, () ->
+ {
+ pathBuilder.build(keyspace, table, snapshot, component);
+ });
+ String msg = "Table random not found, path searched: src/test/resources/data/TestKeyspace";
+ assertEquals(msg, thrownException.getMessage());
+ }
+
+ @Test
+ public void testSnapshotNotFound()
+ {
+ final String keyspace = "TestKeyspace";
+ final String table = "TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b";
+ final String snapshot = "random";
+ final String component = "TestKeyspace-TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b-Data.db";
+ FileNotFoundException thrownException = assertThrows(FileNotFoundException.class, () ->
+ {
+ pathBuilder.build(keyspace, table, snapshot, component);
+ });
+ String msg = "Snapshot random not found, path searched: src/test/resources/data/TestKeyspace" +
+ "/TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b";
+ assertEquals(msg, thrownException.getMessage());
+ }
+
+ @Test
+ public void testPartialTableName() throws FileNotFoundException
+ {
+ final String keyspace = "TestKeyspace";
+ final String table = "TestTable";
+ final String snapshot = "TestSnapshot";
+ final String component = "TestKeyspace-TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b-Data.db";
+ Path filePath = pathBuilder.build(keyspace, table, snapshot, component);
+ assertEquals(expectedFilePath, filePath.toString());
+ }
+
+ @Test
+ public void testEmptyDataDir() throws IOException
+ {
+ String dataDir = new File("./").getCanonicalPath() + "/src/test/resources/instance";
+
+ final String keyspace = "TestKeyspace";
+ final String table = "TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b";
+ final String snapshot = "TestSnapshot";
+ final String component = "TestKeyspace-TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b-Data.db";
+
+ FilePathBuilder pathBuilder = new CachedFilePathBuilder(Collections.singletonList(dataDir));
+ FileNotFoundException thrownException = assertThrows(FileNotFoundException.class, () ->
+ {
+ pathBuilder.build(keyspace, table, snapshot, component);
+ });
+ String msg = "directory empty or does not exist!";
+ Assertions.assertThat(thrownException.getMessage()).contains(msg);
+ }
+}
diff --git a/src/test/java/org/apache/cassandra/sidecar/RangeTest.java b/src/test/java/org/apache/cassandra/sidecar/RangeTest.java
new file mode 100644
index 0000000..82c5d5e
--- /dev/null
+++ b/src/test/java/org/apache/cassandra/sidecar/RangeTest.java
@@ -0,0 +1,86 @@
+package org.apache.cassandra.sidecar;
+
+import org.junit.jupiter.api.Test;
+
+import org.apache.cassandra.sidecar.exceptions.RangeException;
+import org.apache.cassandra.sidecar.models.Range;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * RangeTest
+ */
+public class RangeTest
+{
+ @Test
+ public void testValidPartialRange()
+ {
+ final String rangeHeaderVal = "bytes=2-";
+ final Range range = Range.parseHeader(rangeHeaderVal, 5);
+ assertEquals(3, range.length());
+ assertEquals(2, range.start());
+ assertEquals(4, range.end());
+ assertTrue(range.isValidHttpRange());
+ }
+
+ @Test
+ public void testValidFullRange()
+ {
+ final String rangeHeaderVal = "bytes=0-100";
+ final Range range = Range.parseHeader(rangeHeaderVal, 500);
+ assertEquals(101, range.length());
+ assertEquals(0, range.start());
+ assertEquals(100, range.end());
+ assertTrue(range.isValidHttpRange());
+ }
+
+ @Test
+ public void testInvalidRangeFormat()
+ {
+ final String rangeVal = "2344--3432";
+ IllegalArgumentException thrownException = assertThrows(IllegalArgumentException.class, () ->
+ {
+ Range.parse(rangeVal);
+ });
+ String msg = "Supported Range formats are <start>-<end>, <start>-, -<suffix-length>";
+ assertEquals(msg, thrownException.getMessage());
+ }
+
+ @Test
+ public void testInvalidSuffixLength()
+ {
+ final String rangeVal = "-0";
+ IllegalArgumentException thrownException = assertThrows(IllegalArgumentException.class, () ->
+ {
+ Range.parse(rangeVal, Long.MAX_VALUE);
+ });
+ String msg = "Suffix length in -0 cannot be less than or equal to 0";
+ assertEquals(msg, thrownException.getMessage());
+ }
+
+ @Test
+ public void testInvalidRangeBoundary()
+ {
+ final String rangeVal = "9-2";
+ RangeException thrownException = assertThrows(RangeException.class, () ->
+ {
+ Range.parse(rangeVal);
+ });
+ String msg = "Range does not satisfy boundary requirements";
+ assertEquals(msg, thrownException.getMessage());
+ }
+
+ @Test
+ public void testWrongRangeUnitUsed()
+ {
+ final String rangeVal = "bits=0-";
+ UnsupportedOperationException thrownException = assertThrows(UnsupportedOperationException.class, () ->
+ {
+ Range.parseHeader(rangeVal, 5);
+ });
+ String msg = "Unsupported range unit only bytes are allowed";
+ assertEquals(msg, thrownException.getMessage());
+ }
+}
diff --git a/src/test/java/org/apache/cassandra/sidecar/StreamSSTableComponentTest.java b/src/test/java/org/apache/cassandra/sidecar/StreamSSTableComponentTest.java
new file mode 100644
index 0000000..b6cc605
--- /dev/null
+++ b/src/test/java/org/apache/cassandra/sidecar/StreamSSTableComponentTest.java
@@ -0,0 +1,308 @@
+package org.apache.cassandra.sidecar;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.inject.Guice;
+import com.google.inject.Injector;
+import com.google.inject.util.Modules;
+import io.vertx.core.Vertx;
+import io.vertx.core.http.HttpServer;
+import io.vertx.ext.web.client.WebClient;
+import io.vertx.ext.web.codec.BodyCodec;
+import io.vertx.junit5.VertxExtension;
+import io.vertx.junit5.VertxTestContext;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Test for StreamSSTableComponent
+ */
+@ExtendWith(VertxExtension.class)
+public class StreamSSTableComponentTest
+{
+ private static final Logger logger = LoggerFactory.getLogger(StreamSSTableComponentTest.class);
+ private Vertx vertx;
+ private HttpServer server;
+ private Configuration config;
+
+ @BeforeEach
+ void setUp() throws InterruptedException
+ {
+ Injector injector = Guice.createInjector(Modules.override(new MainModule()).with(new TestModule()));
+ server = injector.getInstance(HttpServer.class);
+ vertx = injector.getInstance(Vertx.class);
+ config = injector.getInstance(Configuration.class);
+
+ VertxTestContext context = new VertxTestContext();
+ server.listen(config.getPort(), context.completing());
+
+ context.awaitCompletion(5, TimeUnit.SECONDS);
+ }
+
+ @AfterEach
+ void tearDown() throws InterruptedException
+ {
+ final CountDownLatch closeLatch = new CountDownLatch(1);
+ server.close(res -> closeLatch.countDown());
+ vertx.close();
+ if (closeLatch.await(60, TimeUnit.SECONDS))
+ logger.info("Close event received before timeout.");
+ else
+ logger.error("Close event timed out.");
+ }
+
+ @Test
+ void testRoute(VertxTestContext context)
+ {
+ WebClient client = WebClient.create(vertx);
+ String testRoute = "/keyspace/TestKeyspace/table/TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b/snapshot" +
+ "/TestSnapshot/component/TestKeyspace-TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b-Data.db";
+ client.get(config.getPort(), "localhost", "/api/v1/stream" + testRoute)
+ .as(BodyCodec.buffer())
+ .send(context.succeeding(response -> context.verify(() ->
+ {
+ assertEquals(200, response.statusCode());
+ assertEquals("data", response.bodyAsString());
+ context.completeNow();
+ })));
+ }
+
+ @Test
+ void testKeyspaceNotFound(VertxTestContext context)
+ {
+ WebClient client = WebClient.create(vertx);
+ String testRoute = "/keyspace/random/table/TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b/snapshot" +
+ "/TestSnapshot/component/TestKeyspace-TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b-Data.db";
+ client.get(config.getPort(), config.getHost(), "/api/v1/stream" + testRoute)
+ .send(context.succeeding(response -> context.verify(() ->
+ {
+ assertEquals(404, response.statusCode());
+ context.completeNow();
+ })));
+ }
+
+ @Test
+ void testSnapshotNotFound(VertxTestContext context)
+ {
+ WebClient client = WebClient.create(vertx);
+ String testRoute = "/keyspace/TestKeyspace/table/TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b/snapshot" +
+ "/random/component/TestKeyspace-TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b-Data.db";
+ client.get(config.getPort(), config.getHost(), "/api/v1/stream" + testRoute)
+ .send(context.succeeding(response -> context.verify(() ->
+ {
+ assertEquals(404, response.statusCode());
+ context.completeNow();
+ })));
+ }
+
+ @Test
+ void testForbiddenKeyspace(VertxTestContext context)
+ {
+ WebClient client = WebClient.create(vertx);
+ String testRoute = "/keyspace/system/table/TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b/snapshot" +
+ "/TestSnapshot/component/TestKeyspace-TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b-Data.db";
+ client.get(config.getPort(), config.getHost(), "/api/v1/stream" + testRoute)
+ .send(context.succeeding(response -> context.verify(() ->
+ {
+ assertEquals(403, response.statusCode());
+ assertEquals("system keyspace is forbidden", response.statusMessage());
+ context.completeNow();
+ })));
+ }
+
+ @Test
+ void testIncorrectKeyspaceFormat(VertxTestContext context)
+ {
+ WebClient client = WebClient.create(vertx);
+ String testRoute = "/keyspace/k*s/table/TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b/snapshot" +
+ "/TestSnapshot/component/TestKeyspace-TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b-Data.db";
+ client.get(config.getPort(), config.getHost(), "/api/v1/stream" + testRoute)
+ .send(context.succeeding(response -> context.verify(() ->
+ {
+ assertEquals(400, response.statusCode());
+ assertEquals("Invalid path params found", response.statusMessage());
+ context.completeNow();
+ })));
+ }
+
+ @Test
+ void testIncorrectComponentFormat(VertxTestContext context)
+ {
+ WebClient client = WebClient.create(vertx);
+ String testRoute = "/keyspace/TestKeyspace/table/TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b/snapshot" +
+ "/TestSnapshot/component/TestKeyspace-TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b-Data...db";
+ client.get(config.getPort(), config.getHost(), "/api/v1/stream" + testRoute)
+ .send(context.succeeding(response -> context.verify(() ->
+ {
+ assertEquals(400, response.statusCode());
+ assertEquals("Invalid path params found", response.statusMessage());
+ context.completeNow();
+ })));
+ }
+
+ @Test
+ void testAccessDeniedToCertainComponents(VertxTestContext context)
+ {
+ WebClient client = WebClient.create(vertx);
+ String testRoute = "/keyspace/TestKeyspace/table/TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b/snapshot" +
+ "/TestSnapshot/component/TestKeyspace-TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b-Digest.crc32d";
+ client.get(config.getPort(), config.getHost(), "/api/v1/stream" + testRoute)
+ .send(context.succeeding(response -> context.verify(() ->
+ {
+ assertEquals(400, response.statusCode());
+ assertEquals("Invalid path params found", response.statusMessage());
+ context.completeNow();
+ })));
+ }
+
+ @Test
+ void testPartialTableName(VertxTestContext context)
+ {
+ WebClient client = WebClient.create(vertx);
+ String testRoute = "/keyspace/TestKeyspace/table/TestTable/snapshot/TestSnapshot/component" +
+ "/TestKeyspace-TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b-Data.db";
+ client.get(config.getPort(), config.getHost(), "/api/v1/stream" + testRoute)
+ .putHeader("Range", "bytes=0-")
+ .as(BodyCodec.buffer())
+ .send(context.succeeding(response -> context.verify(() ->
+ {
+ assertEquals(200, response.statusCode());
+ assertEquals("data", response.bodyAsString());
+ context.completeNow();
+ })));
+ }
+
+ @Test
+ void testInvalidRange(VertxTestContext context)
+ {
+ WebClient client = WebClient.create(vertx);
+ String testRoute = "/keyspace/TestKeyspace/table/TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b/snapshot" +
+ "/TestSnapshot/component/TestKeyspace-TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b-Data.db";
+ client.get(config.getPort(), config.getHost(), "/api/v1/stream" + testRoute)
+ .putHeader("Range", "bytes=4-3")
+ .send(context.succeeding(response -> context.verify(() ->
+ {
+ assertEquals(416, response.statusCode());
+ context.completeNow();
+ })));
+ }
+
+ @Test
+ void testRangeExceeds(VertxTestContext context)
+ {
+ WebClient client = WebClient.create(vertx);
+ String testRoute = "/keyspace/TestKeyspace/table/TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b/snapshot" +
+ "/TestSnapshot/component/TestKeyspace-TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b-Data.db";
+ client.get(config.getPort(), config.getHost(), "/api/v1/stream" + testRoute)
+ .putHeader("Range", "bytes=5-9")
+ .send(context.succeeding(response -> context.verify(() ->
+ {
+ assertEquals(416, response.statusCode());
+ context.completeNow();
+ })));
+ }
+
+ @Test
+ void testPartialRangeExceeds(VertxTestContext context)
+ {
+ WebClient client = WebClient.create(vertx);
+ String testRoute = "/keyspace/TestKeyspace/table/TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b/snapshot" +
+ "/TestSnapshot/component/TestKeyspace-TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b-Data.db";
+ client.get(config.getPort(), config.getHost(), "/api/v1/stream" + testRoute)
+ .putHeader("Range", "bytes=5-")
+ .send(context.succeeding(response -> context.verify(() ->
+ {
+ assertEquals(416, response.statusCode());
+ context.completeNow();
+ })));
+ }
+
+ @Test
+ void testRangeBoundaryExceeds(VertxTestContext context)
+ {
+ WebClient client = WebClient.create(vertx);
+ String testRoute = "/keyspace/TestKeyspace/table/TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b/snapshot" +
+ "/TestSnapshot/component/TestKeyspace-TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b-Data.db";
+ client.get(config.getPort(), config.getHost(), "/api/v1/stream" + testRoute)
+ .putHeader("Range", "bytes=0-999999")
+ .as(BodyCodec.buffer())
+ .send(context.succeeding(response -> context.verify(() ->
+ {
+ assertEquals(200, response.statusCode());
+ assertEquals("data", response.bodyAsString());
+ context.completeNow();
+ })));
+ }
+
+ @Test
+ void testPartialRangeStreamed(VertxTestContext context)
+ {
+ WebClient client = WebClient.create(vertx);
+ String testRoute = "/keyspace/TestKeyspace/table/TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b/snapshot" +
+ "/TestSnapshot/component/TestKeyspace-TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b-Data.db";
+ client.get(config.getPort(), config.getHost(), "/api/v1/stream" + testRoute)
+ .putHeader("Range", "bytes=0-2") // 3 bytes streamed
+ .as(BodyCodec.buffer())
+ .send(context.succeeding(response -> context.verify(() ->
+ {
+ assertEquals(206, response.statusCode());
+ assertEquals("dat", response.bodyAsString());
+ context.completeNow();
+ })));
+ }
+
+ @Test
+ void testSuffixRange(VertxTestContext context)
+ {
+ WebClient client = WebClient.create(vertx);
+ String testRoute = "/keyspace/TestKeyspace/table/TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b/snapshot" +
+ "/TestSnapshot/component/TestKeyspace-TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b-Data.db";
+ client.get(config.getPort(), config.getHost(), "/api/v1/stream" + testRoute)
+ .putHeader("Range", "bytes=-2") // last 2 bytes streamed
+ .as(BodyCodec.buffer())
+ .send(context.succeeding(response -> context.verify(() ->
+ {
+ assertEquals(206, response.statusCode());
+ assertEquals("ta", response.bodyAsString());
+ context.completeNow();
+ })));
+ }
+
+ @Test
+ void testSuffixRangeExceeds(VertxTestContext context)
+ {
+ WebClient client = WebClient.create(vertx);
+ String testRoute = "/keyspace/TestKeyspace/table/TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b/snapshot" +
+ "/TestSnapshot/component/TestKeyspace-TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b-Data.db";
+ client.get(config.getPort(), config.getHost(), "/api/v1/stream" + testRoute)
+ .putHeader("Range", "bytes=-5")
+ .send(context.succeeding(response -> context.verify(() ->
+ {
+ assertEquals(416, response.statusCode());
+ context.completeNow();
+ })));
+ }
+
+ @Test
+ void testInvalidRangeUnit(VertxTestContext context)
+ {
+ WebClient client = WebClient.create(vertx);
+ String testRoute = "/keyspace/TestKeyspace/table/TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b/snapshot" +
+ "/TestSnapshot/component/TestKeyspace-TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b-Data.db";
+ client.get(config.getPort(), config.getHost(), "/api/v1/stream" + testRoute)
+ .putHeader("Range", "bits=0-2")
+ .send(context.succeeding(response -> context.verify(() ->
+ {
+ assertEquals(416, response.statusCode());
+ context.completeNow();
+ })));
+ }
+}
diff --git a/src/test/java/org/apache/cassandra/sidecar/TestModule.java b/src/test/java/org/apache/cassandra/sidecar/TestModule.java
index 67c28d0..755ad4a 100644
--- a/src/test/java/org/apache/cassandra/sidecar/TestModule.java
+++ b/src/test/java/org/apache/cassandra/sidecar/TestModule.java
@@ -18,13 +18,14 @@
package org.apache.cassandra.sidecar;
+import java.util.Collections;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.google.inject.AbstractModule;
import com.google.inject.Provides;
import com.google.inject.Singleton;
-
import org.apache.cassandra.sidecar.common.CassandraAdapterDelegate;
import org.apache.cassandra.sidecar.common.CassandraVersionProvider;
import org.apache.cassandra.sidecar.common.MockCassandraFactory;
@@ -66,10 +67,14 @@ public class TestModule extends AbstractModule
return new Configuration.Builder()
.setCassandraHost("INVALID_FOR_TEST")
.setCassandraPort(0)
+ .setCassandraDataDirs(Collections.singletonList("src/test/resources/data"))
.setHost("127.0.0.1")
.setPort(6475)
.setHealthCheckFrequency(1000)
.setSslEnabled(false)
+ .setRateLimitStreamRequestsPerSecond(1)
+ .setThrottleDelayInSeconds(5)
+ .setThrottleTimeoutInSeconds(10)
.build();
}
@@ -85,5 +90,4 @@ public class TestModule extends AbstractModule
builder.add(new MockCassandraFactory());
return builder.build();
}
-
}
diff --git a/src/test/java/org/apache/cassandra/sidecar/TestSslModule.java b/src/test/java/org/apache/cassandra/sidecar/TestSslModule.java
index 883e04e..4ecb9f2 100644
--- a/src/test/java/org/apache/cassandra/sidecar/TestSslModule.java
+++ b/src/test/java/org/apache/cassandra/sidecar/TestSslModule.java
@@ -19,6 +19,7 @@
package org.apache.cassandra.sidecar;
import java.io.File;
+import java.util.Collections;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -52,6 +53,7 @@ public class TestSslModule extends TestModule
return new Configuration.Builder()
.setCassandraHost("INVALID_FOR_TEST")
.setCassandraPort(0)
+ .setCassandraDataDirs(Collections.singletonList("src/test/resources/data"))
.setHost("127.0.0.1")
.setPort(6475)
.setHealthCheckFrequency(1000)
@@ -60,6 +62,7 @@ public class TestSslModule extends TestModule
.setTrustStorePath(trustStorePath)
.setTrustStorePassword(trustStorePassword)
.setSslEnabled(true)
+ .setRateLimitStreamRequestsPerSecond(1)
.build();
}
}
diff --git a/src/test/java/org/apache/cassandra/sidecar/ThrottleTest.java b/src/test/java/org/apache/cassandra/sidecar/ThrottleTest.java
new file mode 100644
index 0000000..d77c8e5
--- /dev/null
+++ b/src/test/java/org/apache/cassandra/sidecar/ThrottleTest.java
@@ -0,0 +1,109 @@
+package org.apache.cassandra.sidecar;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.inject.Guice;
+import com.google.inject.Injector;
+import com.google.inject.util.Modules;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.vertx.core.Vertx;
+import io.vertx.core.http.HttpServer;
+import io.vertx.ext.web.client.HttpResponse;
+import io.vertx.ext.web.client.WebClient;
+import io.vertx.ext.web.codec.BodyCodec;
+import io.vertx.junit5.VertxExtension;
+import io.vertx.junit5.VertxTestContext;
+
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * Test rate limiting stream requests
+ */
+@ExtendWith(VertxExtension.class)
+public class ThrottleTest
+{
+ private static final Logger logger = LoggerFactory.getLogger(ThrottleTest.class);
+ private Vertx vertx;
+ private HttpServer server;
+ private Configuration config;
+
+ @BeforeEach
+ void setUp() throws InterruptedException
+ {
+
+ Injector injector = Guice.createInjector(Modules.override(new MainModule()).with(new TestModule()));
+ server = injector.getInstance(HttpServer.class);
+ vertx = injector.getInstance(Vertx.class);
+ config = injector.getInstance(Configuration.class);
+
+ VertxTestContext context = new VertxTestContext();
+ server.listen(config.getPort(), context.completing());
+
+ context.awaitCompletion(5, SECONDS);
+ }
+
+ @AfterEach
+ void tearDown() throws InterruptedException
+ {
+ final CountDownLatch closeLatch = new CountDownLatch(1);
+ server.close(res -> closeLatch.countDown());
+ vertx.close();
+ if (closeLatch.await(60, SECONDS))
+ logger.info("Close event received before timeout.");
+ else
+ logger.error("Close event timed out.");
+ }
+
+ @Test
+ void testStreamRequestsThrottled() throws Exception
+ {
+ String testRoute = "/keyspace/TestKeyspace/table/TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b/snapshot" +
+ "/TestSnapshot/component/TestKeyspace-TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b-Data.db";
+
+ for (int i = 0; i < 20; i++)
+ {
+ unblockingClientRequest(testRoute);
+ }
+
+ HttpResponse response = blockingClientRequest(testRoute);
+ assertEquals(HttpResponseStatus.TOO_MANY_REQUESTS.code(), response.statusCode());
+
+ long secsToWait = Long.parseLong(response.getHeader("Retry-After"));
+ Thread.sleep(SECONDS.toMillis(secsToWait));
+
+ HttpResponse finalResp = blockingClientRequest(testRoute);
+ assertEquals(HttpResponseStatus.OK.code(), finalResp.statusCode());
+ assertEquals("data", finalResp.bodyAsString());
+ }
+
+ private void unblockingClientRequest(String route)
+ {
+ WebClient client = WebClient.create(vertx);
+ client.get(config.getPort(), "localhost", "/api/v1/stream" + route)
+ .as(BodyCodec.buffer())
+ .send(resp ->
+ {
+ // do nothing
+ });
+ }
+
+ private HttpResponse blockingClientRequest(String route) throws ExecutionException, InterruptedException
+ {
+ WebClient client = WebClient.create(vertx);
+ CompletableFuture<HttpResponse> future = new CompletableFuture<>();
+ client.get(config.getPort(), "localhost", "/api/v1/stream" + route)
+ .as(BodyCodec.buffer())
+ .send(resp -> future.complete(resp.result()));
+ return future.get();
+ }
+}
diff --git a/src/test/resources/data/TestKeyspace/TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b/snapshots/TestSnapshot/TestKeyspace-TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b-Data.db b/src/test/resources/data/TestKeyspace/TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b/snapshots/TestSnapshot/TestKeyspace-TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b-Data.db
new file mode 100644
index 0000000..6320cd2
--- /dev/null
+++ b/src/test/resources/data/TestKeyspace/TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b/snapshots/TestSnapshot/TestKeyspace-TestTable-54ea95ce-bba2-4e0a-a9be-e428e5d7160b-Data.db
@@ -0,0 +1 @@
+data
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org
[cassandra-sidecar] 02/02: Avoid having sidecar's health response
code depend on Cassandra's health information
Posted by yc...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ycai pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra-sidecar.git
commit 4d88af6eacd8aaf50fb04d6889c8f8c395a4ff45
Author: Saranya Krishnakumar <sa...@apple.com>
AuthorDate: Wed Sep 22 15:50:39 2021 -0700
Avoid having sidecar's health response code depend on Cassandra's health information
patch by Saranya Krishnakumar; reviewed by Yifan Cai, Dinesh Joshi for CASSANDRASC-29
---
.../org/apache/cassandra/sidecar/MainModule.java | 5 +++-
...lthService.java => CassandraHealthService.java} | 8 +++----
.../cassandra/sidecar/routes/HealthService.java | 28 +++++-----------------
.../sidecar/AbstractHealthServiceTest.java | 28 ++++++++++++++++++----
.../cassandra/sidecar/FilePathBuilderTest.java | 1 -
.../org/apache/cassandra/sidecar/TestModule.java | 9 -------
6 files changed, 37 insertions(+), 42 deletions(-)
diff --git a/src/main/java/org/apache/cassandra/sidecar/MainModule.java b/src/main/java/org/apache/cassandra/sidecar/MainModule.java
index 223ab97..701900c 100644
--- a/src/main/java/org/apache/cassandra/sidecar/MainModule.java
+++ b/src/main/java/org/apache/cassandra/sidecar/MainModule.java
@@ -45,6 +45,7 @@ import org.apache.cassandra.sidecar.cassandra40.Cassandra40Factory;
import org.apache.cassandra.sidecar.common.CQLSession;
import org.apache.cassandra.sidecar.common.CassandraAdapterDelegate;
import org.apache.cassandra.sidecar.common.CassandraVersionProvider;
+import org.apache.cassandra.sidecar.routes.CassandraHealthService;
import org.apache.cassandra.sidecar.routes.HealthService;
import org.apache.cassandra.sidecar.routes.StreamSSTableComponent;
import org.apache.cassandra.sidecar.routes.SwaggerOpenApiResource;
@@ -102,7 +103,8 @@ public class MainModule extends AbstractModule
@Singleton
private VertxRequestHandler configureServices(Vertx vertx,
HealthService healthService,
- StreamSSTableComponent ssTableComponent)
+ StreamSSTableComponent ssTableComponent,
+ CassandraHealthService cassandraHealthService)
{
VertxResteasyDeployment deployment = new VertxResteasyDeployment();
deployment.start();
@@ -111,6 +113,7 @@ public class MainModule extends AbstractModule
r.addPerInstanceResource(SwaggerOpenApiResource.class);
r.addSingletonResource(healthService);
r.addSingletonResource(ssTableComponent);
+ r.addSingletonResource(cassandraHealthService);
return new VertxRequestHandler(vertx, deployment);
}
diff --git a/src/main/java/org/apache/cassandra/sidecar/routes/HealthService.java b/src/main/java/org/apache/cassandra/sidecar/routes/CassandraHealthService.java
similarity index 93%
copy from src/main/java/org/apache/cassandra/sidecar/routes/HealthService.java
copy to src/main/java/org/apache/cassandra/sidecar/routes/CassandraHealthService.java
index bd15fde..cc17ef8 100644
--- a/src/main/java/org/apache/cassandra/sidecar/routes/HealthService.java
+++ b/src/main/java/org/apache/cassandra/sidecar/routes/CassandraHealthService.java
@@ -40,14 +40,14 @@ import org.apache.cassandra.sidecar.common.CassandraAdapterDelegate;
* Provides a simple REST endpoint to determine if a node is available
*/
@Singleton
-@Path("/api/v1/__health")
-public class HealthService
+@Path("/api/v1/cassandra/__health")
+public class CassandraHealthService
{
private static final Logger logger = LoggerFactory.getLogger(HealthService.class);
private final CassandraAdapterDelegate cassandra;
@Inject
- public HealthService(CassandraAdapterDelegate cassandra)
+ public CassandraHealthService(CassandraAdapterDelegate cassandra)
{
this.cassandra = cassandra;
}
@@ -60,7 +60,7 @@ public class HealthService
})
@Produces(MediaType.APPLICATION_JSON)
@GET
- public Response doGet()
+ public Response getCassandraHealth()
{
Boolean up = cassandra.isUp();
int status = up ? HttpResponseStatus.OK.code() : HttpResponseStatus.SERVICE_UNAVAILABLE.code();
diff --git a/src/main/java/org/apache/cassandra/sidecar/routes/HealthService.java b/src/main/java/org/apache/cassandra/sidecar/routes/HealthService.java
index bd15fde..dd0a46a 100644
--- a/src/main/java/org/apache/cassandra/sidecar/routes/HealthService.java
+++ b/src/main/java/org/apache/cassandra/sidecar/routes/HealthService.java
@@ -26,45 +26,29 @@ import javax.ws.rs.core.Response;
import com.google.common.collect.ImmutableMap;
-import com.google.inject.Inject;
import com.google.inject.Singleton;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.swagger.v3.oas.annotations.Operation;
-import io.swagger.v3.oas.annotations.responses.ApiResponse;
import io.vertx.core.json.Json;
import io.vertx.core.logging.Logger;
import io.vertx.core.logging.LoggerFactory;
-import org.apache.cassandra.sidecar.common.CassandraAdapterDelegate;
/**
- * Provides a simple REST endpoint to determine if a node is available
+ * Provides a simple REST endpoint to determine if Sidecar is available
*/
@Singleton
@Path("/api/v1/__health")
public class HealthService
{
private static final Logger logger = LoggerFactory.getLogger(HealthService.class);
- private final CassandraAdapterDelegate cassandra;
- @Inject
- public HealthService(CassandraAdapterDelegate cassandra)
- {
- this.cassandra = cassandra;
- }
-
- @Operation(summary = "Health Check for Cassandra's status",
- description = "Returns HTTP 200 if Cassandra is available, 503 otherwise",
- responses = {
- @ApiResponse(responseCode = "200", description = "Cassandra is available"),
- @ApiResponse(responseCode = "503", description = "Cassandra is not available")
- })
+ @Operation(summary = "Health Check for Sidecar's status",
+ description = "Returns HTTP 200 if Sidecar is available")
@Produces(MediaType.APPLICATION_JSON)
@GET
- public Response doGet()
+ public Response getSidecarHealth()
{
- Boolean up = cassandra.isUp();
- int status = up ? HttpResponseStatus.OK.code() : HttpResponseStatus.SERVICE_UNAVAILABLE.code();
- return Response.status(status).entity(Json.encode(ImmutableMap.of("status", up ?
- "OK" : "NOT_OK"))).build();
+ return Response.status(HttpResponseStatus.OK.code()).entity(Json.encode(ImmutableMap.of("status", "OK")))
+ .build();
}
}
diff --git a/src/test/java/org/apache/cassandra/sidecar/AbstractHealthServiceTest.java b/src/test/java/org/apache/cassandra/sidecar/AbstractHealthServiceTest.java
index 946be3c..9b19b8f 100644
--- a/src/test/java/org/apache/cassandra/sidecar/AbstractHealthServiceTest.java
+++ b/src/test/java/org/apache/cassandra/sidecar/AbstractHealthServiceTest.java
@@ -96,11 +96,10 @@ public abstract class AbstractHealthServiceTest
logger.error("Close event timed out.");
}
- @DisplayName("Should return HTTP 200 OK when check=True")
+ @DisplayName("Should return HTTP 200 OK if sidecar server is running")
@Test
- public void testHealthCheckReturns200OK(VertxTestContext testContext)
+ public void testSidecarHealthCheckReturnsOK(VertxTestContext testContext)
{
- when(cassandra.isUp()).thenReturn(true);
WebClient client = getClient();
client.get(config.getPort(), "localhost", "/api/v1/__health")
@@ -109,6 +108,7 @@ public abstract class AbstractHealthServiceTest
.send(testContext.succeeding(response -> testContext.verify(() ->
{
Assert.assertEquals(200, response.statusCode());
+ Assert.assertEquals("{\"status\":\"OK\"}", response.body());
testContext.completeNow();
})));
}
@@ -129,20 +129,38 @@ public abstract class AbstractHealthServiceTest
return options;
}
+ @DisplayName("Should return HTTP 200 OK when check=True")
+ @Test
+ public void testHealthCheckReturns200OK(VertxTestContext testContext)
+ {
+ when(cassandra.isUp()).thenReturn(true);
+ WebClient client = getClient();
+
+ client.get(config.getPort(), "localhost", "/api/v1/cassandra/__health")
+ .as(BodyCodec.string())
+ .ssl(isSslEnabled())
+ .send(testContext.succeeding(response -> testContext.verify(() ->
+ {
+ Assert.assertEquals(200, response.statusCode());
+ Assert.assertEquals("{\"status\":\"OK\"}", response.body());
+ testContext.completeNow();
+ })));
+ }
+
@DisplayName("Should return HTTP 503 Failure when check=False")
@Test
public void testHealthCheckReturns503Failure(VertxTestContext testContext)
{
-
when(cassandra.isUp()).thenReturn(false);
WebClient client = getClient();
- client.get(config.getPort(), "localhost", "/api/v1/__health")
+ client.get(config.getPort(), "localhost", "/api/v1/cassandra/__health")
.as(BodyCodec.string())
.ssl(isSslEnabled())
.send(testContext.succeeding(response -> testContext.verify(() ->
{
Assert.assertEquals(503, response.statusCode());
+ Assert.assertEquals("{\"status\":\"NOT_OK\"}", response.body());
testContext.completeNow();
})));
}
diff --git a/src/test/java/org/apache/cassandra/sidecar/FilePathBuilderTest.java b/src/test/java/org/apache/cassandra/sidecar/FilePathBuilderTest.java
index 734ca22..2f37749 100644
--- a/src/test/java/org/apache/cassandra/sidecar/FilePathBuilderTest.java
+++ b/src/test/java/org/apache/cassandra/sidecar/FilePathBuilderTest.java
@@ -18,7 +18,6 @@ import org.assertj.core.api.Assertions;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
-import static org.junit.jupiter.api.Assertions.assertTrue;
/**
* FilePathBuilderTest
diff --git a/src/test/java/org/apache/cassandra/sidecar/TestModule.java b/src/test/java/org/apache/cassandra/sidecar/TestModule.java
index 755ad4a..4cf7d9f 100644
--- a/src/test/java/org/apache/cassandra/sidecar/TestModule.java
+++ b/src/test/java/org/apache/cassandra/sidecar/TestModule.java
@@ -29,7 +29,6 @@ import com.google.inject.Singleton;
import org.apache.cassandra.sidecar.common.CassandraAdapterDelegate;
import org.apache.cassandra.sidecar.common.CassandraVersionProvider;
import org.apache.cassandra.sidecar.common.MockCassandraFactory;
-import org.apache.cassandra.sidecar.routes.HealthService;
import static org.mockito.Mockito.mock;
@@ -47,14 +46,6 @@ public class TestModule extends AbstractModule
return mock(CassandraAdapterDelegate.class);
}
- @Singleton
- @Provides
- public HealthService healthService(CassandraAdapterDelegate delegate)
- {
- return new HealthService(delegate);
- }
-
-
@Provides
@Singleton
public Configuration configuration()
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org