Skip to content

Commit

Permalink
HDDS-10390. MiniOzoneCluster to support S3 Gateway (apache#7281)
Browse files Browse the repository at this point in the history
  • Loading branch information
@ivandika3
ivandika3 authored Oct 13, 2024
1 parent 8eef589 commit c044b79
Show file tree
Hide file tree
Showing 11 changed files with 1,293 additions and 19 deletions.
12 changes: 12 additions & 0 deletions hadoop-ozone/integration-test/pom.xml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -141,6 +141,10 @@ https://maven.apache.org/xsd/maven-4.0.0.xsd">
<groupId>org.slf4j</groupId>
<artifactId>*</artifactId>
</exclusion>
<exclusion>
<groupId>com.sun.jersey</groupId>
<artifactId>jersey-servlet</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
Expand All @@ -161,6 +165,10 @@ https://maven.apache.org/xsd/maven-4.0.0.xsd">
<groupId>org.slf4j</groupId>
<artifactId>*</artifactId>
</exclusion>
<exclusion>
<groupId>com.sun.jersey</groupId>
<artifactId>jersey-servlet</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
Expand Down Expand Up @@ -216,6 +224,10 @@ https://maven.apache.org/xsd/maven-4.0.0.xsd">
<groupId>org.slf4j</groupId>
<artifactId>*</artifactId>
</exclusion>
<exclusion>
<groupId>com.sun.jersey</groupId>
<artifactId>jersey-servlet</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
Expand Down
38 changes: 37 additions & 1 deletion hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/MiniOzoneCluster.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -35,11 +35,14 @@
import org.apache.hadoop.ozone.client.OzoneClient;
import org.apache.hadoop.ozone.om.OzoneManager;
import org.apache.hadoop.ozone.recon.ReconServer;
import org.apache.hadoop.ozone.s3.Gateway;
import org.apache.hadoop.security.authentication.client.AuthenticationException;
import org.apache.ozone.test.GenericTestUtils;
import org.apache.ratis.util.ExitUtils;
import org.apache.ratis.util.function.CheckedFunction;

import com.amazonaws.services.s3.AmazonS3;

/**
* Interface used for MiniOzoneClusters.
*/
Expand Down Expand Up @@ -142,10 +145,17 @@ void waitForPipelineTobeReady(HddsProtos.ReplicationFactor factor,
/**
* Returns a {@link ReconServer} instance.
*
* @return List of {@link ReconServer}
* @return {@link ReconServer} instance if it is initialized, otherwise null.
*/
ReconServer getReconServer();

/**
* Returns a {@link Gateway} instance.
*
* @return {@link Gateway} instance if it is initialized, otherwise null.
*/
Gateway getS3G();

/**
* Returns an {@link OzoneClient} to access the {@link MiniOzoneCluster}.
* The caller is responsible for closing the client after use.
Expand All @@ -154,6 +164,11 @@ void waitForPipelineTobeReady(HddsProtos.ReplicationFactor factor,
*/
OzoneClient newClient() throws IOException;

/**
* Returns an {@link AmazonS3} to access the {@link MiniOzoneCluster}.
*/
AmazonS3 newS3Client();

/**
* Returns StorageContainerLocationClient to communicate with
* {@link StorageContainerManager} associated with the MiniOzoneCluster.
Expand Down Expand Up @@ -219,6 +234,21 @@ void restartHddsDatanode(DatanodeDetails dn, boolean waitForDatanode)
*/
void stopRecon();

/**
* Start S3G.
*/
void startS3G();

/**
* Restart S3G.
*/
void restartS3G();

/**
* Stop S3G.
*/
void stopS3G();

/**
* Shutdown the MiniOzoneCluster and delete the storage dirs.
*/
Expand Down Expand Up @@ -273,6 +303,7 @@ abstract class Builder {
protected String omId = UUID.randomUUID().toString();

protected boolean includeRecon = false;
protected boolean includeS3G = false;

protected int dnInitialVersion = DatanodeVersion.FUTURE_VERSION.toProtoValue();
protected int dnCurrentVersion = DatanodeVersion.COMBINED_PUTBLOCK_WRITECHUNK_RPC.toProtoValue();
Expand Down Expand Up @@ -382,6 +413,11 @@ public Builder includeRecon(boolean include) {
return this;
}

public Builder includeS3G(boolean include) {
this.includeS3G = include;
return this;
}

/**
* Constructs and returns MiniOzoneCluster.
*
Expand Down
133 changes: 131 additions & 2 deletions ...op-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/MiniOzoneClusterImpl.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,14 @@
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import com.amazonaws.ClientConfiguration;
import com.amazonaws.auth.AWSCredentialsProvider;
import com.amazonaws.auth.AWSStaticCredentialsProvider;
import com.amazonaws.auth.BasicAWSCredentials;
import com.amazonaws.client.builder.AwsClientBuilder;
import com.amazonaws.regions.Regions;
import com.amazonaws.services.s3.AmazonS3;
import com.amazonaws.services.s3.AmazonS3ClientBuilder;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.hdds.DatanodeVersion;
import org.apache.hadoop.hdds.HddsConfigKeys;
Expand All @@ -58,6 +66,7 @@
import org.apache.hadoop.hdds.scm.server.StorageContainerManager;
import org.apache.hadoop.hdds.security.symmetric.SecretKeyClient;
import org.apache.hadoop.hdds.security.x509.certificate.client.CertificateClient;
import org.apache.hadoop.hdds.server.http.HttpConfig;
import org.apache.hadoop.hdds.utils.IOUtils;
import org.apache.hadoop.hdds.utils.db.CodecBuffer;
import org.apache.hadoop.hdds.utils.db.CodecTestUtil;
Expand All @@ -73,6 +82,10 @@
import org.apache.hadoop.ozone.om.OzoneManager;
import org.apache.hadoop.ozone.recon.ConfigurationProvider;
import org.apache.hadoop.ozone.recon.ReconServer;
import org.apache.hadoop.ozone.s3.Gateway;
import org.apache.hadoop.ozone.s3.OzoneClientCache;
import org.apache.hadoop.ozone.s3.OzoneConfigurationHolder;
import org.apache.hadoop.ozone.s3.S3GatewayConfigKeys;
import org.apache.hadoop.security.authentication.client.AuthenticationException;
import org.apache.ozone.test.GenericTestUtils;

Expand All @@ -84,9 +97,14 @@
import static org.apache.hadoop.hdds.recon.ReconConfigKeys.OZONE_RECON_DATANODE_ADDRESS_KEY;
import static org.apache.hadoop.hdds.recon.ReconConfigKeys.OZONE_RECON_HTTP_ADDRESS_KEY;
import static org.apache.hadoop.hdds.recon.ReconConfigKeys.OZONE_RECON_TASK_SAFEMODE_WAIT_THRESHOLD;
import static org.apache.hadoop.hdds.server.http.HttpConfig.getHttpPolicy;
import static org.apache.hadoop.hdds.server.http.HttpServer2.HTTPS_SCHEME;
import static org.apache.hadoop.hdds.server.http.HttpServer2.HTTP_SCHEME;
import static org.apache.hadoop.ozone.recon.ReconServerConfigKeys.OZONE_RECON_DB_DIR;
import static org.apache.hadoop.ozone.recon.ReconServerConfigKeys.OZONE_RECON_OM_SNAPSHOT_DB_DIR;
import static org.apache.hadoop.ozone.recon.ReconServerConfigKeys.OZONE_RECON_SCM_DB_DIR;
import static org.apache.hadoop.ozone.s3.S3GatewayConfigKeys.OZONE_S3G_HTTPS_ADDRESS_KEY;
import static org.apache.hadoop.ozone.s3.S3GatewayConfigKeys.OZONE_S3G_HTTP_ADDRESS_KEY;
import static org.apache.ozone.test.GenericTestUtils.PortAllocator.anyHostWithFreePort;
import static org.apache.ozone.test.GenericTestUtils.PortAllocator.getFreePort;
import static org.apache.ozone.test.GenericTestUtils.PortAllocator.localhostWithFreePort;
Expand Down Expand Up @@ -120,6 +138,7 @@ public class MiniOzoneClusterImpl implements MiniOzoneCluster {
private OzoneManager ozoneManager;
private final List<HddsDatanodeService> hddsDatanodes;
private ReconServer reconServer;
private Gateway s3g;

// Timeout for the cluster to be ready
private int waitForClusterToBeReadyTimeout = 120000; // 2 min
Expand All @@ -136,13 +155,15 @@ private MiniOzoneClusterImpl(OzoneConfiguration conf,
OzoneManager ozoneManager,
StorageContainerManager scm,
List<HddsDatanodeService> hddsDatanodes,
ReconServer reconServer) {
ReconServer reconServer,
Gateway s3g) {
this.conf = conf;
this.ozoneManager = ozoneManager;
this.scm = scm;
this.hddsDatanodes = hddsDatanodes;
this.reconServer = reconServer;
this.scmConfigurator = scmConfigurator;
this.s3g = s3g;
}

/**
Expand Down Expand Up @@ -268,6 +289,11 @@ public ReconServer getReconServer() {
return this.reconServer;
}

@Override
public Gateway getS3G() {
return this.s3g;
}

@Override
public int getHddsDatanodeIndex(DatanodeDetails dn) throws IOException {
for (HddsDatanodeService service : hddsDatanodes) {
Expand All @@ -286,6 +312,54 @@ public OzoneClient newClient() throws IOException {
return client;
}

@Override
public AmazonS3 newS3Client() {
// TODO: Parameterize tests between Virtual host style and Path style
return createS3Client(true);
}

public AmazonS3 createS3Client(boolean enablePathStyle) {
final String accessKey = "user";
final String secretKey = "password";
final Regions region = Regions.DEFAULT_REGION;

final String protocol;
final HttpConfig.Policy webPolicy = getHttpPolicy(conf);
String host;

if (webPolicy.isHttpsEnabled()) {
protocol = HTTPS_SCHEME;
host = conf.get(OZONE_S3G_HTTPS_ADDRESS_KEY);
} else {
protocol = HTTP_SCHEME;
host = conf.get(OZONE_S3G_HTTP_ADDRESS_KEY);
}

String endpoint = protocol + "://" + host;

AWSCredentialsProvider credentials = new AWSStaticCredentialsProvider(
new BasicAWSCredentials(accessKey, secretKey)
);


ClientConfiguration clientConfiguration = new ClientConfiguration();
LOG.info("S3 Endpoint is {}", endpoint);

AmazonS3 s3Client =
AmazonS3ClientBuilder.standard()
.withPathStyleAccessEnabled(enablePathStyle)
.withEndpointConfiguration(
new AwsClientBuilder.EndpointConfiguration(
endpoint, region.getName()
)
)
.withClientConfiguration(clientConfiguration)
.withCredentials(credentials)
.build();

return s3Client;
}

protected OzoneClient createClient() throws IOException {
return OzoneClientFactory.getRpcClient(conf);
}
Expand Down Expand Up @@ -428,6 +502,7 @@ public void stop() {
stopDatanodes(hddsDatanodes);
stopSCM(scm);
stopRecon(reconServer);
stopS3G(s3g);
}

private void startHddsDatanode(HddsDatanodeService datanode) {
Expand Down Expand Up @@ -467,6 +542,23 @@ public void stopRecon() {
stopRecon(reconServer);
}

@Override
public void startS3G() {
s3g = new Gateway();
s3g.execute(NO_ARGS);
}

@Override
public void restartS3G() {
stopS3G(s3g);
startS3G();
}

@Override
public void stopS3G() {
stopS3G(s3g);
}

private CertificateClient getCAClient() {
return this.caClient;
}
Expand Down Expand Up @@ -521,6 +613,19 @@ private static void stopRecon(ReconServer reconServer) {
}
}

private static void stopS3G(Gateway s3g) {
try {
if (s3g != null) {
LOG.info("Stopping S3G");
// TODO (HDDS-11539): Remove this workaround once the @PreDestroy issue is fixed
OzoneClientCache.closeClient();
s3g.stop();
}
} catch (Exception e) {
LOG.error("Exception while shutting down S3 Gateway.", e);
}
}

/**
* Builder for configuring the MiniOzoneCluster to run.
*/
Expand All @@ -544,15 +649,17 @@ public MiniOzoneCluster build() throws IOException {
OzoneManager om = null;
ReconServer reconServer = null;
List<HddsDatanodeService> hddsDatanodes = Collections.emptyList();
Gateway s3g = null;
try {
scm = createAndStartSingleSCM();
om = createAndStartSingleOM();
s3g = createS3G();
reconServer = createRecon();
hddsDatanodes = createHddsDatanodes();

MiniOzoneClusterImpl cluster = new MiniOzoneClusterImpl(conf,
scmConfigurator, om, scm,
hddsDatanodes, reconServer);
hddsDatanodes, reconServer, s3g);

cluster.setCAClient(certClient);
cluster.setSecretKeyClient(secretKeyClient);
Expand All @@ -567,6 +674,9 @@ public MiniOzoneCluster build() throws IOException {
if (includeRecon) {
stopRecon(reconServer);
}
if (includeS3G) {
stopS3G(s3g);
}
if (startDataNodes) {
stopDatanodes(hddsDatanodes);
}
Expand Down Expand Up @@ -740,6 +850,16 @@ protected ReconServer createRecon() {
return reconServer;
}

protected Gateway createS3G() {
Gateway s3g = null;
if (includeS3G) {
configureS3G();
s3g = new Gateway();
s3g.execute(NO_ARGS);
}
return s3g;
}

/**
* Creates HddsDatanodeService(s) instance.
*
Expand Down Expand Up @@ -806,5 +926,14 @@ protected void configureRecon() {
ConfigurationProvider.setConfiguration(conf);
}

private void configureS3G() {
OzoneConfigurationHolder.resetConfiguration();

conf.set(S3GatewayConfigKeys.OZONE_S3G_HTTP_ADDRESS_KEY, localhostWithFreePort());
conf.set(S3GatewayConfigKeys.OZONE_S3G_HTTPS_ADDRESS_KEY, localhostWithFreePort());

OzoneConfigurationHolder.setConfiguration(conf);
}

}
}
Loading

0 comments on commit c044b79

Please sign in to comment.