Fix Docker ML batch jobs and malformed result download URLs.

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

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)

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

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) {

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
+}

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)
     }

R/zzz.R

@@ -0,0 +1,3 @@
+.onLoad <- function(libname, pkgname) {
+    configure_openeocraft_runtime()
+}

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:

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

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:

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")