Skip to content

Commit 3d9911b

Browse files
CoderYellowCoderYellow
committed
feat(extensions): add S3 and GCS upload extensions with programmatic and config-based support
- Added `S3UploadExtension` and `GcsUploadExtension` for S3 and GCS file/directory uploads after test services start. - Introduced support for configuring upload sources via `.toml` and programmatic APIs. - Enhanced `BigDataExtensionsConfigLoader` and `BigDataExtensionsBuilder` to support new extensions. - Updated Gradle build with AWS S3 SDK and Google Cloud Storage SDK dependencies. - Included detailed examples and outputs in user documentation for new extensions.
1 parent cfe8c97 commit 3d9911b

8 files changed

Lines changed: 624 additions & 0 deletions

File tree

‎doc/user-guide.adoc‎

Lines changed: 58 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1346,6 +1346,64 @@ Outputs:
13461346
* `spark-gcs-bucket.bucket`
13471347
* `spark-gcs-bucket.gs.uri`
13481348

1349+
=== Object Store Upload Extensions
1350+
1351+
The S3 and GCS upload extensions create the target bucket by default and upload local files or folders after the object-store services start.
1352+
They use the AWS S3 SDK and Google Cloud Storage SDK against the endpoints exposed by `BigDataTestKit`.
1353+
1354+
[source,toml]
1355+
----
1356+
[[s3Uploads]]
1357+
id = "seed-s3"
1358+
bucket = "demo-s3"
1359+
prefix = "input"
1360+
sources = [
1361+
{ source = "file:src/test/resources/data/users.json", key = "input/users.json", contentType = "application/json" },
1362+
{ source = "src/test/resources/data/events", prefix = "events" }
1363+
]
1364+
1365+
[[gcsUploads]]
1366+
id = "seed-gcs"
1367+
bucket = "demo-gcs"
1368+
project = "bigdata-test"
1369+
prefix = "input"
1370+
createBucket = true
1371+
sources = [
1372+
{ source = "classpath:data/products.json", key = "input/products.json", contentType = "application/json" }
1373+
]
1374+
----
1375+
1376+
Programmatic config:
1377+
1378+
[source,kotlin]
1379+
----
1380+
extensions.s3Upload("demo-s3") {
1381+
id = "seed-s3"
1382+
prefix = "input"
1383+
file("src/test/resources/data/users.json", key = "input/users.json", contentType = "application/json")
1384+
directory("src/test/resources/data/events", prefix = "events")
1385+
}
1386+
1387+
extensions.gcsUpload("demo-gcs") {
1388+
id = "seed-gcs"
1389+
project = "bigdata-test"
1390+
file("classpath:data/products.json", key = "input/products.json", contentType = "application/json")
1391+
}
1392+
----
1393+
1394+
`source` accepts `file:<path>` or a plain filesystem path for files and directories.
1395+
Single files can also use `classpath:<resource>`.
1396+
Directory uploads should use filesystem paths because classpath directories are not reliably listable after resources are packaged in a jar.
1397+
When `key` is omitted for a file, the object key is built from the extension `prefix`, source `prefix`, and file name.
1398+
For directories, object keys are built from the extension `prefix`, source `prefix`, and the relative file path under the directory.
1399+
1400+
Outputs:
1401+
1402+
* `<id>.bucket`
1403+
* `<id>.s3a.uri` or `<id>.gs.uri`
1404+
* `<id>.uploaded-count`
1405+
* `<id>.uploaded.<index>.key`
1406+
13491407
=== S3 JCEKS Extension
13501408

13511409
The `s3Jceks` extension creates a Hadoop credential provider file in HDFS and stores the S3 emulator credentials in it.

‎extensions/build.gradle.kts‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,8 @@ dependencies {
4040
compileOnly(libs.kafkaAvroSerializer)
4141
compileOnly(libs.kafkaSchemaRegistryClient)
4242
compileOnly(libs.avro)
43+
compileOnly(libs.awsSdkS3)
44+
compileOnly(libs.googleCloudStorage)
4345
compileOnly(libs.sparkSql)
4446
compileOnly(libs.sparkHive)
4547
compileOnly(libs.servletApi)
@@ -52,6 +54,7 @@ dependencies {
5254
}
5355
shadedRuntime(libs.awsSdkS3)
5456
shadedRuntime(libs.awsSdkS3TransferManager)
57+
shadedRuntime(libs.googleCloudStorage)
5558
shadedRuntime(libs.kafkaAvroSerializer)
5659
shadedRuntime(libs.kafkaSchemaRegistryClient)
5760
shadedRuntime(libs.lz4Java)
@@ -75,6 +78,8 @@ dependencies {
7578
testImplementation(libs.kafkaAvroSerializer)
7679
testImplementation(libs.kafkaSchemaRegistryClient)
7780
testImplementation(libs.avro)
81+
testImplementation(libs.awsSdkS3)
82+
testImplementation(libs.googleCloudStorage)
7883
testImplementation(libs.sparkSql)
7984
testImplementation(libs.sparkHive)
8085
testRuntimeOnly(libs.junitJupiterEngine)

‎extensions/src/main/kotlin/org/openprojectx/bigdata/test/extensions/config/BigDataExtensionsBuilder.kt‎

Lines changed: 79 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,10 @@ import org.openprojectx.bigdata.test.extensions.kafka.KafkaAvroRecordSeed
1212
import org.openprojectx.bigdata.test.extensions.kafka.KafkaAvroSeedExtension
1313
import org.openprojectx.bigdata.test.extensions.kafka.KafkaAvroTopicSeed
1414
import org.openprojectx.bigdata.test.extensions.objectstore.GcsBucketExtension
15+
import org.openprojectx.bigdata.test.extensions.objectstore.GcsUploadExtension
16+
import org.openprojectx.bigdata.test.extensions.objectstore.ObjectStoreUploadSource
1517
import org.openprojectx.bigdata.test.extensions.objectstore.S3BucketExtension
18+
import org.openprojectx.bigdata.test.extensions.objectstore.S3UploadExtension
1619
import org.openprojectx.bigdata.test.extensions.spark.SparkSqlPreparationExtension
1720
import org.openprojectx.bigdata.test.extensions.spark.SparkSqlPreparationStatement
1821
import java.util.function.Consumer
@@ -68,6 +71,22 @@ class BigDataExtensionsBuilder {
6871
gcsBucket(bucket = bucket, id = id, project = "bigdata-test")
6972
}
7073

74+
fun s3Upload(bucket: String, configure: ObjectStoreUploadBuilder.() -> Unit) {
75+
extensions += ObjectStoreUploadBuilder(bucket = bucket, id = "s3-upload-$bucket").apply(configure).buildS3()
76+
}
77+
78+
fun s3Upload(bucket: String, configure: Consumer<ObjectStoreUploadBuilder>) {
79+
extensions += ObjectStoreUploadBuilder(bucket = bucket, id = "s3-upload-$bucket").also { configure.accept(it) }.buildS3()
80+
}
81+
82+
fun gcsUpload(bucket: String, configure: GcsUploadBuilder.() -> Unit) {
83+
extensions += GcsUploadBuilder(bucket = bucket, id = "gcs-upload-$bucket").apply(configure).buildGcs()
84+
}
85+
86+
fun gcsUpload(bucket: String, configure: Consumer<GcsUploadBuilder>) {
87+
extensions += GcsUploadBuilder(bucket = bucket, id = "gcs-upload-$bucket").also { configure.accept(it) }.buildGcs()
88+
}
89+
7190
fun sparkSqlPreparation(configure: SparkSqlPreparationBuilder.() -> Unit) {
7291
extensions += SparkSqlPreparationBuilder().apply(configure).build()
7392
}
@@ -79,6 +98,66 @@ class BigDataExtensionsBuilder {
7998
internal fun build(): List<BigDataExtension> = extensions.toList()
8099
}
81100

101+
open class ObjectStoreUploadBuilder internal constructor(
102+
protected val bucket: String,
103+
var id: String,
104+
) {
105+
var prefix: String = ""
106+
var createBucket: Boolean = true
107+
protected val sources = mutableListOf<ObjectStoreUploadSource>()
108+
109+
fun file(source: String, key: String? = null, contentType: String? = null) {
110+
sources += ObjectStoreUploadSource(source = source, key = key, contentType = contentType)
111+
}
112+
113+
fun file(source: String) {
114+
file(source = source, key = null, contentType = null)
115+
}
116+
117+
fun file(source: String, key: String) {
118+
file(source = source, key = key, contentType = null)
119+
}
120+
121+
fun directory(source: String, prefix: String = "", recursive: Boolean = true, contentType: String? = null) {
122+
sources += ObjectStoreUploadSource(
123+
source = source,
124+
prefix = prefix,
125+
recursive = recursive,
126+
contentType = contentType,
127+
)
128+
}
129+
130+
fun directory(source: String) {
131+
directory(source = source, prefix = "", recursive = true, contentType = null)
132+
}
133+
134+
internal fun buildS3(): S3UploadExtension =
135+
S3UploadExtension(
136+
id = id,
137+
bucket = bucket,
138+
prefix = prefix,
139+
createBucket = createBucket,
140+
sources = sources.toList(),
141+
)
142+
}
143+
144+
class GcsUploadBuilder internal constructor(
145+
bucket: String,
146+
id: String,
147+
) : ObjectStoreUploadBuilder(bucket = bucket, id = id) {
148+
var project: String = "bigdata-test"
149+
150+
internal fun buildGcs(): GcsUploadExtension =
151+
GcsUploadExtension(
152+
id = id,
153+
bucket = bucket,
154+
prefix = prefix,
155+
project = project,
156+
createBucket = createBucket,
157+
sources = sources.toList(),
158+
)
159+
}
160+
82161
class S3JceksBuilder {
83162
var id: String = "s3-jceks"
84163
var hdfsDir: String = "/bigdata-test/config"

‎extensions/src/main/kotlin/org/openprojectx/bigdata/test/extensions/config/BigDataExtensionsConfigLoader.kt‎

Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,10 @@ import org.openprojectx.bigdata.test.extensions.kafka.KafkaAvroRecordSeed
2020
import org.openprojectx.bigdata.test.extensions.kafka.KafkaAvroSeedExtension
2121
import org.openprojectx.bigdata.test.extensions.kafka.KafkaAvroTopicSeed
2222
import org.openprojectx.bigdata.test.extensions.objectstore.GcsBucketExtension
23+
import org.openprojectx.bigdata.test.extensions.objectstore.GcsUploadExtension
24+
import org.openprojectx.bigdata.test.extensions.objectstore.ObjectStoreUploadSource
2325
import org.openprojectx.bigdata.test.extensions.objectstore.S3BucketExtension
26+
import org.openprojectx.bigdata.test.extensions.objectstore.S3UploadExtension
2427
import org.openprojectx.bigdata.test.extensions.spark.SparkSqlPreparationExtensionProvider
2528
import java.util.ServiceLoader
2629

@@ -84,6 +87,33 @@ class BigDataExtensionsConfigLoader(
8487
)
8588
}
8689
}
90+
root["s3Uploads"]?.jsonArray?.forEach { item ->
91+
val config = item.jsonObject
92+
if (config.boolean("enabled", default = true)) {
93+
val bucket = config.string("bucket")
94+
extensions += S3UploadExtension(
95+
id = config.string("id", "s3-upload-$bucket"),
96+
bucket = bucket,
97+
prefix = config.string("prefix", ""),
98+
createBucket = config.boolean("createBucket", default = true),
99+
sources = config.uploadSources(),
100+
)
101+
}
102+
}
103+
root["gcsUploads"]?.jsonArray?.forEach { item ->
104+
val config = item.jsonObject
105+
if (config.boolean("enabled", default = true)) {
106+
val bucket = config.string("bucket")
107+
extensions += GcsUploadExtension(
108+
id = config.string("id", "gcs-upload-$bucket"),
109+
bucket = bucket,
110+
prefix = config.string("prefix", ""),
111+
project = config.string("project", "bigdata-test"),
112+
createBucket = config.boolean("createBucket", default = true),
113+
sources = config.uploadSources(),
114+
)
115+
}
116+
}
87117
root["extensions"]?.jsonArray?.forEach { item ->
88118
val config = item.jsonObject
89119
val type = config.string("type")
@@ -114,12 +144,27 @@ class BigDataExtensionsConfigLoader(
114144
)
115145
}
116146

147+
private fun JsonObject.uploadSources(): List<ObjectStoreUploadSource> =
148+
this["sources"]?.jsonArray?.map { item ->
149+
val source = item.jsonObject
150+
ObjectStoreUploadSource(
151+
source = source.string("source"),
152+
key = source.optionalString("key"),
153+
prefix = source.string("prefix", ""),
154+
recursive = source.boolean("recursive", default = true),
155+
contentType = source.optionalString("contentType"),
156+
)
157+
}.orEmpty()
158+
117159
private fun JsonObject.string(name: String): String =
118160
this[name]?.jsonPrimitive?.contentOrNull ?: error("Missing required extension config field '$name'")
119161

120162
private fun JsonObject.string(name: String, default: String): String =
121163
this[name]?.jsonPrimitive?.contentOrNull ?: default
122164

165+
private fun JsonObject.optionalString(name: String): String? =
166+
this[name]?.jsonPrimitive?.contentOrNull
167+
123168
private fun JsonObject.int(name: String, default: Int): Int =
124169
this[name]?.jsonPrimitive?.intOrNull ?: default
125170

‎extensions/src/main/kotlin/org/openprojectx/bigdata/test/extensions/core/BigDataExtensionResourceLoader.kt‎

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,4 +24,42 @@ class BigDataExtensionResourceLoader(
2424
location.startsWith("file:") -> Files.readString(Path.of(location.removePrefix("file:")))
2525
else -> location
2626
}
27+
28+
fun readBytes(location: String): ByteArray = when {
29+
location.startsWith("classpath:") -> {
30+
val resource = location.removePrefix("classpath:").trimStart('/')
31+
classLoader.getResource(resource)?.readBytes()
32+
?: resourceDirectories
33+
.asSequence()
34+
.map { it.resolve(resource) }
35+
.firstOrNull(Files::isRegularFile)
36+
?.let(Files::readAllBytes)
37+
?: error(
38+
"Classpath resource '$resource' was not found in the extension runtime classpath " +
39+
"or configured resource directories: ${resourceDirectories.joinToString()}",
40+
)
41+
}
42+
location.startsWith("file:") -> Files.readAllBytes(Path.of(location.removePrefix("file:")))
43+
else -> Files.readAllBytes(Path.of(location))
44+
}
45+
46+
fun resolveDirectory(location: String): Path {
47+
val path = when {
48+
location.startsWith("file:") -> Path.of(location.removePrefix("file:"))
49+
location.startsWith("classpath:") -> {
50+
val resource = location.removePrefix("classpath:").trimStart('/')
51+
resourceDirectories
52+
.asSequence()
53+
.map { it.resolve(resource) }
54+
.firstOrNull(Files::isDirectory)
55+
?: error(
56+
"Classpath directory '$resource' cannot be resolved from the extension resource directories. " +
57+
"Use a file path for directory uploads when resources are packaged in a jar.",
58+
)
59+
}
60+
else -> Path.of(location)
61+
}
62+
require(Files.isDirectory(path)) { "Upload source directory '$location' does not exist or is not a directory" }
63+
return path
64+
}
2765
}

0 commit comments

Comments
 (0)