Merge pull request #74 from open-AIMS/feat/minio-dropin

Author: PeterBaker0

.env.dist

@@ -0,0 +1,13 @@
+# Local web-api
+API_ENDPOINT=http://localhost:5000
+AWS_REGION=ap-southeast-2
+JOB_TYPES=SUITABILITY_ASSESSMENT,REGIONAL_ASSESSMENT,TEST
+WORKER_USERNAME=worker@email.com
+WORKER_PASSWORD=password
+JULIA_DEBUG=ReefGuideAPI
+CONFIG_PATH=config.toml
+
+# Minio (local)
+S3_ENDPOINT=http://localhost:9000
+MINIO_USERNAME=minioadmin
+MINIO_PASSWORD=minioadmin

.env.local

@@ -2,7 +2,7 @@
 
 julia_version = "1.11.5"
 manifest_format = "2.0"
-project_hash = "6fb79bd8da16ed2240c62044e56e825aaaef4a46"
+project_hash = "3d5a0b33a7af4f2946e13fad2322181d045cdacf"
 
 [[deps.ADTypes]]
 git-tree-sha1 = "e2478490447631aedba0823d4d7a80b2cc8cdb32"
@@ -1614,6 +1614,12 @@ git-tree-sha1 = "bc95bf4149bf535c09602e3acdf950d9b4376227"
 uuid = "9237b28f-5490-5468-be7b-bb81f5f5e6cf"
 version = "10.1.4+3"
 
+[[deps.Minio]]
+deps = ["AWS", "AWSS3", "FilePathsBase", "Pkg", "URIs", "minio_jll"]
+git-tree-sha1 = "ab0cf8db62040e835197b4e9416ab34baa89b693"
+uuid = "4281f0d9-7ae0-406e-9172-b7277c1efa20"
+version = "0.2.2"
+
 [[deps.Missings]]
 deps = ["DataAPI"]
 git-tree-sha1 = "ec4f7fbeab05d7747bdf98eb74d130a2a2ed298d"
@@ -2677,6 +2683,12 @@ git-tree-sha1 = "86addc139bca85fdf9e7741e10977c45785727b7"
 uuid = "337d8026-41b4-5cde-a456-74a10e5b31d1"
 version = "1.11.3+0"
 
+[[deps.minio_jll]]
+deps = ["Artifacts", "JLLWrappers", "Libdl", "Pkg"]
+git-tree-sha1 = "59287c60d58776bbdecef63f8fda50ff345bb4dd"
+uuid = "ad83bbfc-c153-533a-bb97-700a7db721e0"
+version = "1.0.0+1"
+
 [[deps.nghttp2_jll]]
 deps = ["Artifacts", "Libdl"]
 uuid = "8e850ede-7688-5339-a07c-302acd2aaf8d"

Manifest-v1.11.toml

@@ -35,6 +35,7 @@ LinearAlgebra = "37e2e46d-f89d-539d-b4ee-838fcccc9c8e"
 Logging = "56ddb016-857b-54e1-b83d-db4d58db5568"
 MKL_jll = "856f044c-d86e-5d09-b602-aeab76dc8ba7"
 Memoization = "6fafb56a-5788-4b4e-91ca-c0cea6611c73"
+Minio = "4281f0d9-7ae0-406e-9172-b7277c1efa20"
 Mmap = "a63ad114-7e13-5084-954f-fe012c677804"
 NearestNeighbors = "b8a86587-4115-5ab1-83bc-aa920d37bbce"
 NetCDF = "30363a11-5582-574a-97bb-aa9a979735b9"
@@ -63,6 +64,7 @@ ExtendableSparse = "1.7.1"
 JSON3 = "1.14.2"
 Logging = "1.11.0"
 MKL_jll = "2024.2.0"
+Minio = "0.2.2"
 NearestNeighbors = "0.4.20"
 Oxygen = "1.7.1"
 Random = "1.11.0"

Project.toml

@@ -35,6 +35,65 @@ using ReefGuideAPI
 ReefGuideAPI.start_server(".config.toml")
 ```
 
+## Running the worker
+
+### Local quick start
+
+Ensure your local stack of [reefguide](https://github.com/open-AIMS/reefguide) is running. You can then do the below:
+
+Setup env file:
+
+```
+cp .env.local .env
+```
+
+Then run the worker
+
+```julia
+] add DotEnv
+using DotEnv
+DotEnv.load!()
+using ReefGuideAPI
+ReefGuideAPI.start_worker()
+```
+
+### Env examples
+
+The example below points to a local setup.
+
+```
+API_ENDPOINT=http://localhost:5000
+AWS_REGION=ap-southeast-2
+JOB_TYPES=SUITABILITY_ASSESSMENT,REGIONAL_ASSESSMENT,TEST
+WORKER_USERNAME=worker@email.com
+WORKER_PASSWORD=password
+JULIA_DEBUG=ReefGuideAPI
+CONFIG_PATH=config.toml
+AWS_ACCESS_KEY_ID=minioadmin
+AWS_SECRET_ACCESS_KEY=minioadmin
+
+# For minio dropin
+S3_ENDPOINT=http://localhost:9000
+MINIO_USERNAME=minioadmin
+MINIO_PASSWORD=minioadmin
+```
+
+### Running against a real deployment
+
+Typically you would deploy a production worker using a separate process to manage the env variables. But the below shows an example .env file (noting we do not include the minio dropin)
+
+```
+API_ENDPOINT='https://web-api.reefguide.example.com'
+AWS_REGION='ap-southeast-2'
+JOB_TYPES=SUITABILITY_ASSESSMENT,TEST
+WORKER_USERNAME=worker@service.com
+WORKER_PASSWORD=<password>
+POLL_INTERVAL_MS=5000
+IDLE_TIMEOUT_MS=600000
+JULIA_DEBUG=ReefGuideAPI
+CONFIG_PATH=config.toml
+```
+
 ### Using Docker
 
 ```bash

README.md

@@ -22,6 +22,8 @@ struct WorkerConfig
     config_path::String
     "AWS Region"
     aws_region::String
+    "S3 Endpoint for local S3 compatibility"
+    s3_endpoint::OptionalValue{String}
 
     # Kwarg constructor
     function WorkerConfig(;
@@ -34,7 +36,8 @@ struct WorkerConfig
         # Polling interval 2 second by default
         poll_interval_ms::Int64=2000,
         # Idle timeout 5 minutes by default
-        idle_timeout_ms::Int64=5 * 60 * 1000
+        idle_timeout_ms::Int64=5 * 60 * 1000,
+        s3_endpoint::OptionalValue{String}=nothing
     )
         return new(
             api_endpoint,
@@ -44,7 +47,8 @@ struct WorkerConfig
             poll_interval_ms,
             idle_timeout_ms,
             config_path,
-            aws_region
+            aws_region,
+            s3_endpoint
         )
     end
 end
@@ -149,6 +153,9 @@ function load_config_from_env()::WorkerConfig
         @warn "AWS_REGION environment variable not set, defaulting to $(aws_region)"
     end
 
+    # Get S3 endpoint
+    s3_endpoint = get_env("S3_ENDPOINT", false)
+
     # Optional environment variables with defaults (2 seconds)
     poll_interval_ms::Int64 = parse(
         Int64, something(get_env("POLL_INTERVAL_MS", false), string(2 * 1000))
@@ -167,7 +174,8 @@ function load_config_from_env()::WorkerConfig
         poll_interval_ms,
         idle_timeout_ms,
         config_path,
-        aws_region
+        aws_region,
+        s3_endpoint
     )
 end
 

src/job_worker/config.jl

@@ -53,11 +53,12 @@ struct HandlerContext
     "The path to the config file"
     config_path::String
     aws_region::String
+    s3_endpoint::OptionalValue{String}
 
     function HandlerContext(;
-        storage_uri::String, config_path::String, aws_region::String="ap-southeast-2"
+        storage_uri::String, config_path::String, aws_region::String="ap-southeast-2", s3_endpoint::OptionalValue{String}=nothing
     )
-        return new(storage_uri, config_path, aws_region)
+        return new(storage_uri, config_path, aws_region, s3_endpoint)
     end
 end
 
@@ -308,7 +309,7 @@ function handle_job(
     end
 
     # Now upload this to s3
-    client = S3StorageClient(; region=context.aws_region)
+    client = S3StorageClient(; region=context.aws_region, s3_endpoint=context.s3_endpoint)
 
     # Output file names
     output_file_name_rel = "regional_assessment.tiff"
@@ -415,7 +416,7 @@ function handle_job(
     end
 
     # Now upload this to s3
-    client = S3StorageClient(; region=context.aws_region)
+    client = S3StorageClient(; region=context.aws_region, s3_endpoint=context.s3_endpoint)
 
     # Output file names
     output_file_name_rel = "suitable.geojson"

src/job_worker/handlers.jl

@@ -6,6 +6,7 @@ using AWSS3
 using AWS
 using Random
 using JSONWebTokens
+using Minio
 
 include("config.jl")
 include("ecs.jl")

src/job_worker/job_worker.jl

@@ -32,12 +32,14 @@ S3 Storage Client implementation
 """
 struct S3StorageClient <: StorageClient
     region::String
+    s3_endpoint::OptionalValue{String}
 
     # Constructor with defaults
     function S3StorageClient(;
-        region::String
+        region::String,
+        s3_endpoint::OptionalValue{String}=nothing
     )
-        return new(region)
+        return new(region, s3_endpoint)
     end
 end
 
@@ -59,7 +61,17 @@ function upload_file(
 
         @debug "Uploading file from $(local_path) to $(storage_path)"
 
-        aws = AWS.AWSConfig(; region=client.region)
+        aws =
+            !isnothing(client.s3_endpoint) ?
+            # Use minio special config if we need to
+            Minio.MinioConfig(
+                client.s3_endpoint;
+                # TODO would be nicer to pass through config tree
+                username=ENV["MINIO_USERNAME"],
+                password=ENV["MINIO_PASSWORD"]
+            ) :
+            # Otherwise use typical AWS config
+            AWS.AWSConfig(; region=client.region)
 
         # Read the file content
         file_data = Base.read(local_path)

src/job_worker/storage_client.jl

@@ -51,16 +51,28 @@ struct JobContext
     config_path::String
     "AWS region for s3 storage"
     aws_region::String
+    "S3 endpoint"
+    s3_endpoint::OptionalValue{String}
+
     "Constructor that takes all fields"
     function JobContext(;
         job::Job,
         assignment::JobAssignment,
         http_client::AuthApiClient,
         task_metadata::Any,
         config_path::String,
-        aws_region::String="ap-southeast-2"
+        aws_region::String="ap-southeast-2",
+        s3_endpoint::OptionalValue{String}=nothing
     )
-        return new(job, assignment, http_client, task_metadata, config_path, aws_region)
+        return new(
+            job,
+            assignment,
+            http_client,
+            task_metadata,
+            config_path,
+            aws_region,
+            s3_endpoint
+        )
     end
 end
 
@@ -115,7 +127,8 @@ function process(::TypedJobHandler, context::JobContext)
             HandlerContext(;
                 storage_uri=storage_uri,
                 config_path=context.config_path,
-                aws_region=context.aws_region
+                aws_region=context.aws_region,
+                s3_endpoint=context.s3_endpoint
             )
         )
 
@@ -388,7 +401,8 @@ function process_job_completely(worker::WorkerService, job::Job)
             http_client=worker.http_client,
             task_metadata=worker.metadata,
             config_path=worker.config.config_path,
-            aws_region=worker.config.aws_region
+            aws_region=worker.config.aws_region,
+            s3_endpoint=worker.config.s3_endpoint
         )
 
         # Process the job with the handler