Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Adds "proxy" functionality to s3ninja. #197

Merged
merged 6 commits into from Apr 30, 2022
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
1 change: 0 additions & 1 deletion pom.xml
Expand Up @@ -56,7 +56,6 @@
<groupId>com.amazonaws</groupId>
<artifactId>aws-java-sdk-s3</artifactId>
<version>1.11.1034</version>
<scope>test</scope>
</dependency>
</dependencies>
</project>
150 changes: 150 additions & 0 deletions src/main/java/ninja/AwsUpstream.java
@@ -0,0 +1,150 @@
package ninja;

import com.amazonaws.ClientConfiguration;
import com.amazonaws.HttpMethod;
import com.amazonaws.auth.AWSStaticCredentialsProvider;
import com.amazonaws.auth.BasicAWSCredentials;
import com.amazonaws.client.builder.AwsClientBuilder;
import com.amazonaws.services.s3.AmazonS3;
import com.amazonaws.services.s3.AmazonS3ClientBuilder;
import com.amazonaws.services.s3.model.GeneratePresignedUrlRequest;
import sirius.kernel.commons.Strings;
import sirius.kernel.di.std.ConfigValue;
import sirius.kernel.di.std.Register;

import java.net.URL;
import java.util.Optional;
import java.util.stream.Stream;

/**
* Represents an upstream S3 instance which can be used in case an object is not found locally.
*
* <br>To enable this functionality the ConfigValue defined in this class must be set accordingly.
* <br>The minimal required fields are:<ul>
* <li>{@link AwsUpstream#s3EndPoint}</li>
* <li>{@link AwsUpstream#s3AccessKey}</li>
* <li>{@link AwsUpstream#s3SecretKey}</li>
* </ul>
* For details for the config name and expected value check each defined ConfigValue.
*/
@Register(classes = AwsUpstream.class)
public class AwsUpstream {
private static final String FALLBACK_REGION = "EU";
private static final int SOCKET_TIMEOUT = 60 * 1000 * 5;
/**
* The secret key to connect to the upstream S3 instance.
* When this value is not set, the proxy functionality is not enabled.
*/
@ConfigValue("upstreamAWS.secretKey")
private String s3SecretKey;

/**
* The access key to connect to the upstream S3 instance.
* When this value is not set, the proxy functionality is not enabled.
*/
@ConfigValue("upstreamAWS.accessKey")
private String s3AccessKey;

/**
* The endpoint used to connect to the upstream S3 instance.
* When this value is not set, the proxy functionality is not enabled.
*/
@ConfigValue("upstreamAWS.endPoint")
private String s3EndPoint;

/**
* The signing region used to connect to the upstream S3 instance.
* If not given, the value "EU" is used.
*/
@ConfigValue("upstreamAWS.signingRegion")
private String s3SigningRegion;

/**
* The signer type used to connect to the upstream S3 instance.
* This config is optional and will be ignored if missing.
*/
@ConfigValue("upstreamAWS.signerType")
private String s3SignerType;

private AmazonS3 client;

/**
* Checks if the (minimum) needed parameter are available to create the client.
*
* @return true if the minimum required config values are set.
*/
public boolean isConfigured() {
return Stream.of(s3EndPoint, s3AccessKey, s3SecretKey).allMatch(Strings::isFilled);
}

/**
* Getter for the client instance to connect to the upstream instance.
* Creates an instance if needed.
*
* @return client instance to upstream instance
* @throws IllegalStateException if called when not configured
*/
public AmazonS3 fetchClient() throws IllegalStateException {
if (client == null) {
client = createAWSClient();
}
return client;
}

private AmazonS3 createAWSClient() throws IllegalStateException {
sabieber marked this conversation as resolved.
Show resolved Hide resolved
if (!isConfigured()) {
throw new IllegalStateException("Use of not configured instance");
}
AWSStaticCredentialsProvider credentialsProvider =
new AWSStaticCredentialsProvider(new BasicAWSCredentials(s3AccessKey, s3SecretKey));
AwsClientBuilder.EndpointConfiguration endpointConfiguration = new AwsClientBuilder.EndpointConfiguration(
s3EndPoint,
Optional.ofNullable(s3SigningRegion).orElse(FALLBACK_REGION));
ClientConfiguration config = new ClientConfiguration().withSocketTimeout(SOCKET_TIMEOUT);
Optional.ofNullable(s3SignerType).ifPresent(config::withSignerOverride);

return AmazonS3ClientBuilder.standard()
.withClientConfiguration(config)
.withPathStyleAccessEnabled(true)
.withCredentials(credentialsProvider)
.withEndpointConfiguration(endpointConfiguration)
.build();
}

/**
* Creates the url used to tunnel request to upstream instance.
* <br><b>Important: If you do not request the content, the connection must use the method "HEAD"!</b>
*
* @param bucket from which an object is fetched
* @param object which should be fetched
* @param requestFile signalized if the content is needed or not
* @return an url which can be used to perform the matching request.
* @throws IllegalStateException if called when not configured
*/
public URL generateGetObjectURL(Bucket bucket, StoredObject object, boolean requestFile) throws IllegalStateException {
GeneratePresignedUrlRequest request = new GeneratePresignedUrlRequest(bucket.getName(), object.getKey());
if (requestFile) {
request.setMethod(HttpMethod.GET);
} else {
request.setMethod(HttpMethod.HEAD);
}

return fetchClient().generatePresignedUrl(request);
}

public void setS3SecretKey(String s3SecretKey) {
this.s3SecretKey = s3SecretKey;
}

public void setS3AccessKey(String s3AccessKey) {
this.s3AccessKey = s3AccessKey;
}

public void setS3EndPoint(String s3EndPoint) {
this.s3EndPoint = s3EndPoint;
}

public void setS3SignerType(String s3SignerType) {
this.s3SignerType = s3SignerType;
}
}
27 changes: 27 additions & 0 deletions src/main/java/ninja/S3Dispatcher.java
Expand Up @@ -19,6 +19,7 @@
import ninja.errors.S3ErrorCode;
import ninja.errors.S3ErrorSynthesizer;
import ninja.queries.S3QuerySynthesizer;
import org.asynchttpclient.BoundRequestBuilder;
import sirius.kernel.async.CallContext;
import sirius.kernel.commons.Callback;
import sirius.kernel.commons.Hasher;
Expand Down Expand Up @@ -47,6 +48,7 @@
import java.io.IOException;
import java.io.RandomAccessFile;
import java.net.InetAddress;
import java.net.URL;
import java.nio.channels.FileChannel;
import java.time.Instant;
import java.time.ZoneOffset;
Expand All @@ -64,6 +66,7 @@
import java.util.Properties;
import java.util.Set;
import java.util.TreeSet;
import java.util.function.Consumer;
import java.util.regex.Matcher;
import java.util.stream.Collectors;

Expand Down Expand Up @@ -116,6 +119,9 @@ private static class S3Request {
@ConfigValue("storage.multipartDir")
private String multipartDir;

@Part
private AwsUpstream awsUpstream;

private final Set<String> multipartUploads = Collections.synchronizedSet(new TreeSet<>());

private final Counter uploadIdCounter = new Counter();
Expand Down Expand Up @@ -645,6 +651,20 @@ private void deleteObject(final WebContext webContext, final Bucket bucket, fina
StoredObject object = bucket.getObject(id);
object.delete();

// If it exists online, we mark it locally as "deleted"
if (awsUpstream.isConfigured() && awsUpstream.fetchClient().doesObjectExist(bucket.getName(), id)) {
try {
object.markDeleted();
} catch (IOException ignored) {
signalObjectError(webContext,
bucket.getName(),
id,
S3ErrorCode.InternalError,
Strings.apply("Error while marking file as deleted"));
return;
}
}

webContext.respondWith().status(HttpResponseStatus.NO_CONTENT);
signalObjectSuccess(webContext);
}
Expand Down Expand Up @@ -771,6 +791,13 @@ private void copyObject(WebContext webContext, Bucket bucket, String id, String
*/
private void getObject(WebContext webContext, Bucket bucket, String id, boolean sendFile) throws IOException {
StoredObject object = bucket.getObject(id);
if (!object.exists() && !object.isMarkedDeleted() && awsUpstream.isConfigured()) {
URL fetchURL = awsUpstream.generateGetObjectURL(bucket, object, sendFile);
Consumer<BoundRequestBuilder> requestTuner = requestBuilder -> requestBuilder.setMethod(sendFile ? "GET" : "HEAD");
webContext.enableTiming(null).respondWith().tunnel(fetchURL.toString(), requestTuner, null, null);
return;
}

if (!object.exists()) {
signalObjectError(webContext, bucket.getName(), id, S3ErrorCode.NoSuchKey, "Object does not exist");
return;
Expand Down
22 changes: 22 additions & 0 deletions src/main/java/ninja/StoredObject.java
Expand Up @@ -29,6 +29,8 @@
*/
public class StoredObject {

private static final String DELETED_MARKER = "DeletedMarker";

private final File file;

private final String key;
Expand Down Expand Up @@ -222,6 +224,26 @@ public void setProperties(Map<String, String> properties) throws IOException {
}
}

/**
* Checks if the marker for "deleted" is set.
* When an object is marked as "deleted" it can not be requested anymore.
* @return true if this file is "deleted"
*/
public boolean isMarkedDeleted() {
return getProperties().containsKey(DELETED_MARKER);
}

/**
* Sets the object as "deleted", all requests onto this object are handled as if it is deleted.
* <br><b>This method does not perform an actual delete!<br>To perform an actual delete please check {@link StoredObject#delete} </b>
* @throws IOException if the properties could not be updated
*/
public void markDeleted() throws IOException {
Map<String, String> fileProperties = getProperties();
fileProperties.put(DELETED_MARKER, "true");
setProperties(fileProperties);
}

/**
* Checks whether the given string is valid for use as object key.
* <p>
Expand Down