From eec8df52e7f073833846d20a4723dbb275dddc13 Mon Sep 17 00:00:00 2001 From: Brian Pondi Date: Mon, 22 Jun 2026 22:28:54 +0200 Subject: [PATCH] Fix Docker ML batch jobs and malformed result download URLs. Bootstrap callr workers without serializing the API, reconcile dead workers to error, and default api_base_url in Docker so job STAC assets download correctly; harden compose/Docker for sits/torch (thread limits, caret, workspace volume) and add UC2 tuning client helpers. --- DESCRIPTION | 2 + NAMESPACE | 1 + R/api-openeo_v1.R | 4 + R/jobs.R | 81 ++++++- R/runtime-config.R | 91 ++++++++ R/utils-api.R | 15 ++ R/zzz.R | 3 + docker-compose.yaml | 24 ++- docker/Dockerfile | 10 +- docker/docker-compose.yaml | 22 +- docker/plumber.R | 16 +- docker/server.R | 29 +++ docker/verify-r-deps.R | 2 +- .../01_ml_api_eo_data_cubes.ipynb | 95 +++++---- inst/demo-lps-2025/05_ml_tuning_random.ipynb | 63 +++--- .../SENTINEL-2_MSI_20LMR_class_2022-01-05.tif | Bin 1375 -> 1220 bytes .../data/outputs/job-results.json | 2 +- .../data/outputs_random/job-results.json | 2 +- .../data/outputs_random/tuning_results | 2 +- inst/examples/07_ml_tempcnn_tune_grid_uc2.R | 200 ++++++++++++++++++ inst/examples/download_job_results.R | 59 ++++++ inst/examples/uc2_job_client.R | 156 ++++++++++++++ inst/ml/processes.R | 74 +++++-- 23 files changed, 839 insertions(+), 114 deletions(-) create mode 100644 R/runtime-config.R create mode 100644 R/zzz.R create mode 100644 inst/examples/07_ml_tempcnn_tune_grid_uc2.R create mode 100644 inst/examples/download_job_results.R create mode 100644 inst/examples/uc2_job_client.R diff --git a/DESCRIPTION b/DESCRIPTION index acbacda..43fc486 100644 --- a/DESCRIPTION +++ b/DESCRIPTION @@ -56,5 +56,7 @@ Collate: 'jobs.R' 'pgraphs.R' 'mock.R' + 'runtime-config.R' 'run.R' + 'zzz.R' Config/roxygen2/version: 8.0.0 diff --git a/NAMESPACE b/NAMESPACE index 326b937..b81a917 100644 --- a/NAMESPACE +++ b/NAMESPACE @@ -20,6 +20,7 @@ S3method(data_serializer,openeo_tar) S3method(get_serializer,openeo_gtiff) S3method(get_serializer,openeo_json) S3method(get_serializer,openeo_rds) +export(configure_openeocraft_runtime) export(add_link) export(add_wellknown_version) export(api_conformance) diff --git a/R/api-openeo_v1.R b/R/api-openeo_v1.R index 1ca7306..5bb9678 100644 --- a/R/api-openeo_v1.R +++ b/R/api-openeo_v1.R @@ -213,6 +213,10 @@ api_job_info.openeo_v1 <- function(api, req, res, job_id) { } # Retrieve the job from the jobs_list job <- jobs[[job_id]] + reconciled <- job_reconcile_bg_process(api, user, job_id) + if (!is.null(reconciled)) { + job <- reconciled + } res$status <- 200L # TODO: populate links? job diff --git a/R/jobs.R b/R/jobs.R index 6d74177..29536f2 100644 --- a/R/jobs.R +++ b/R/jobs.R @@ -124,19 +124,15 @@ job_del_dir <- function(api, user, job_id) { } procs_read_rds <- function(api) { - file <- file.path(api_workdir(api), "procs.rds") - if (!file.exists(file)) { + procs <- api_attr(api, "bg_procs") + if (is.null(procs)) { return(list()) } - procs <- readRDS(file) procs } procs_save_rds <- function(api, procs) { - file <- file.path(api_workdir(api), "procs.rds") - tryCatch(saveRDS(procs, file), error = function(e) { - api_stop(500, "Could not save the procs file") - }) + api_attr(api, "bg_procs") <- procs invisible(NULL) } logs_read_rds <- function(api, user, job_id) { @@ -211,17 +207,80 @@ job_async <- function(api, req, user, job_id) { job_dir <- job_get_dir(api, user, job_id) # Update job status job_upd_status(api, user, job_id, "running") + cmdargs <- c("--slave", "--no-save", "--no-restore") proc <- suppressMessages(callr::r_bg( - func = function(api, req, user, job_id) { - openeocraft::job_sync(api, req, user, job_id) + func = function(user, job_id) { + api <- openeocraft:::.openeocraft_worker_api() + openeocraft::job_sync(api, req = list(), user = user, job_id = job_id) }, - args = list(api, req, user, job_id), + args = list(user, job_id), + cmdargs = cmdargs, stdout = file.path(job_dir, "_stdout.log"), stderr = file.path(job_dir, "_stderr.log"), - poll_connection = FALSE + poll_connection = FALSE, + supervise = TRUE )) proc } + +#' Mark running jobs as error when the callr worker has died. +#' +#' @keywords internal +job_reconcile_bg_process <- function(api, user, job_id) { + jobs <- job_read_rds(api, user) + if (!(job_id %in% names(jobs))) { + return(NULL) + } + job <- jobs[[job_id]] + if (!(job$status %in% c("running", "created"))) { + return(job) + } + procs <- procs_read_rds(api) + proc <- procs[[job_id]] + worker_dead <- is.null(proc) + if (!worker_dead) { + worker_dead <- !tryCatch(proc$is_alive(), error = function(e) FALSE) + } + if (!worker_dead) { + return(job) + } + created <- tryCatch( + as.POSIXct(job$created, tz = "UTC"), + error = function(e) NA + ) + if (!is.na(created)) { + age_sec <- as.numeric(difftime(Sys.time(), created, units = "secs")) + if (age_sec < 15) { + return(job) + } + } + exit_code <- tryCatch(proc$get_exit_code(), error = function(e) NA_integer_) + log_path <- file.path(job_get_dir(api, user, job_id), "_stderr.log") + tail_err <- character(0) + if (file.exists(log_path)) { + lines <- readLines(log_path, warn = FALSE) + lines <- lines[nzchar(lines)] + if (length(lines)) { + tail_err <- tail(lines, 3L) + } + } + msg <- paste( + c( + sprintf( + "Background worker exited unexpectedly (exit code %s).", + exit_code + ), + tail_err + ), + collapse = "\n" + ) + log_append(api, user, job_id, 100L, "error", msg) + job_upd_status(api, user, job_id, .job_status_error) + procs[[job_id]] <- NULL + procs_save_rds(api, procs) + jobs <- job_read_rds(api, user) + jobs[[job_id]] +} #' @rdname job_helpers #' @export job_info <- function(api, user, job_id) { diff --git a/R/runtime-config.R b/R/runtime-config.R new file mode 100644 index 0000000..ad093ed --- /dev/null +++ b/R/runtime-config.R @@ -0,0 +1,91 @@ +#' Configure thread limits and Docker resource caps for openeocraft workers. +#' +#' Called from package load and at job start so callr background workers inherit +#' the same limits as the Plumber parent (server.R options do not propagate). +#' +#' @return Invisibly NULL. +#' @keywords internal +configure_openeocraft_runtime <- function() { + for (v in c( + "OMP_NUM_THREADS", + "MKL_NUM_THREADS", + "OPENBLAS_NUM_THREADS", + "TORCH_NUM_THREADS" + )) { + if (!nzchar(Sys.getenv(v, unset = ""))) { + Sys.setenv(v = "1") + } + } + + if (!file.exists("/.dockerenv")) { + return(invisible(NULL)) + } + + options( + openeocraft.resource_fraction = as.numeric( + Sys.getenv("OPENEOCRAFT_RESOURCE_FRACTION", "0.5") + ), + openeocraft.multicores_max = as.integer( + Sys.getenv("OPENEOCRAFT_MULTICORES_MAX", "4") + ), + openeocraft.memsize = as.integer( + Sys.getenv("OPENEOCRAFT_MEMSIZE", "16") + ), + openeocraft.memsize_auto = FALSE + ) + message( + "[runtime] Docker resource limits: multicores_max=", + getOption("openeocraft.multicores_max"), + ", memsize=", + getOption("openeocraft.memsize"), + " GB, resource_fraction=", + getOption("openeocraft.resource_fraction") + ) + invisible(NULL) +} + +#' Build API + process namespace for callr job workers. +#' +#' The Plumber parent holds a live API with a populated process namespace; +#' serializing that object into callr workers is unreliable. Workers bootstrap +#' the same configuration from disk instead. +#' +#' @return openEO API object with processes loaded. +#' @keywords internal +.openeocraft_worker_api <- function() { + work_dir <- if (file.exists("/.dockerenv")) { + "/var/openeo" + } else { + "~/openeo-tests" + } + processes_file <- "/opt/dockerfiles/inst/ml/processes.R" + if (!file.exists(processes_file)) { + processes_file <- system.file("ml/processes.R", package = "openeocraft") + } + file <- system.file("ml/db.rds", package = "openeocraft") + stac_api <- openstac::create_stac( + id = "openlandmap", + title = "OpenLandMap STAC API", + description = "OpenLandMap STAC API" + ) + stac_api <- openstac::set_db(stac_api, driver = "local", file = file) + api <- create_openeo_v1( + id = "openeocraft", + title = "openEO compliant R backend", + description = "openEOcraft worker", + backend_version = "0.3.1", + stac_api = stac_api, + work_dir = work_dir, + production = FALSE + ) + set_credentials(api, file = "~/openeo-credentials.rds") + base_url <- Sys.getenv("OPENEOCRAFT_API_BASE_URL", unset = "") + if (!nzchar(base_url) && file.exists("/.dockerenv")) { + base_url <- "http://127.0.0.1:8000" + } + if (nzchar(base_url)) { + assign("api_base_url", base_url, envir = attr(api, "env")) + } + load_processes(api, processes_file) + api +} diff --git a/R/utils-api.R b/R/utils-api.R index 775e376..76129ea 100644 --- a/R/utils-api.R +++ b/R/utils-api.R @@ -97,8 +97,23 @@ api_success <- function(status, ...) { } #' @rdname api_helpers #' @export +.openeocraft_default_api_base_url <- function() { + env_host <- Sys.getenv("OPENEOCRAFT_API_BASE_URL", unset = "") + if (nzchar(env_host)) { + return(env_host) + } + if (file.exists("/.dockerenv")) { + return("http://127.0.0.1:8000") + } + NULL +} + get_host <- function(api, req) { host <- api_attr(api, "api_base_url") + if (!is.null(host) && nzchar(host)) { + return(host) + } + host <- .openeocraft_default_api_base_url() if (!is.null(host)) { return(host) } diff --git a/R/zzz.R b/R/zzz.R new file mode 100644 index 0000000..484276d --- /dev/null +++ b/R/zzz.R @@ -0,0 +1,3 @@ +.onLoad <- function(libname, pkgname) { + configure_openeocraft_runtime() +} diff --git a/docker-compose.yaml b/docker-compose.yaml index 0dfcda1..5649a6a 100644 --- a/docker-compose.yaml +++ b/docker-compose.yaml @@ -1,12 +1,30 @@ services: openeocraft: + platform: linux/amd64 # Required for torch - no ARM64 Linux binaries available build: context: . dockerfile: docker/Dockerfile environment: - - TZ=Etc/UTC - - DEBIAN_FRONTEND=noninteractive + TZ: Etc/UTC + DEBIAN_FRONTEND: noninteractive + OMP_NUM_THREADS: "1" + MKL_NUM_THREADS: "1" + OPENBLAS_NUM_THREADS: "1" + TORCH_NUM_THREADS: "1" + # sits fork workers crash in Docker; keep sequential (see .openeocraft_multicores) + OPENEOCRAFT_MULTICORES_MAX: "1" + OPENEOCRAFT_SKIP_CUBE_COPY: "true" + OPENEOCRAFT_API_BASE_URL: "http://127.0.0.1:8000" + OPENEOCRAFT_MEMSIZE: "16" + OPENEOCRAFT_RESOURCE_FRACTION: "0.5" + cpus: "8" + mem_limit: 32g container_name: openeocraft ports: - "8000:8000" - restart: always + restart: unless-stopped + volumes: + - openeocraft-workspace:/var/openeo/workspace + +volumes: + openeocraft-workspace: diff --git a/docker/Dockerfile b/docker/Dockerfile index 7567646..f44bb21 100644 --- a/docker/Dockerfile +++ b/docker/Dockerfile @@ -5,6 +5,13 @@ FROM --platform=${BUILD_PLATFORM} r-base:4.5.0 ENV DEBIAN_FRONTEND=noninteractive ENV TZ=Etc/UTC ENV CXXFLAGS="-O2 -pipe -std=gnu++17" +ENV OMP_NUM_THREADS=1 +ENV MKL_NUM_THREADS=1 +ENV OPENBLAS_NUM_THREADS=1 +ENV TORCH_NUM_THREADS=1 +ENV OPENEOCRAFT_MULTICORES_MAX=4 +ENV OPENEOCRAFT_MEMSIZE=16 +ENV OPENEOCRAFT_RESOURCE_FRACTION=0.5 RUN apt-get update && apt-get install -y --no-install-recommends \ ca-certificates gnupg apt-transport-https \ @@ -18,6 +25,7 @@ RUN apt-get update && apt-get install -y --no-install-recommends \ libnetcdf-dev libcurl4-openssl-dev libcpprest-dev \ libsqlite3-dev libboost-all-dev \ libsodium-dev libudunits2-dev libuv1-dev \ + libharfbuzz-dev libfribidi-dev libfreetype6-dev \ zlib1g-dev libunwind-dev libssl-dev libxml2-dev \ doxygen graphviz \ && apt-get clean && rm -rf /var/lib/apt/lists/* @@ -38,7 +46,7 @@ RUN R -e "install.packages(c('remotes'), repos='https://cloud.r-project.org')" RUN R -e "install.packages(c('devtools'), repos='https://cloud.r-project.org')" # Install basic packages first (these are dependencies for others) -RUN R -e "install.packages(c('gdalcubes','plumber','useful','s2','sf','rstac','geojsonsf', 'jsonlite', 'base64enc', 'ids', 'callr' , 'sits'), repos='https://cloud.r-project.org')" +RUN R -e "install.packages(c('gdalcubes','plumber','useful','s2','sf','rstac','geojsonsf', 'jsonlite', 'base64enc', 'ids', 'callr' , 'sits', 'caret'), repos='https://cloud.r-project.org')" # Install torch's Lantern library (required for deep learning models like TempCNN). # TORCH_INSTALL=1 auto-confirms the interactive prompt; CUDA=cpu avoids GPU binary diff --git a/docker/docker-compose.yaml b/docker/docker-compose.yaml index 682bd35..c5b1df9 100644 --- a/docker/docker-compose.yaml +++ b/docker/docker-compose.yaml @@ -5,9 +5,25 @@ services: context: .. dockerfile: docker/Dockerfile environment: - - TZ=Etc/UTC - - DEBIAN_FRONTEND=noninteractive + TZ: Etc/UTC + DEBIAN_FRONTEND: noninteractive + OMP_NUM_THREADS: "1" + MKL_NUM_THREADS: "1" + OPENBLAS_NUM_THREADS: "1" + TORCH_NUM_THREADS: "1" + OPENEOCRAFT_MULTICORES_MAX: "1" + OPENEOCRAFT_SKIP_CUBE_COPY: "true" + OPENEOCRAFT_MEMSIZE: "16" + OPENEOCRAFT_RESOURCE_FRACTION: "0.5" + OPENEOCRAFT_API_BASE_URL: "http://127.0.0.1:8000" + cpus: "8" + mem_limit: 32g container_name: openeocraft ports: - "8000:8000" - restart: always + restart: unless-stopped + volumes: + - openeocraft-workspace:/var/openeo/workspace + +volumes: + openeocraft-workspace: diff --git a/docker/plumber.R b/docker/plumber.R index 1449236..9856a2b 100644 --- a/docker/plumber.R +++ b/docker/plumber.R @@ -66,17 +66,31 @@ file <- system.file("ml/db.rds", package = "openeocraft") stac_api <- openstac::set_db(stac_api, driver = "local", file = file) # Create openEO API object +work_dir <- if (file.exists("/.dockerenv")) { + "/var/openeo" +} else { + "~/openeo-tests" +} + api <- create_openeo_v1( id = "openeocraft", title = "openEO compliant R backend", description = "OpenEOcraft offers a robust R framework designed for the development and deployment of openEO API applications.", backend_version = "0.3.1", stac_api = stac_api, - work_dir = "~/openeo-tests", + work_dir = work_dir, conforms_to = NULL, production = FALSE ) +api_base_url <- Sys.getenv("OPENEOCRAFT_API_BASE_URL", unset = "") +if (!nzchar(api_base_url) && file.exists("/.dockerenv")) { + api_base_url <- "http://127.0.0.1:8000" +} +if (nzchar(api_base_url)) { + assign("api_base_url", api_base_url, envir = attr(api, "env")) +} + # set_credentials(api, file = "~/openeo-tests/openeo-credentials.rds") set_credentials(api, file = "~/openeo-credentials.rds") new_credential(api, user = "rolf", password = "123456") diff --git a/docker/server.R b/docker/server.R index cd3ae49..103c668 100644 --- a/docker/server.R +++ b/docker/server.R @@ -3,6 +3,35 @@ # Must be set here, before library() calls, for it to take effect in time. Sys.setenv(TORCH_INSTALL = "1") +# Limit BLAS/torch threads before any worker forks (sits_regularize uses multicore). +Sys.setenv( + OMP_NUM_THREADS = Sys.getenv("OMP_NUM_THREADS", "1"), + MKL_NUM_THREADS = Sys.getenv("MKL_NUM_THREADS", "1"), + OPENBLAS_NUM_THREADS = Sys.getenv("OPENBLAS_NUM_THREADS", "1"), + TORCH_NUM_THREADS = Sys.getenv("TORCH_NUM_THREADS", "1") +) + +# Plumber parent only; callr job workers pick up OPENEOCRAFT_* via processes.R + .onLoad. +if (file.exists("/.dockerenv")) { + options( + openeocraft.resource_fraction = as.numeric( + Sys.getenv("OPENEOCRAFT_RESOURCE_FRACTION", "0.5") + ), + openeocraft.multicores_max = as.integer( + Sys.getenv("OPENEOCRAFT_MULTICORES_MAX", "4") + ), + openeocraft.memsize = as.integer(Sys.getenv("OPENEOCRAFT_MEMSIZE", "16")), + openeocraft.memsize_auto = FALSE + ) + message( + "[startup] Docker resource limits: multicores_max=", + getOption("openeocraft.multicores_max"), + ", memsize=", + getOption("openeocraft.memsize"), + " GB" + ) +} + # TO-DO: show R errors into terminal # Determine plumber file path - works both locally and in Docker docker_path <- "/opt/dockerfiles/docker/plumber.R" diff --git a/docker/verify-r-deps.R b/docker/verify-r-deps.R index 9625524..a101fb9 100644 --- a/docker/verify-r-deps.R +++ b/docker/verify-r-deps.R @@ -4,7 +4,7 @@ pkgs <- c( "gdalcubes", "plumber", "useful", "s2", "sf", "rstac", "geojsonsf", - "jsonlite", "base64enc", "ids", "callr", "sits", + "jsonlite", "base64enc", "ids", "callr", "sits", "caret", "torch", "openstac", "openeocraft", "swagger" diff --git a/inst/demo-lps-2025/01_ml_api_eo_data_cubes.ipynb b/inst/demo-lps-2025/01_ml_api_eo_data_cubes.ipynb index 53c973c..e6766fb 100644 --- a/inst/demo-lps-2025/01_ml_api_eo_data_cubes.ipynb +++ b/inst/demo-lps-2025/01_ml_api_eo_data_cubes.ipynb @@ -22,7 +22,7 @@ }, { "cell_type": "code", - "execution_count": 23, + "execution_count": 1, "id": "3843b5af", "metadata": {}, "outputs": [], @@ -32,7 +32,7 @@ }, { "cell_type": "code", - "execution_count": 24, + "execution_count": 2, "id": "71be7901", "metadata": {}, "outputs": [], @@ -42,7 +42,7 @@ }, { "cell_type": "code", - "execution_count": 25, + "execution_count": 3, "id": "91320354", "metadata": {}, "outputs": [ @@ -52,7 +52,7 @@ "" ] }, - "execution_count": 25, + "execution_count": 3, "metadata": {}, "output_type": "execute_result" } @@ -63,7 +63,7 @@ }, { "cell_type": "code", - "execution_count": 26, + "execution_count": 4, "id": "5fb3d8b7", "metadata": {}, "outputs": [ @@ -79,7 +79,7 @@ " 'cdse-sentinel-2-l2a']" ] }, - "execution_count": 26, + "execution_count": 4, "metadata": {}, "output_type": "execute_result" } @@ -98,7 +98,7 @@ }, { "cell_type": "code", - "execution_count": 27, + "execution_count": 5, "id": "b538a852", "metadata": {}, "outputs": [ @@ -182,7 +182,7 @@ " 'cloud']}}}" ] }, - "execution_count": 27, + "execution_count": 5, "metadata": {}, "output_type": "execute_result" } @@ -193,7 +193,7 @@ }, { "cell_type": "code", - "execution_count": 28, + "execution_count": 6, "id": "5908f1f7", "metadata": {}, "outputs": [ @@ -277,7 +277,7 @@ " 'cloud']}}}" ] }, - "execution_count": 28, + "execution_count": 6, "metadata": {}, "output_type": "execute_result" } @@ -298,7 +298,7 @@ }, { "cell_type": "code", - "execution_count": 29, + "execution_count": 7, "id": "bdefeff4", "metadata": {}, "outputs": [ @@ -361,7 +361,7 @@ }, { "cell_type": "code", - "execution_count": 30, + "execution_count": 8, "id": "f9a9acb5", "metadata": {}, "outputs": [ @@ -416,7 +416,7 @@ " 'rel': 'about'}]}" ] }, - "execution_count": 30, + "execution_count": 8, "metadata": {}, "output_type": "execute_result" } @@ -427,7 +427,7 @@ }, { "cell_type": "code", - "execution_count": 31, + "execution_count": 9, "id": "9e1c9a33", "metadata": {}, "outputs": [ @@ -558,7 +558,7 @@ " 'rel': 'about'}]}" ] }, - "execution_count": 31, + "execution_count": 9, "metadata": {}, "output_type": "execute_result" } @@ -594,7 +594,7 @@ }, { "cell_type": "code", - "execution_count": 32, + "execution_count": 10, "id": "3b8e084b", "metadata": {}, "outputs": [], @@ -610,7 +610,7 @@ }, { "cell_type": "code", - "execution_count": 33, + "execution_count": 11, "id": "db29bbdf", "metadata": {}, "outputs": [], @@ -637,7 +637,7 @@ }, { "cell_type": "code", - "execution_count": 34, + "execution_count": 12, "id": "cc555722", "metadata": {}, "outputs": [], @@ -662,7 +662,7 @@ }, { "cell_type": "code", - "execution_count": 35, + "execution_count": 13, "id": "823e6ba9", "metadata": {}, "outputs": [], @@ -690,7 +690,7 @@ }, { "cell_type": "code", - "execution_count": 36, + "execution_count": 14, "id": "91a5405a", "metadata": {}, "outputs": [], @@ -713,7 +713,7 @@ }, { "cell_type": "code", - "execution_count": 37, + "execution_count": 15, "id": "652b6cb2", "metadata": {}, "outputs": [], @@ -733,7 +733,7 @@ }, { "cell_type": "code", - "execution_count": 38, + "execution_count": 16, "id": "7bfa9f13", "metadata": {}, "outputs": [], @@ -753,7 +753,7 @@ }, { "cell_type": "code", - "execution_count": 39, + "execution_count": 17, "id": "d7c4de9a", "metadata": {}, "outputs": [], @@ -765,7 +765,7 @@ }, { "cell_type": "code", - "execution_count": 40, + "execution_count": 18, "id": "d56dbc8a", "metadata": {}, "outputs": [ @@ -788,15 +788,15 @@ " }\n", " \n", " \n", - " \n", + " \n", " \n", " " ], "text/plain": [ - "" + "" ] }, - "execution_count": 40, + "execution_count": 18, "metadata": {}, "output_type": "execute_result" } @@ -807,7 +807,7 @@ }, { "cell_type": "code", - "execution_count": 41, + "execution_count": 19, "id": "8ca118fb", "metadata": {}, "outputs": [], @@ -820,7 +820,7 @@ }, { "cell_type": "code", - "execution_count": 42, + "execution_count": 20, "id": "57b33b17", "metadata": {}, "outputs": [ @@ -828,21 +828,24 @@ "name": "stdout", "output_type": "stream", "text": [ - "0:00:00 Job '92958c2545b379cc79335e4f42c8822f': send 'start'\n", - "0:00:04 Job '92958c2545b379cc79335e4f42c8822f': running (progress N/A)\n", - "0:00:09 Job '92958c2545b379cc79335e4f42c8822f': running (progress N/A)\n", - "0:00:15 Job '92958c2545b379cc79335e4f42c8822f': running (progress N/A)\n", - "0:00:23 Job '92958c2545b379cc79335e4f42c8822f': running (progress N/A)\n", - "0:00:32 Job '92958c2545b379cc79335e4f42c8822f': running (progress N/A)\n", - "0:00:45 Job '92958c2545b379cc79335e4f42c8822f': running (progress N/A)\n", - "0:01:00 Job '92958c2545b379cc79335e4f42c8822f': running (progress N/A)\n", - "0:01:19 Job '92958c2545b379cc79335e4f42c8822f': running (progress N/A)\n", - "0:01:43 Job '92958c2545b379cc79335e4f42c8822f': running (progress N/A)\n", - "0:02:13 Job '92958c2545b379cc79335e4f42c8822f': running (progress N/A)\n", - "0:02:50 Job '92958c2545b379cc79335e4f42c8822f': running (progress N/A)\n", - "0:03:37 Job '92958c2545b379cc79335e4f42c8822f': running (progress N/A)\n", - "0:04:35 Job '92958c2545b379cc79335e4f42c8822f': running (progress N/A)\n", - "0:05:35 Job '92958c2545b379cc79335e4f42c8822f': finished (progress N/A)\n" + "0:00:00 Job '7a28728bc7000037448db8e09ee5e7e4': send 'start'\n", + "0:00:05 Job '7a28728bc7000037448db8e09ee5e7e4': running (progress N/A)\n", + "0:00:10 Job '7a28728bc7000037448db8e09ee5e7e4': running (progress N/A)\n", + "0:00:16 Job '7a28728bc7000037448db8e09ee5e7e4': running (progress N/A)\n", + "0:00:24 Job '7a28728bc7000037448db8e09ee5e7e4': running (progress N/A)\n", + "0:00:34 Job '7a28728bc7000037448db8e09ee5e7e4': running (progress N/A)\n", + "0:00:46 Job '7a28728bc7000037448db8e09ee5e7e4': running (progress N/A)\n", + "0:01:01 Job '7a28728bc7000037448db8e09ee5e7e4': running (progress N/A)\n", + "0:01:21 Job '7a28728bc7000037448db8e09ee5e7e4': running (progress N/A)\n", + "0:01:44 Job '7a28728bc7000037448db8e09ee5e7e4': running (progress N/A)\n", + "0:02:14 Job '7a28728bc7000037448db8e09ee5e7e4': running (progress N/A)\n", + "0:02:52 Job '7a28728bc7000037448db8e09ee5e7e4': running (progress N/A)\n", + "0:03:38 Job '7a28728bc7000037448db8e09ee5e7e4': running (progress N/A)\n", + "0:04:36 Job '7a28728bc7000037448db8e09ee5e7e4': running (progress N/A)\n", + "0:05:37 Job '7a28728bc7000037448db8e09ee5e7e4': running (progress N/A)\n", + "0:06:37 Job '7a28728bc7000037448db8e09ee5e7e4': running (progress N/A)\n", + "0:07:37 Job '7a28728bc7000037448db8e09ee5e7e4': running (progress N/A)\n", + "0:08:37 Job '7a28728bc7000037448db8e09ee5e7e4': finished (progress N/A)\n" ] } ], @@ -853,7 +856,7 @@ }, { "cell_type": "code", - "execution_count": 43, + "execution_count": 21, "id": "4a07e0f2", "metadata": {}, "outputs": [ @@ -864,7 +867,7 @@ " PosixPath('data/outputs/job-results.json')]" ] }, - "execution_count": 43, + "execution_count": 21, "metadata": {}, "output_type": "execute_result" } diff --git a/inst/demo-lps-2025/05_ml_tuning_random.ipynb b/inst/demo-lps-2025/05_ml_tuning_random.ipynb index 2849159..6261fbf 100644 --- a/inst/demo-lps-2025/05_ml_tuning_random.ipynb +++ b/inst/demo-lps-2025/05_ml_tuning_random.ipynb @@ -118,35 +118,36 @@ "name": "stdout", "output_type": "stream", "text": [ - "0:00:00 Job '312cf908482d1a71c09904414d0e8047': send 'start'\n", - "0:00:04 Job '312cf908482d1a71c09904414d0e8047': running (progress N/A)\n", - "0:00:09 Job '312cf908482d1a71c09904414d0e8047': running (progress N/A)\n", - "0:00:15 Job '312cf908482d1a71c09904414d0e8047': running (progress N/A)\n", - "0:00:23 Job '312cf908482d1a71c09904414d0e8047': running (progress N/A)\n", - "0:00:33 Job '312cf908482d1a71c09904414d0e8047': running (progress N/A)\n", - "0:00:45 Job '312cf908482d1a71c09904414d0e8047': running (progress N/A)\n", - "0:01:01 Job '312cf908482d1a71c09904414d0e8047': running (progress N/A)\n", - "0:01:20 Job '312cf908482d1a71c09904414d0e8047': running (progress N/A)\n", - "0:01:43 Job '312cf908482d1a71c09904414d0e8047': running (progress N/A)\n", - "0:02:13 Job '312cf908482d1a71c09904414d0e8047': running (progress N/A)\n", - "0:02:51 Job '312cf908482d1a71c09904414d0e8047': running (progress N/A)\n", - "0:03:37 Job '312cf908482d1a71c09904414d0e8047': running (progress N/A)\n", - "0:04:35 Job '312cf908482d1a71c09904414d0e8047': running (progress N/A)\n", - "0:05:35 Job '312cf908482d1a71c09904414d0e8047': running (progress N/A)\n", - "0:06:35 Job '312cf908482d1a71c09904414d0e8047': running (progress N/A)\n", - "0:07:35 Job '312cf908482d1a71c09904414d0e8047': running (progress N/A)\n", - "0:08:35 Job '312cf908482d1a71c09904414d0e8047': running (progress N/A)\n", - "0:09:35 Job '312cf908482d1a71c09904414d0e8047': running (progress N/A)\n", - "0:10:35 Job '312cf908482d1a71c09904414d0e8047': running (progress N/A)\n", - "0:11:35 Job '312cf908482d1a71c09904414d0e8047': running (progress N/A)\n", - "0:12:35 Job '312cf908482d1a71c09904414d0e8047': running (progress N/A)\n", - "0:13:35 Job '312cf908482d1a71c09904414d0e8047': running (progress N/A)\n", - "0:14:35 Job '312cf908482d1a71c09904414d0e8047': running (progress N/A)\n", - "0:15:35 Job '312cf908482d1a71c09904414d0e8047': running (progress N/A)\n", - "0:16:35 Job '312cf908482d1a71c09904414d0e8047': running (progress N/A)\n", - "0:17:35 Job '312cf908482d1a71c09904414d0e8047': running (progress N/A)\n", - "0:18:36 Job '312cf908482d1a71c09904414d0e8047': running (progress N/A)\n", - "0:19:36 Job '312cf908482d1a71c09904414d0e8047': finished (progress N/A)\n" + "0:00:00 Job '1bcf2080950cee5adfd551623f36c6e8': send 'start'\n", + "0:00:05 Job '1bcf2080950cee5adfd551623f36c6e8': running (progress N/A)\n", + "0:00:10 Job '1bcf2080950cee5adfd551623f36c6e8': running (progress N/A)\n", + "0:00:16 Job '1bcf2080950cee5adfd551623f36c6e8': running (progress N/A)\n", + "0:00:24 Job '1bcf2080950cee5adfd551623f36c6e8': running (progress N/A)\n", + "0:00:34 Job '1bcf2080950cee5adfd551623f36c6e8': running (progress N/A)\n", + "0:00:46 Job '1bcf2080950cee5adfd551623f36c6e8': running (progress N/A)\n", + "0:01:02 Job '1bcf2080950cee5adfd551623f36c6e8': running (progress N/A)\n", + "0:01:21 Job '1bcf2080950cee5adfd551623f36c6e8': running (progress N/A)\n", + "0:01:45 Job '1bcf2080950cee5adfd551623f36c6e8': running (progress N/A)\n", + "0:02:14 Job '1bcf2080950cee5adfd551623f36c6e8': running (progress N/A)\n", + "0:02:52 Job '1bcf2080950cee5adfd551623f36c6e8': running (progress N/A)\n", + "0:03:38 Job '1bcf2080950cee5adfd551623f36c6e8': running (progress N/A)\n", + "0:04:36 Job '1bcf2080950cee5adfd551623f36c6e8': running (progress N/A)\n", + "0:05:36 Job '1bcf2080950cee5adfd551623f36c6e8': running (progress N/A)\n", + "0:06:36 Job '1bcf2080950cee5adfd551623f36c6e8': running (progress N/A)\n", + "0:07:36 Job '1bcf2080950cee5adfd551623f36c6e8': running (progress N/A)\n", + "0:08:36 Job '1bcf2080950cee5adfd551623f36c6e8': running (progress N/A)\n", + "0:09:36 Job '1bcf2080950cee5adfd551623f36c6e8': running (progress N/A)\n", + "0:10:36 Job '1bcf2080950cee5adfd551623f36c6e8': running (progress N/A)\n", + "0:11:36 Job '1bcf2080950cee5adfd551623f36c6e8': running (progress N/A)\n", + "0:12:36 Job '1bcf2080950cee5adfd551623f36c6e8': running (progress N/A)\n", + "0:13:36 Job '1bcf2080950cee5adfd551623f36c6e8': running (progress N/A)\n", + "0:14:37 Job '1bcf2080950cee5adfd551623f36c6e8': running (progress N/A)\n", + "0:15:37 Job '1bcf2080950cee5adfd551623f36c6e8': running (progress N/A)\n", + "0:16:37 Job '1bcf2080950cee5adfd551623f36c6e8': running (progress N/A)\n", + "0:17:37 Job '1bcf2080950cee5adfd551623f36c6e8': running (progress N/A)\n", + "0:18:37 Job '1bcf2080950cee5adfd551623f36c6e8': running (progress N/A)\n", + "0:19:37 Job '1bcf2080950cee5adfd551623f36c6e8': running (progress N/A)\n", + "0:20:37 Job '1bcf2080950cee5adfd551623f36c6e8': finished (progress N/A)\n" ] } ], @@ -243,8 +244,8 @@ "name": "stdout", "output_type": "stream", "text": [ - "Job id: 312cf908482d1a71c09904414d0e8047\n", - "Model file: http://127.0.0.1:8000/files/jobs/312cf908482d1a71c09904414d0e8047/models/tempcnn-tuned-random.rds\n" + "Job id: 1bcf2080950cee5adfd551623f36c6e8\n", + "Model file: http://127.0.0.1:8000/files/jobs/1bcf2080950cee5adfd551623f36c6e8/models/tempcnn-tuned-random.rds\n" ] } ], diff --git a/inst/demo-lps-2025/data/outputs/SENTINEL-2_MSI_20LMR_class_2022-01-05.tif b/inst/demo-lps-2025/data/outputs/SENTINEL-2_MSI_20LMR_class_2022-01-05.tif index 8de540c165209c9b5adc08090fe0455f24ce04e2..a1640ab229988b9b7a1d01805c035498611568a1 100644 GIT binary patch delta 686 zcmV;f0#W_n3d9MZ0ST4@00000002{wkrT5V0T=;)G!y_1M%pR0)X=`qx_h5MBtcK zJOiVDeB6IAVu}L@#fVePBSxI0pv_6)q@Za;kS0bvM0kV`6Z2tr0f6SH;UzaCNs=@N z00R*Tg@_&!0C6xPi((`QDMW}Nf+93TK!|Wc1VliD>$U^im%YWaj2S1Jv84x|v9?hI zFjXF4@9OYQ$GZb`9y5sS=y?e(+X14F2Uuo*iUY`(svodOK)k}dO|sR@5*j=u(0EUZ z9^mbULG>3A<|~>W5ITV&V7Nr(M$bl$qrd7@<^fdOZlJCV7{OvJy2c{N2^I4Lmc{4A z-r{I)LQVr?FdtwNMZRA}2kV^Ea&1xV^s!JwgK>8eUQ-_6voGPK|LPu4+X|@GUn>hN+5;HUcuT;-oLdg_ z<=x;G1|t9%a{fGZU!MaC%X$D%^#YQAxi&n&7$&P)D&j0l-Q1fZ;y#7o)osc>OJ9 UmOr`pRzFSfKn4N;5C8!H0G~i7D*ylh delta 842 zcmV-Q1GW6b3Ev8!0SPe!00000002{wkrT5V0T=;)3>g5%<31XoB`GupKij|Q|NsC0 z|NsC0{}7SA|Np=36fXch05t$}BB5{9$wY)Ob#AkHU$8D zWi0vv9w^6{&1QiBp2S5Ll7WCqI6L|=D2SQnjtLf1Dvye z>--unNO^!ZeCV&_0lZcL5bYo(Zv2;uo-*VC^;Q^G03a3olJT&N>zf;8 z@%3N-$_LItFbgGXK$PODj0h;%WDG^WC9Ed{!I@Q2wwA7*kM7-KV(v&Y2%g z_It;SfA6CQFc@<0xq)tI$0kHkVMM0{>aBZ6$^&*%e7vij07eiv&pd$R1a$I$N5c16 zF}>OUBLFm+`1*!Z*(4CqN&t)H0Y?J_2mveV8UIrrWPqt*Byg`|^5Mr+r#yg@fUGI$ z6Q~UbNH>!Qn1umj8E{C+^M?%d0IL`(x5Pxt1F%AQaP$&@(JFdCJYfU~DsMdxSW?j} zIY4H1#X%s;4PdZD^MAj7f1C= 1) { + spatial_extent <- extent_full +} else { + lon_c <- (extent_full$west + extent_full$east) / 2 + lat_c <- (extent_full$south + extent_full$north) / 2 + lon_span <- extent_full$east - extent_full$west + lat_span <- extent_full$north - extent_full$south + k <- sqrt(area_frac) + spatial_extent <- list( + west = lon_c - lon_span * k / 2, + east = lon_c + lon_span * k / 2, + south = lat_c - lat_span * k / 2, + north = lat_c + lat_span * k / 2 + ) +} + +temporal_start <- Sys.getenv("UC2_TEMPORAL_START", unset = "2022-01-01") +temporal_end <- Sys.getenv("UC2_TEMPORAL_END", unset = "2022-12-31") +if (!nzchar(temporal_start)) { + temporal_start <- "2022-01-01" +} +if (!nzchar(temporal_end)) { + temporal_end <- "2022-12-31" +} +temporal_extent <- c(temporal_start, temporal_end) +cube_period <- Sys.getenv("UC2_REGULARIZE_PERIOD", unset = "P16D") +if (!nzchar(cube_period)) { + cube_period <- "P16D" +} + +res_raw <- Sys.getenv("UC2_GRID_RESOLUTION", unset = "300") +if (!nzchar(res_raw)) { + res_raw <- "300" +} +resolution <- suppressWarnings(as.numeric(res_raw)) +if (!is.finite(resolution) || resolution <= 0) { + resolution <- 300 +} + +message(sprintf( + "Inference cube: area fraction=%s, west=%.6f east=%.6f south=%.6f north=%.6f, temporal=%s..%s, period=%s, resolution=%gm", + format(area_frac, scientific = FALSE, digits = 10), + spatial_extent$west, + spatial_extent$east, + spatial_extent$south, + spatial_extent$north, + temporal_extent[[1L]], + temporal_extent[[2L]], + cube_period, + resolution +)) + +band_spec <- trimws(Sys.getenv("UC2_COLLECTION_BANDS", unset = "")) +load_args <- list( + id = "mpc-sentinel-2-l2a", + spatial_extent = spatial_extent, + temporal_extent = temporal_extent +) +if (nzchar(band_spec)) { + band_list <- trimws(strsplit(band_spec, ",", fixed = TRUE)[[1]]) + band_list <- band_list[nzchar(band_list)] + if (length(band_list)) { + load_args$bands <- band_list + } +} +if (is.null(load_args$bands)) { + load_args$bands <- list( + "B02", "B03", "B04", "B05", "B06", "B07", "B08", + "B11", "B12", "B8A" + ) +} +datacube <- do.call(p$load_collection, load_args) + +datacube <- p$cube_regularize(data = datacube, period = cube_period, resolution = resolution) +datacube <- p$ndvi(data = datacube, red = "B04", nir = "B08", target_band = "NDVI") + +message("Running inference (ml_predict)...") +data <- p$ml_predict(data = datacube, model = tempcnn_model) + +ml_job <- p$save_result(data = data, format = "GTiff") +job <- create_job( + graph = ml_job, + title = "TempCNN fine-tuning + inference", + con = connection +) +job <- start_job(job, con = connection) + +writeLines(as.character(job$id), file.path(output_dir, "last_job_id.txt")) +append_poll_log(sprintf("Submitted job id=%s log=%s", job$id, get(".poll_log_path", envir = .GlobalEnv))) + +message("Submitted job successfully:") +print(job) + +wait_until_job_terminal( + job = job, + con = connection, + poll_sec = poll_seconds, + max_sec = max_wait_seconds +) + +dest <- download_job_results(job = job, con = connection, output_dir = output_dir) +append_poll_log(sprintf("Done. Results in %s", dest)) +message("Done.") diff --git a/inst/examples/download_job_results.R b/inst/examples/download_job_results.R new file mode 100644 index 0000000..0f474cf --- /dev/null +++ b/inst/examples/download_job_results.R @@ -0,0 +1,59 @@ +#!/usr/bin/env Rscript +# Poll a submitted job (optional) and download results to OUTPUT_DIR. +# +# Usage: +# export OPENEO_HOST=http://127.0.0.1:8000 OPENEO_USER=brian OPENEO_PASSWORD=... +# export OPENEO_JOB_ID= # or rely on OUTPUT_DIR/last_job_id.txt +# export OUTPUT_DIR=/tmp/openeocraft-docker-results +# Rscript inst/examples/download_job_results.R +# +# Set WAIT_FOR_JOB=false to skip polling (job already finished). + +library(openeo) + +args <- commandArgs(trailingOnly = FALSE) +file_arg <- grep("^--file=", args, value = TRUE) +if (length(file_arg)) { + script_dir <- dirname(sub("^--file=", "", file_arg[[1L]])) + helpers <- file.path(script_dir, "uc2_job_client.R") +} else { + helpers <- "inst/examples/uc2_job_client.R" +} +source(helpers, local = TRUE) + +host <- normalize_host(require_env("OPENEO_HOST")) +user <- require_env("OPENEO_USER") +password <- require_env("OPENEO_PASSWORD") +output_dir <- Sys.getenv("OUTPUT_DIR", unset = "/tmp/openeocraft-docker-results") +dir.create(output_dir, recursive = TRUE, showWarnings = FALSE) +init_poll_log(output_dir) + +poll_seconds <- as.numeric(Sys.getenv("JOB_POLL_SECONDS", unset = "30")) +max_wait_seconds <- as.numeric(Sys.getenv("JOB_MAX_WAIT_SECONDS", unset = "172800")) +if (!is.finite(poll_seconds) || poll_seconds < 1) { + poll_seconds <- 30 +} +if (!is.finite(max_wait_seconds) || max_wait_seconds < poll_seconds) { + max_wait_seconds <- 172800 +} + +wait_for_job <- tolower(Sys.getenv("WAIT_FOR_JOB", unset = "true")) +wait_for_job <- wait_for_job %in% c("1", "true", "yes", "y") + +connection <- connect_with_retry(host = host, user = user, password = password) +job <- resolve_job(con = connection) + +append_poll_log(sprintf("Resolved job %s (wait=%s)", job$id, wait_for_job)) + +if (wait_for_job) { + wait_until_job_terminal( + job = job, + con = connection, + poll_sec = poll_seconds, + max_sec = max_wait_seconds + ) +} + +dest <- download_job_results(job = job, con = connection, output_dir = output_dir) +append_poll_log("Done.") +message("Results in: ", dest) diff --git a/inst/examples/uc2_job_client.R b/inst/examples/uc2_job_client.R new file mode 100644 index 0000000..bf13f71 --- /dev/null +++ b/inst/examples/uc2_job_client.R @@ -0,0 +1,156 @@ +# Shared helpers for UC2 example scripts: poll logging and result download. + +require_env <- function(name) { + value <- Sys.getenv(name, unset = "") + if (!nzchar(value)) { + stop(sprintf("Missing required environment variable: %s", name), call. = FALSE) + } + value +} + +normalize_host <- function(raw_host) { + if (grepl("^https?://", raw_host)) { + raw_host + } else { + paste0("http://", raw_host) + } +} + +connect_with_retry <- function(host, user, password, attempts = 12, sleep_seconds = 5) { + last_error <- NULL + for (i in seq_len(attempts)) { + message(sprintf("Connection attempt %d/%d", i, attempts)) + trial <- try({ + con <- connect(host = host) + login(con = con, user = user, password = password) + }, silent = TRUE) + if (!inherits(trial, "try-error")) { + return(trial) + } + last_error <- trial + if (i < attempts) { + Sys.sleep(sleep_seconds) + } + } + stop( + sprintf( + "Failed to connect/login to backend '%s' after %d attempts. Last error: %s", + host, attempts, as.character(last_error) + ), + call. = FALSE + ) +} + +job_status <- function(job, con) { + info <- describe_job(job, con = con) + status_raw <- info$status + if (is.null(status_raw)) "" else tolower(as.character(status_raw)) +} + +init_poll_log <- function(output_dir) { + poll_log <- Sys.getenv("JOB_POLL_LOG", unset = "") + if (!nzchar(poll_log)) { + poll_log <- file.path(output_dir, "job_poll.log") + } + dir.create(dirname(poll_log), recursive = TRUE, showWarnings = FALSE) + assign(".poll_log_path", poll_log, envir = .GlobalEnv) + invisible(poll_log) +} + +append_poll_log <- function(msg) { + poll_log <- get(".poll_log_path", envir = .GlobalEnv) + line <- sprintf("%s %s", format(Sys.time(), "%Y-%m-%d %H:%M:%S"), msg) + cat(line, "\n", file = poll_log, append = TRUE) + message(msg) + flush.console() +} + +wait_until_job_terminal <- function(job, con, poll_sec, max_sec) { + start <- Sys.time() + repeat { + status <- job_status(job, con = con) + elapsed <- as.numeric(difftime(Sys.time(), start, units = "secs")) + append_poll_log(sprintf("[%.0fs] Job %s status: %s", elapsed, job$id, status)) + + if (status %in% c("finished", "completed", "done")) { + return(invisible(TRUE)) + } + if (status %in% c("error", "failed")) { + stop(sprintf("Job failed with status: %s", status), call. = FALSE) + } + if (status %in% c("canceled", "cancelled")) { + stop(sprintf("Job was canceled: %s", status), call. = FALSE) + } + if (elapsed > max_sec) { + stop( + sprintf( + "Timed out after %.0fs waiting for job (max JOB_MAX_WAIT_SECONDS=%s)", + elapsed, + max_sec + ), + call. = FALSE + ) + } + Sys.sleep(poll_sec) + } +} + +resolve_job <- function(con, job_id = NULL) { + if (is.null(job_id) || !nzchar(job_id)) { + job_id <- Sys.getenv("OPENEO_JOB_ID", unset = "") + } + if (!nzchar(job_id)) { + last_file <- Sys.getenv("JOB_ID_FILE", unset = "") + if (!nzchar(last_file)) { + out <- Sys.getenv("OUTPUT_DIR", unset = "/tmp/openeocraft-docker-results") + last_file <- file.path(out, "last_job_id.txt") + } + if (file.exists(last_file)) { + job_id <- trimws(readLines(last_file, n = 1L, warn = FALSE)) + } + } + if (!nzchar(job_id)) { + stop("Set OPENEO_JOB_ID or provide last_job_id.txt from a prior submit.", call. = FALSE) + } + jobs <- list_jobs(con = con) + job <- jobs[[job_id]] + if (is.null(job)) { + stop("Job not found: ", job_id, call. = FALSE) + } + job +} + +download_job_results <- function(job, con, output_dir) { + dir.create(output_dir, recursive = TRUE, showWarnings = FALSE) + dest <- file.path(output_dir, job$id) + dir.create(dest, recursive = TRUE, showWarnings = FALSE) + + api_ok <- tryCatch({ + download_results(job = job, folder = dest, con = con) + TRUE + }, error = function(e) { + append_poll_log(sprintf("download_results failed: %s", conditionMessage(e))) + FALSE + }) + if (api_ok) { + append_poll_log(sprintf("Downloaded via API to %s", dest)) + return(invisible(dest)) + } + + container <- Sys.getenv("OPENEO_DOCKER_CONTAINER", unset = "openeocraft") + user <- Sys.getenv("OPENEO_USER", unset = "brian") + job_dir <- sprintf("/var/openeo/workspace/%s/jobs/%s", user, job$id) + cmd <- sprintf( + "docker cp %s:%s/. %s/", + shQuote(container), + shQuote(job_dir), + shQuote(dest) + ) + append_poll_log(sprintf("Falling back to: %s", cmd)) + status <- system(cmd) + if (status != 0L) { + stop("Docker copy fallback failed with exit code ", status, call. = FALSE) + } + append_poll_log(sprintf("Downloaded via docker cp to %s", dest)) + invisible(dest) +} diff --git a/inst/ml/processes.R b/inst/ml/processes.R index 64c483e..7a58c72 100644 --- a/inst/ml/processes.R +++ b/inst/ml/processes.R @@ -22,6 +22,13 @@ # Numeric scalar in [0.05, 1], from getOption("openeocraft.resource_fraction") # or 0.75 when unset/invalid. .openeocraft_resource_fraction <- function() { + env_frac <- base::Sys.getenv("OPENEOCRAFT_RESOURCE_FRACTION", unset = "") + if (base::nzchar(env_frac)) { + f <- base::suppressWarnings(base::as.numeric(env_frac)) + if (base::is.finite(f)) { + return(base::min(1, base::max(0.05, f))) + } + } f <- base::getOption("openeocraft.resource_fraction", 0.75) if (!base::is.numeric(f) || base::length(f) != 1L || base::is.na(f)) { return(0.75) @@ -30,6 +37,20 @@ base::min(1, base::max(0.05, f)) } +.openeocraft_in_docker <- function() { + base::file.exists("/.dockerenv") +} + +.openeocraft_skip_cube_copy <- function() { + if (!.openeocraft_in_docker()) { + return(FALSE) + } + base::identical( + base::tolower(base::Sys.getenv("OPENEOCRAFT_SKIP_CUBE_COPY", "true")), + "true" + ) +} + # Total system RAM in gigabytes (best effort across OS APIs). # # Args: @@ -104,6 +125,11 @@ # Returns: # Positive integer; at least 1, at most getOption("openeocraft.multicores_max", 16L). .openeocraft_multicores <- function() { + # sits uses fork-based parallel workers; in Docker they often die (OOM / + # amd64 emulation) and leave zombie R while the job hangs at ~0% CPU. + if (.openeocraft_in_docker()) { + return(1L) + } frac <- .openeocraft_resource_fraction() cores <- base::tryCatch( parallel::detectCores(logical = FALSE), @@ -116,6 +142,13 @@ } usable <- base::max(1L, base::as.integer(base::floor(cores * frac))) cap <- base::getOption("openeocraft.multicores_max", 16L) + env_cap <- base::Sys.getenv("OPENEOCRAFT_MULTICORES_MAX", unset = "") + if (base::nzchar(env_cap)) { + env_cap <- base::suppressWarnings(base::as.integer(env_cap)) + if (!base::is.na(env_cap) && env_cap >= 1L) { + cap <- env_cap + } + } if (!base::is.numeric(cap) || base::length(cap) != 1L || base::is.na(cap)) { cap <- 16L } @@ -176,6 +209,13 @@ # Integer GB: explicit openeocraft.memsize, or auto from total RAM * fraction, # or 8 when auto is off or detection fails (clamped to [1, 256] when auto). .openeocraft_memsize <- function() { + env_mem <- base::Sys.getenv("OPENEOCRAFT_MEMSIZE", unset = "") + if (base::nzchar(env_mem)) { + mem <- base::suppressWarnings(base::as.integer(env_mem)) + if (!base::is.na(mem) && mem >= 1L) { + return(mem) + } + } explicit <- base::getOption("openeocraft.memsize", NULL) if (!base::is.null(explicit)) { mem <- explicit @@ -2763,21 +2803,27 @@ cube_regularize <- function(data, resolution, period) { "source" %in% base::names(data) && !base::all(data$source == "LOCAL") if (is_remote) { - raw_dir <- base::file.path(job_dir, "raw") - if (!base::dir.exists(raw_dir)) { - base::dir.create(raw_dir, recursive = TRUE) + if (!.openeocraft_skip_cube_copy()) { + raw_dir <- base::file.path(job_dir, "raw") + if (!base::dir.exists(raw_dir)) { + base::dir.create(raw_dir, recursive = TRUE) + } + base::message( + "[cube_regularize] sits_cube_copy(multicores=", mc, ") ..." + ) + data <- sits::sits_cube_copy( + cube = data, + roi = roi, + multicores = mc, + output_dir = raw_dir + ) + base::message("[cube_regularize] sits_cube_copy() returned") + .openeocraft_log_cube_diag(data, "After local copy (before regularize)") + } else { + base::message( + "[cube_regularize] Skipping sits_cube_copy (Docker / OPENEOCRAFT_SKIP_CUBE_COPY)" + ) } - base::message( - "[cube_regularize] sits_cube_copy(multicores=", mc, ") ..." - ) - data <- sits::sits_cube_copy( - cube = data, - roi = roi, - multicores = mc, - output_dir = raw_dir - ) - base::message("[cube_regularize] sits_cube_copy() returned") - .openeocraft_log_cube_diag(data, "After local copy (before regularize)") } .openeocraft_log_cube_diag(data, "Entering sits_regularize")