This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-7207-c37f6ab7c68f15e7a1aa283d0c29b84e11f1ed6d in repository https://gitbox.apache.org/repos/asf/texera.git
commit 2f77d5ab93f94c8e9b33195f805a35e63c52f970 Author: Kary Zheng <[email protected]> AuthorDate: Fri Aug 14 09:40:52 2026 +0000 test(workflow-operator): route Python tests to amber-integration (#7207) ### What changes were proposed in this PR? The `amber` job runs the `WorkflowOperator` test suite and installs no Python packages. A test that *executes* an operator's generated template, rather than only `py_compile`-ing it, therefore fails its dependency probe and cancels — and a cancellation is neither a pass nor a failure, so the suite still reports "All tests passed" and the gap leaves no trace in the build. This gives `WorkflowOperator` the split `amber` already has: - **Tag** — `IntegrationTest` under `common/workflow-operator/src/test`. amber's cannot be reused: it lives in `amber/src/test/integration`, and amber depends on this module rather than the reverse. - **Filter** — `common/workflow-operator/build.sbt` reads the same `AMBER_TEST_FILTER`. The `amber` job already sets it on the step that invokes `WorkflowOperator/jacoco`, so no workflow change is needed for the exclusion to take effect. - **Job** — `"WorkflowOperator/test"` added to `amber-integration`, which already runs `integration-only` with `amber/requirements.txt` and `amber/operator-requirements.txt` installed (pandas 2.2.3, plotly 5.24.1). - **Shared logic** — the env-var-to-ScalaTest-args mapping moves to `project/TestFilters.scala`, beside the build's other shared helpers (`AddMetaInfLicenseFiles.scala`, `JdkOptions.scala`); each module passes its own env var and tag, the two a caller has to match to the workflow and to the annotation. One behavior changes, and amber inherits it: a value that is neither of the two now fails at project load, where the inline code fell through to running everything. Nothing sets a third value, so what this catches is a typo that would otherwise leave a job running the whole suite while reporting the subset it asked for. - **First user** — `PythonCodeRawInvalidTextSpec` gains a tagged case asserting that pandas and plotly import in the interpreter it already resolves: `py_compile` only parses the emitted code, while running it needs the packages it imports. Being tagged, it also exercises the routing, and it turns a missing install in `amber-integration` into a failure rather than the silent cancellation above. Elsewhere a bare interpreter is a local-setup fact, so it cancels instead. The second half addresses the scaling problem in the same issue: testing operators one at a time does not scale when each one costs a spawn. - **Batching** — the `py_compile` check spawned `python -I -S -B -m py_compile` once per `PythonOperatorDescriptor`, 117 of them serially, where the interpreter boot is the entire cost and the compile is under a millisecond. It now goes through `PythonWorkerPool`: a worker launched once with the same `-I -S` isolation, serving many sources over a line-delimited JSON protocol. What the check accepts is unchanged — `compile(source, path, "exec")` is what `py_compile` does before writing a `.pyc`, and raises the same SyntaxError — and any worker the pool cannot give out or keep falls back to the spawn, so behavior is never worse than before: one that would not start, would not report ready, or died mid-job all leave through the same exception, and the check counts the descriptors that took the spawn into its summary so a run the pool served none of does not read as a green pooled run. `TEXERA_TEST_PYTHON_WORKER=0` selects the old path outright. - **Parallelism** — the descriptors are fanned out across the pool's workers, four at a time; the fan-out's executor is sized to the pool's cap, so nothing runs past it. Together: 1119 ms to 310 ms. That cap is per sub-pool, keyed by script, arguments and environment, and is env-overridable, so it is the fan-out's own sizing that bounds this — not a ScalaTest parameter, which no suite in the module currently parallelizes under anyway. - **Where the pool lives** — this module's test scope, not beside its caller. The tests this PR unblocks execute generated templates against pandas and plotly, where a spawn costs 260-310 ms, mostly the imports, dwarfing the ~4 ms a job computes for. Without a shared seam each such test hand-rolls a driver, a stdout protocol, a timeout and an interpreter probe — the runtime test in #7149 already does. Other modules reach it through a `test->test` dependency. ### Any related issues, documentation, discussions? Fixes #7186. Found while reviewing #7149, whose runtime test is the case cancelling today; tagging it is a one-line follow-up once this is in, and converting its hand-rolled driver to the pool is the natural second one. The pooled-worker design, and the measurements behind those figures, are in #6975. ### How was this PR tested? - `amber` is unchanged by the extraction: `show WorkflowExecutionService/Test/testOptions` under all three env values gives arguments byte-identical to its previous inline code — `-l <tag>`, `-n <tag>`, nothing when unset. `WorkflowOperator` was checked the same way and yields its own tag in those positions. - The new case, run as `WorkflowOperator/testOnly *PythonCodeRawInvalidTextSpec`: passes under `integration-only` (1 test selected); is excluded under `skip-integration` while the spec's two existing tests still run; and, pointed at a bare interpreter, fails under `integration-only` and cancels with no filter set. - The pooled check agrees with the spawn it replaces: 117/117 descriptors pass either way, and `TEXERA_TEST_PYTHON_WORKER=0` flips between them. Timed with `-oD`: 1119 ms spawning, 310 ms pooled. - Red-checked, since a green suite only proves the two paths agree on valid code: appending a syntax error to every generated source takes the pooled path to `ok=0/117` and fails the test, so it still detects what it is there to detect. - Each way a worker could escape the pool's borrow/return discipline was reproduced before being fixed, and `PythonWorkerPoolSpec` keeps a case on each: an interrupted caller used to leave a worker neither returned nor discarded, and the case that refills every slot and then asks for that many jobs again spins at the cap until its bound without the fix; a first line that is not the protocol used to raise `JsonParseException` past the caller's fallback, and the fixture now has a mode that writes one; an interpreter that cannot be started arrives as the pool's own exception rather than `ProcessBuilder`'s. - The fallback is never worse than the spawn it replaces, checked end to end: with the pooled path pointed at an interpreter that does not exist, all 117 descriptors fall back, all 117 still pass, and the summary reports 117 fallbacks. - `WorkflowOperator/test` under `skip-integration`: 286 suites, 2163 tests, none failed. - `sbt scalafmtCheckAll` is clean, and `WorkflowOperator`'s `Compile` and `Test` scalafix checks pass. ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Claude Opus 5) --------- Signed-off-by: Kary Zheng <[email protected]> Co-authored-by: Claude Opus 5 (1M context) <[email protected]> Co-authored-by: Xinyuan Lin <[email protected]> Co-authored-by: Yicong Huang <[email protected]> --- .github/workflows/build.yml | 13 +- amber/build.sbt | 20 +- common/workflow-operator/build.sbt | 13 + .../amber/operator/tags/IntegrationTest.java | 51 +++ .../src/test/resources/python/hanging_worker.py | 87 ++++ .../src/test/resources/python/py_compile_worker.py | 77 ++++ .../amber/util/PythonCodeRawInvalidTextSpec.scala | 161 ++++++- .../amber/util/python/PythonWorkerPool.scala | 503 +++++++++++++++++++++ .../amber/util/python/PythonWorkerPoolSpec.scala | 266 +++++++++++ project/TestFilters.scala | 48 ++ 10 files changed, 1212 insertions(+), 27 deletions(-) diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index c28457b198..120dd1f2aa 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -284,9 +284,13 @@ jobs: # Test config (sbt's `test` task does not transit dependsOn), # so common modules' tests are listed explicitly here. # - # AMBER_TEST_FILTER=skip-integration tells amber/build.sbt to - # exclude @org.apache.texera.amber.tags.IntegrationTest specs; - # those run in the amber-integration job below. + # AMBER_TEST_FILTER=skip-integration excludes the integration-tagged + # tests of two modules, each under its own tag: amber/build.sbt drops + # @org.apache.texera.amber.tags.IntegrationTest, and + # common/workflow-operator/build.sbt drops + # @org.apache.texera.amber.operator.tags.IntegrationTest. Both run in the + # amber-integration job below. The second exclusion is what keeps the + # pandas and plotly probe out of this job, which installs neither. env: AMBER_TEST_FILTER: skip-integration # unit job uses its provisioned postgres catalog; default (rest) needs a Lakekeeper not run here @@ -631,6 +635,8 @@ jobs: # specs. The Java @TagAnnotation makes the marker visible to # ScalaTest's reflection, so `-n TAG` correctly narrows the # run. + # WorkflowOperator/test is here for the same reason, with its own tag: + # its Python-dependent tests need packages the `amber` job lacks. # # scalafmtCheckAll + scalafixAll --check are run here as well # because an integration-only PR fires only the @@ -657,6 +663,7 @@ jobs: run: | sbt scalafmtCheckAll \ "scalafixAll --check" \ + "WorkflowOperator/test" \ "WorkflowExecutionService/test" - name: Build amber dist for the texera-web boot test # Boot smoke test for texera-web (mirrors platform-integration for the diff --git a/amber/build.sbt b/amber/build.sbt index 06d1949692..b1f8a0d481 100644 --- a/amber/build.sbt +++ b/amber/build.sbt @@ -69,19 +69,13 @@ Test / unmanagedSourceDirectories += baseDirectory.value / "src" / "test" / "int // scalafix still cover it and `sbt Test/runMain` can invoke benches. Test / unmanagedSourceDirectories += baseDirectory.value / "src" / "bench" / "scala" -// Test-filter switch driven by the AMBER_TEST_FILTER env var so the -// amber and amber-integration CI jobs select disjoint subsets without -// each invocation having to embed a `set Tests.Argument(...)` prefix. -// skip-integration : exclude @IntegrationTest-tagged specs (amber job) -// integration-only : include only @IntegrationTest-tagged specs (amber-integration job) -// (unset) : run everything (default for local sbt) -Test / testOptions ++= (sys.env.get("AMBER_TEST_FILTER") match { - case Some("skip-integration") => - Seq(Tests.Argument(TestFrameworks.ScalaTest, "-l", "org.apache.texera.amber.tags.IntegrationTest")) - case Some("integration-only") => - Seq(Tests.Argument(TestFrameworks.ScalaTest, "-n", "org.apache.texera.amber.tags.IntegrationTest")) - case _ => Nil -}) +// Lets the amber and amber-integration CI jobs select disjoint subsets without +// each invocation having to embed a `set Tests.Argument(...)` prefix. See +// project/TestFilters.scala. +Test / testOptions ++= TestFilters.integrationSplit( + envVar = "AMBER_TEST_FILTER", + tag = "org.apache.texera.amber.tags.IntegrationTest" +) // Excluding some proto files: PB.generate / excludeFilter := "scalapb.proto" diff --git a/common/workflow-operator/build.sbt b/common/workflow-operator/build.sbt index d1cd1d4cc3..2bb41754a5 100644 --- a/common/workflow-operator/build.sbt +++ b/common/workflow-operator/build.sbt @@ -35,6 +35,19 @@ ThisBuild / conflictManager := ConflictManager.latestRevision // Restrict parallel execution of tests to avoid conflicts Global / concurrentRestrictions += Tags.limit(Tags.Test, 1) +// A test needing more than a bare Python interpreter is tagged, so the amber job +// excludes it, and amber-integration, which installs amber's requirements files, +// runs it. The amber job already sets this env var on the step that invokes +// WorkflowOperator/jacoco, so no workflow change is needed for the exclusion. +// +// PythonCodeRawInvalidTextSpec reads the same env var and value directly, to tell +// a missing package in amber-integration (a defect) from one on a developer's +// machine (a local-setup fact). Changing the variable or the value here without +// changing it there leaves that test cancelling in the job meant to fail it. +Test / testOptions ++= TestFilters.integrationSplit( + envVar = "AMBER_TEST_FILTER", + tag = "org.apache.texera.amber.operator.tags.IntegrationTest" +) ///////////////////////////////////////////////////////////////////////////// // Compiler Options diff --git a/common/workflow-operator/src/test/java/org/apache/texera/amber/operator/tags/IntegrationTest.java b/common/workflow-operator/src/test/java/org/apache/texera/amber/operator/tags/IntegrationTest.java new file mode 100644 index 0000000000..1738a0e8ab --- /dev/null +++ b/common/workflow-operator/src/test/java/org/apache/texera/amber/operator/tags/IntegrationTest.java @@ -0,0 +1,51 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.texera.amber.operator.tags; + +import java.lang.annotation.ElementType; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; + +import org.scalatest.TagAnnotation; + +/** + * Marks a test in this module as needing more than a bare Python interpreter — + * pandas or plotly, which the {@code amber} job does not install. See the + * AMBER_TEST_FILTER block in {@code common/workflow-operator/build.sbt} for how + * it routes to {@code amber-integration}. + * + * <p>Apply it to a whole spec as an annotation, or — so that a spec's cheaper + * assertions stay in the unit job and its coverage report — to a single case: + * {@code test(name, Tag(classOf[IntegrationTest].getName))} in a FunSuite, + * {@code taggedAs} in a FlatSpec. + * + * <p>amber's own tag lives in {@code amber/src/test/integration} and is not + * reachable here, since amber depends on this module rather than the reverse. + * + * <p>Java, not Scala: ScalaTest finds tag annotations by + * {@code java.lang.annotation} reflection, and a Scala {@code StaticAnnotation} + * produces no JVM annotation interface for {@code @TagAnnotation} to mark. + */ +@TagAnnotation +@Retention(RetentionPolicy.RUNTIME) +@Target({ElementType.METHOD, ElementType.TYPE}) +public @interface IntegrationTest { +} diff --git a/common/workflow-operator/src/test/resources/python/hanging_worker.py b/common/workflow-operator/src/test/resources/python/hanging_worker.py new file mode 100644 index 0000000000..95bcdb9aeb --- /dev/null +++ b/common/workflow-operator/src/test/resources/python/hanging_worker.py @@ -0,0 +1,87 @@ +#!/usr/bin/env python3 +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +""" +Test fixture: a worker that stays alive and stops talking, the failure a pool +timeout exists for. A crash is not a substitute — that closes the pipe and the +read returns on its own. + + --hang-before-ready never announce readiness + --deaf announce readiness, then never read stdin at all, which + blocks the parent's write once the request outgrows the + pipe buffer + --babble announce a line that is not the protocol at all + (default) announce readiness, then answer every request except one + asking to hang: {"hang": true} is read and never answered. + {"drop-exit": true} is answered without an exit code + +Whether a request is answered is a property of the request, not of a flag, +because the pool keys a sub-pool by script, args and env: a test that needs a +hung worker and a healthy job in the *same* sub-pool cannot get them from two +different launches. +""" +from __future__ import annotations + +import json +import sys +import time + + +def _sleep_forever() -> None: + # Outlives any timeout a test sets, and the pool kills this process, so the + # sleep is what makes the worker unresponsive rather than slow. + while True: + time.sleep(3600) + + +def main() -> None: + if "--hang-before-ready" in sys.argv: + _sleep_forever() + + if "--babble" in sys.argv: + # Then stay alive: it is the parent that has to end this process, since + # nothing else would reap a worker rejected before it joined the pool. + sys.stdout.write("not a protocol line\n") + sys.stdout.flush() + _sleep_forever() + + sys.stdout.write(json.dumps({"ready": True}) + "\n") + sys.stdout.flush() + + if "--deaf" in sys.argv: + _sleep_forever() + + for line in sys.stdin: + try: + request = json.loads(line) + except ValueError: + request = {} + if request.get("hang"): + _sleep_forever() + if request.get("drop-exit"): + # Parses as the protocol, but leaves out the one field the parent + # cannot supply for itself. + sys.stdout.write(json.dumps({"stdout": "", "stderr": ""}) + "\n") + sys.stdout.flush() + continue + sys.stdout.write(json.dumps({"exit": 0, "stdout": "", "stderr": ""}) + "\n") + sys.stdout.flush() + + +if __name__ == "__main__": + main() diff --git a/common/workflow-operator/src/test/resources/python/py_compile_worker.py b/common/workflow-operator/src/test/resources/python/py_compile_worker.py new file mode 100644 index 0000000000..f7f143b17d --- /dev/null +++ b/common/workflow-operator/src/test/resources/python/py_compile_worker.py @@ -0,0 +1,77 @@ +#!/usr/bin/env python3 +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +""" +Persistent worker that syntax-checks generated operator code, replacing one +`python -I -S -B -m py_compile <file>` spawn per operator descriptor. + +`compile(source, path, "exec", dont_inherit=True)` is what `py_compile` does +before writing a `.pyc` and raises the same SyntaxError; skipping that write is +why neither `-B` nor a temp file is needed here. `dont_inherit` is the half a +pooled worker cannot leave out: without it this module's own `__future__` +import applies to every source it checks, which on CPython 3.12 rejects a +walrus inside an annotation that the spawn accepts. + +Protocol (line-delimited JSON, both directions): + + startup worker -> parent: {"ready": true} + request parent -> worker: {"source": "<code>", "name": "<label>"}\n + response worker -> parent: {"exit": 0, "stdout": "...", "stderr": "..."}\n + +`exit` is 1 when the source does not compile, with the SyntaxError in `stderr`, +mirroring the spawn it replaces so the parent's reporting is unchanged. That +does not end the worker; only a hard interpreter crash does. +""" +from __future__ import annotations + +import json +import sys +import traceback + + +def _compile_one(source: str, name: str) -> "dict[str, object]": + """Compile one generated module. `name` is the filename the traceback shows, + so a report names the descriptor rather than a temp path. + """ + try: + compile(source, name, "exec", dont_inherit=True) + return {"exit": 0, "stdout": "", "stderr": ""} + except (SyntaxError, ValueError): + # ValueError: sources compile() rejects outright, e.g. an embedded NUL. + return {"exit": 1, "stdout": "", "stderr": traceback.format_exc()} + + +def main() -> None: + sys.stdout.write(json.dumps({"ready": True}) + "\n") + sys.stdout.flush() + + for line in sys.stdin: + line = line.strip() + if not line: + continue + try: + req = json.loads(line) + result = _compile_one(req["source"], req.get("name", "<generated>")) + except Exception: # malformed request — report, keep serving + result = {"exit": 1, "stdout": "", "stderr": traceback.format_exc()} + sys.stdout.write(json.dumps(result) + "\n") + sys.stdout.flush() + + +if __name__ == "__main__": + main() diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/util/PythonCodeRawInvalidTextSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/util/PythonCodeRawInvalidTextSpec.scala index 122b1dbae8..41b58c76f5 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/util/PythonCodeRawInvalidTextSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/util/PythonCodeRawInvalidTextSpec.scala @@ -21,15 +21,23 @@ package org.apache.texera.amber.util import com.typesafe.config.ConfigFactory import org.apache.texera.amber.operator.PythonOperatorDescriptor +import org.apache.texera.amber.operator.tags.IntegrationTest import org.apache.texera.amber.pybuilder.PythonReflectionTextUtils.truncateBlock import org.apache.texera.amber.pybuilder.PythonReflectionUtils +import org.apache.texera.amber.util.JSONUtils.objectMapper +import org.apache.texera.amber.util.python.PythonWorkerPool +import org.scalatest.Tag import org.scalatest.funsuite.AnyFunSuite import java.nio.charset.StandardCharsets import java.nio.file.Files import java.util.concurrent -import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicInteger +import java.util.concurrent.{Executors, TimeUnit} +import scala.concurrent.duration.{DurationInt, FiniteDuration} +import scala.concurrent.{Await, ExecutionContext, Future} import scala.util.Try +import scala.util.control.NonFatal /** * Regression tests for validation pipeline used for PythonOperatorDescriptor codegen. @@ -49,8 +57,99 @@ final class PythonCodeRawInvalidTextSpec extends AnyFunSuite { private val MaxDepth: Int = 3 private val AcceptPackages: Seq[String] = Seq("org.apache.texera.amber.operator") + /** Budget for one whole fanned-out pass over every descriptor. Deliberately far + * above a real run (well under a second), so it only ever fires on a hang. + */ + private val PassTimeout: FiniteDuration = 10.minutes + + /** Runs the given work concurrently and returns the results in submission + * order, rethrowing the first failure so it fails the test. Sized to the pool + * so the threads match the workers available to serve them. + * + * Daemon threads: a task parked on a subprocess pipe answers no interrupt, so + * `shutdownNow` need not end it, and a non-daemon one left there would hold the + * JVM — and the build — open after [[PassTimeout]] has already failed the test. + */ + private def awaitAll[T](work: Seq[() => T]): Seq[T] = { + val threads = Executors.newFixedThreadPool( + PythonWorkerPool.maxWorkers, + (r: Runnable) => { + val t = new Thread(r, "py-compile-check") + t.setDaemon(true) + t + } + ) + try { + implicit val ec: ExecutionContext = ExecutionContext.fromExecutorService(threads) + Await.result(Future.sequence(work.map(w => Future(w()))), PassTimeout) + } finally threads.shutdownNow() + } + + /** Count of checks the pooled path could not serve. Not a failure — the spawn each one + * fell back to is the pre-pool behavior — but a number the summary has to + * carry, since a check passing says nothing about which path answered it. + */ + private val spawnFallbacks = new AtomicInteger(0) + + /** Syntax-checks one generated module, through a pooled worker when available. + * + * The worker is launched with the same `-I -S` isolation the one-shot path + * uses, so what the check accepts is unchanged; it just stops paying an + * interpreter boot — the whole cost of a check whose real work is under a + * millisecond — once per descriptor. A worker the pool cannot give out, or + * loses mid-job, falls back to the spawn, so behavior is never worse than + * before the pool. + */ + private def syntaxCheck( + pythonExecutable: String, + pythonSource: String, + descriptorName: String + ): Either[String, Unit] = { + def viaPool: Either[String, Unit] = { + val request = objectMapper.createObjectNode() + request.put("source", pythonSource) + request.put("name", s"$descriptorName.py") + val outcome = PythonWorkerPool.run( + resourcePath = "/python/py_compile_worker.py", + launchArgs = Seq.empty, + pythonExe = pythonExecutable, + request = request, + interpreterArgs = Seq("-I", "-S") + ) + if (outcome.exit == 0) Right(()) + else { + val output = if (outcome.stderr.trim.nonEmpty) outcome.stderr.trim else "(no output)" + Left( + s"py_compile failed (exit=${outcome.exit})\nOutput:\n" + + truncateBlock(output, maxLines = 40, maxChars = 8000) + ) + } + } + + if (PythonWorkerPool.enabled) { + try viaPool + catch { + // Anything the pooled path throws leaves the spawn as the answer, which is + // what makes it never worse than before: not only a worker that died + // mid-job, but equally one the pool could not hand out at all. Those + // arrive as WorkerDiedException; NonFatal also covers the steps outside + // that contract, such as materializing the worker script. Counted, so a + // run the pool served none of does not read as a green pooled run. + case NonFatal(thrown) => + println( + s"[py-compile FALLBACK ${spawnFallbacks.incrementAndGet()}] $descriptorName: " + + s"pooled worker unavailable, spawning instead: " + + truncateBlock(thrown.toString, maxLines = 3, maxChars = 500) + ) + pyCompile(pythonExecutable, pythonSource) + } + } else pyCompile(pythonExecutable, pythonSource) + } + /** * Runs `python -m py_compile` on the provided source, using an isolated interpreter invocation. + * Retained as the pooled path's fallback and as the behavior selected by + * TEXERA_TEST_PYTHON_WORKER=0. * * Isolation flags: * - -I : isolate (ignore user site-packages / env) @@ -225,21 +324,23 @@ final class PythonCodeRawInvalidTextSpec extends AnyFunSuite { } val total = descriptorCandidates.size - var ok = 0 - var checked = 0 - - val allFindings = descriptorCandidates.flatMap { descriptorClass => - checked += 1 + val ok = new AtomicInteger(0) + val checked = new AtomicInteger(0) + // Checked concurrently: the fan-out is what turns the pool's workers into + // parallel interpreters rather than a queue in front of one. The executor is + // sized to maxWorkers, so nothing runs past the cap. + val allFindings = awaitAll(descriptorCandidates.map { descriptorClass => () => val checkResult = PythonReflectionUtils.checkDescriptorWithCode( descriptorClass, rawInvalidText = RawInvalid, maxDepth = MaxDepth ) + checked.incrementAndGet() val pyCompileFindings = checkResult.code.toSeq.flatMap { generatedCode => - pyCompile(pythonExecutable, generatedCode) match { + syntaxCheck(pythonExecutable, generatedCode, descriptorClass.getSimpleName) match { case Left(errorMessage) => Seq(PythonReflectionUtils.Finding(descriptorClass.getName, "py-compile", errorMessage)) case Right(()) => Nil @@ -249,18 +350,56 @@ final class PythonCodeRawInvalidTextSpec extends AnyFunSuite { val findings = checkResult.findings ++ pyCompileFindings if (findings.isEmpty && checkResult.code.nonEmpty) { - ok += 1 - println(s"[py-compile OK $ok/$total | checked $checked/$total] ${descriptorClass.getName}") + println( + s"[py-compile OK ${ok.incrementAndGet()}/$total | " + + s"checked ${checked.get()}/$total] ${descriptorClass.getName}" + ) } findings - } + }).flatten - println(s"[py-compile SUMMARY] ok=$ok/$total") + println( + s"[py-compile SUMMARY] ok=${ok.get()}/$total, spawn fallbacks=${spawnFallbacks.get()}" + ) if (allFindings.nonEmpty) { fail(PythonReflectionUtils.renderReport(allFindings, total = total)) } } + /** py_compile above only parses the emitted code; running it needs the packages + * it imports. Tagged, so only amber-integration — the job that installs them — + * runs this. There a missing package is a defect; elsewhere it is a local-setup + * fact, so cancel rather than fail. + */ + test( + "the Python interpreter operator templates run in should import pandas and plotly", + Tag(classOf[IntegrationTest].getName) + ) { + // Same env var and value the build reads to select this subset, in + // common/workflow-operator/build.sbt; keep the two in step. Nothing enforces + // that from here, since TestFilters is build-scope and cannot be imported: if + // the selector is renamed and this string is not, the test keeps running in + // amber-integration but cancels instead of failing, which is the non-result + // the tag exists to remove. + val provisioned = sys.env.get("AMBER_TEST_FILTER").contains("integration-only") + def unavailable(message: String): Nothing = + if (provisioned) fail(message) else cancel(message) + + val python = loadPythonExeFromUdfConf().getOrElse(unavailable("no runnable python")) + val imported = Try { + val process = new ProcessBuilder(python, "-c", "import pandas, plotly") + .redirectErrorStream(true) + .start() + // Killed on the way out: a probe that ran out of time is still running, and + // would otherwise leak into the rest of the run. + if (process.waitFor(60, TimeUnit.SECONDS)) process.exitValue() == 0 + else { process.destroyForcibly(); false } + } + if (!imported.getOrElse(false)) { + unavailable(s"'$python' cannot import pandas and plotly") + } + } + } diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/util/python/PythonWorkerPool.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/util/python/PythonWorkerPool.scala new file mode 100644 index 0000000000..414ad40ebb --- /dev/null +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/util/python/PythonWorkerPool.scala @@ -0,0 +1,503 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.texera.amber.util.python + +import com.fasterxml.jackson.databind.node.ObjectNode +import com.typesafe.scalalogging.LazyLogging +import org.apache.texera.amber.util.JSONUtils.objectMapper + +import java.io.{BufferedReader, BufferedWriter, InputStreamReader, OutputStreamWriter} +import java.nio.charset.StandardCharsets +import java.nio.file.{Files, Path, StandardCopyOption} +import java.util.concurrent.{ + Callable, + ConcurrentHashMap, + ExecutionException, + ExecutorService, + Executors, + LinkedBlockingQueue, + TimeUnit, + TimeoutException +} +import java.util.concurrent.atomic.AtomicInteger +import scala.annotation.tailrec +import scala.collection.mutable +import scala.jdk.CollectionConverters._ +import scala.util.control.NonFatal + +/** + * Pools of persistent Python "worker" processes that eliminate the per-call + * interpreter-boot + import cost a test otherwise pays on every subprocess + * spawn. Testing operators one at a time does not scale when each one costs a + * spawn: a bare `-I -S` interpreter boots in ~25 ms, and once pandas and plotly + * are imported a spawn costs ~260-310 ms — mostly the imports — dwarfing + * the ~4 ms of real work a job does. A worker pays that once at startup, then + * serves many jobs over its lifetime, so N spawns become one. + * + * Lives in test scope here, rather than beside a single caller, because tests in + * several modules run generated operator code and would otherwise each + * hand-roll a driver, a stdout protocol and a timeout. Other modules reach it + * through a `test->test` dependency on this one. + * + * Generic over the worker script — the pool never interprets the payload — so + * one implementation serves a syntax check, template execution and DataFrame + * comparison alike. Each distinct (resource, interpreterArgs, launchArgs, + * python, env) combination gets its own sub-pool. + * + * Protocol (line-delimited JSON, shared by all worker scripts): + * startup worker -> pool: {"ready": true} + * request pool -> worker: <caller-supplied JSON object>\n + * response worker -> pool: {"exit": <int>, "stdout": "...", "stderr": "..."}\n + * + * Concurrency: callers submit from several threads at once — one test fanning + * its cases out, or suites running in parallel — so each sub-pool holds up to + * [[maxWorkers]] workers, each serving one job at a time (borrow -> run -> + * return). A worker script may chdir per job, so a worker must never run two + * jobs at once — the borrow/return discipline guarantees that. + * + * Robustness: an ordinary *job* failure comes back as an [[Outcome]] with + * `exit != 0` (worker keeps running). A hard interpreter crash ends a worker; + * the pool detects the EOF / broken pipe, discards it, and throws + * [[WorkerDiedException]] so the caller can fall back to a one-shot subprocess + * — behavior is then never worse than the pre-pool path. A worker that stays + * alive but stops answering ends the same way, on the [[Timeouts]] below. + */ +object PythonWorkerPool extends LazyLogging { + + /** Worker response: process-like exit code plus captured streams. */ + final case class Outcome(exit: Int, stdout: String, stderr: String) + + /** Thrown for a worker the pool could not give out or keep: one that would not + * start, one that never signalled ready, or one that died or fell silent + * mid-job. Callers catch this and fall back to a one-shot subprocess. + */ + final class WorkerDiedException(message: String, cause: Throwable = null) + extends RuntimeException(message, cause) + + /** Feature toggle. `TEXERA_TEST_PYTHON_WORKER=0` (or `false`/`off`) forces the + * one-subprocess-per-call paths everywhere — an escape hatch for debugging a + * suspected isolation leak. Default on. + */ + val enabled: Boolean = + !sys.env + .get("TEXERA_TEST_PYTHON_WORKER") + .map(_.trim.toLowerCase) + .exists(Set("0", "false", "off")) + + /** Max live workers per sub-pool, so callers on distinct worker + * scripts add up rather than share this bound. Defaults to 4, override via + * `TEXERA_TEST_PYTHON_WORKERS`. Public so a caller fanning out jobs within one + * test can size that fan-out to the workers it will get. + */ + val maxWorkers: Int = + sys.env + .get("TEXERA_TEST_PYTHON_WORKERS") + .flatMap(s => scala.util.Try(s.trim.toInt).toOption) + .filter(_ > 0) + .getOrElse(4) + + /** How long a caller waits on a worker before the pool kills and discards it. + * A read on a process pipe cannot be interrupted — a suite or executor timeout + * leaves the reading thread stuck on it — so a worker that stays alive without + * answering has to be bounded here. `responseMillis` keeps the 30 seconds the + * one-shot spawn this pool replaced allowed a job; `startupMillis` is longer because + * a worker imports its libraries before it reports ready, and a loaded CI + * machine makes that slow. Override in seconds via + * `TEXERA_TEST_PYTHON_WORKER_TIMEOUT` / `TEXERA_TEST_PYTHON_WORKER_STARTUP_TIMEOUT`. + */ + final case class Timeouts(responseMillis: Long, startupMillis: Long) + + object Timeouts { + private def envSeconds(name: String, default: Long): Long = + sys.env + .get(name) + .flatMap(s => scala.util.Try(s.trim.toLong).toOption) + .filter(_ > 0) + .getOrElse(default) * 1000 + + val Default: Timeouts = Timeouts( + responseMillis = envSeconds("TEXERA_TEST_PYTHON_WORKER_TIMEOUT", 30), + startupMillis = envSeconds("TEXERA_TEST_PYTHON_WORKER_STARTUP_TIMEOUT", 60) + ) + } + + /** + * Run one job through a pooled worker for `resourcePath`, launched as + * `pythonExe <interpreterArgs> <script> <launchArgs>` with extra environment + * `env`. `request` is the worker-specific JSON payload (the pool does not + * interpret it). Throws [[WorkerDiedException]] for every worker the pool could + * not give out or keep — one that would not start, would not report ready, or + * died mid-job — so one `catch` covers a caller's whole fallback. + * + * `interpreterArgs` are the flags that must precede the script — a syntax + * checker wants `-I -S` so it validates under the same isolation a one-shot + * `python -I -S -m py_compile` gave it. `launchArgs` are the script's own + * (e.g. `--deaf`), and `env` carries what a flag cannot — though not + * PYTHONPATH or any other PYTHON* var, which the `-I` above makes CPython ignore. + * All three are part of a worker's identity: one started differently is not + * interchangeable, so it gets its own sub-pool. `timeouts` is not — it bounds + * this call, so a caller whose jobs are slower than most can raise it without + * splitting the pool. + */ + def run( + resourcePath: String, + launchArgs: Seq[String], + pythonExe: String, + request: ObjectNode, + env: Map[String, String] = Map.empty, + interpreterArgs: Seq[String] = Seq.empty, + timeouts: Timeouts = Timeouts.Default + ): Outcome = { + val pool = pools.computeIfAbsent( + Key(resourcePath, pythonExe, interpreterArgs.toList, launchArgs.toList, env.toList.sorted), + _ => new Pool(resourcePath, launchArgs, pythonExe, env, interpreterArgs) + ) + pool.run(request, timeouts) + } + + /** What makes two launches the same worker. Compared field by field rather than + * as one joined string, so a value carrying whatever the separator was — a + * python path with a space in it, a PYTHONPATH — cannot make two different + * launches share a pool. + */ + private final case class Key( + resourcePath: String, + pythonExe: String, + interpreterArgs: List[String], + launchArgs: List[String], + env: List[(String, String)] + ) + + /** How long a caller at the worker cap waits before re-examining it. Not a + * deadline — see [[Pool.borrow]]. + */ + private val CapRecheckMillis: Long = 250 + + private val pools = new ConcurrentHashMap[Key, Pool]() + + Runtime.getRuntime.addShutdownHook(new Thread(() => shutdownAll())) + + private def shutdownAll(): Unit = + pools.values().forEach(_.shutdown()) + + // A single sub-pool: up to `maxWorkers` live workers for one worker script. + private final class Pool( + resourcePath: String, + launchArgs: Seq[String], + pythonExe: String, + env: Map[String, String], + interpreterArgs: Seq[String] + ) { + private val idle = new LinkedBlockingQueue[Worker]() + private val liveCount = new AtomicInteger(0) + private val all = mutable.Set.empty[Worker] // guarded by `all` + @volatile private var script: Path = _ + + def run(request: ObjectNode, timeouts: Timeouts): Outcome = { + val w = borrow(timeouts) + try { + val outcome = w.run(request, timeouts.responseMillis) + idle.offer(w) // healthy — return to pool + outcome + } catch { + // Every throw, not only a WorkerDiedException: an interrupt on this + // thread — what an executor's `shutdownNow` sends — leaves the blocking + // read or write as an InterruptedException, which `NonFatal` excludes and + // [[Worker.run]] therefore does not wrap. A worker left neither returned + // nor discarded costs this sub-pool a slot for the life of the JVM, and + // its answer may still be in flight, so it cannot be reused either way. + case e: Throwable => + discard(w) + throw e + } + } + + @tailrec + private def borrow(timeouts: Timeouts): Worker = { + val existing = idle.poll() + if (existing != null) existing + else if (liveCount.getAndIncrement() < maxWorkers) { + try create(timeouts) + catch { + case e: Throwable => + liveCount.decrementAndGet() + throw e + } + } else { + liveCount.decrementAndGet() + // At the cap. Waiting outright for a returned worker would strand this + // caller when the ones ahead are discarded instead: a discard frees a + // slot without putting anything back. So wait only briefly, then look at + // the cap again — the next pass starts a replacement. A long queue still + // waits as long as it takes; that is the caller's own backlog, not a hang. + val returned = idle.poll(CapRecheckMillis, TimeUnit.MILLISECONDS) + if (returned != null) returned else borrow(timeouts) + } + } + + private def create(timeouts: Timeouts): Worker = { + val cmd = + (((pythonExe +: interpreterArgs) :+ ensureScript().toString) ++ launchArgs).asJava + val pb = new ProcessBuilder(cmd).redirectErrorStream(false) + env.foreach { case (k, v) => pb.environment().put(k, v) } + // A worker that cannot be started is a worker death like any other, so it + // leaves through the same exception: `start` throws a bare IOException — no + // interpreter at that path, or the OS out of processes under a fan-out — and + // a caller's fallback is written against [[WorkerDiedException]]. + val process = + try pb.start() + catch { + case NonFatal(e) => + throw new WorkerDiedException( + s"could not start python worker for $resourcePath: ${e.getMessage}", + e + ) + } + val w = new Worker(process, s"$resourcePath ${launchArgs.mkString(" ")}".trim) + // Anything thrown before the worker joins `all` — a startup that timed out, + // an interrupt on this thread — leaves an interpreter nothing else would + // reap: not the caller, which never gets the handle, and not the shutdown + // hook, which walks `all`. `destroy` is idempotent, so a worker that already + // killed itself on the way out is no exception to that. + try w.awaitReady(timeouts.startupMillis) + catch { + case e: Throwable => + w.destroy() + throw e + } + all.synchronized(all.add(w)) + logger.debug(s"Started python worker for $resourcePath (live=${liveCount.get}/$maxWorkers)") + w + } + + private def discard(w: Worker): Unit = { + all.synchronized(all.remove(w)) + liveCount.decrementAndGet() + w.destroy() + } + + private def ensureScript(): Path = { + if (script == null) synchronized { + if (script == null) { + val stream = getClass.getResourceAsStream(resourcePath) + require(stream != null, s"worker script not found on classpath at $resourcePath") + try { + val tmp = Files.createTempFile("py-worker-", ".py") + Files.copy(stream, tmp, StandardCopyOption.REPLACE_EXISTING) + tmp.toFile.deleteOnExit() + script = tmp + } finally stream.close() + } + } + script + } + + def shutdown(): Unit = + all.synchronized { + all.foreach(_.destroy()) + all.clear() + } + } + + // One live worker process plus its framed-JSON stdin and background drains of + // its stdout (the protocol) and stderr (only non-empty on a hard crash). + private final class Worker(process: Process, label: String) { + private val stdin: BufferedWriter = + new BufferedWriter(new OutputStreamWriter(process.getOutputStream, StandardCharsets.UTF_8)) + private val errBuf = new StringBuilder + + // Protocol lines the worker has written, `None` marking end of stream. A + // dedicated thread owns the blocking read so a caller can wait with a + // timeout: `readLine` on a process pipe answers neither an interrupt nor a + // deadline, and only closing the pipe — killing the process — releases it. + private val lines = new LinkedBlockingQueue[Option[String]]() + + // Owns the writing end for the same reason. + private val writer: ExecutorService = Executors.newSingleThreadExecutor { r => + val t = new Thread(r, "python-worker-stdin") + t.setDaemon(true) + t + } + + private val outThread: Thread = { + val t = new Thread(() => { + val r = + new BufferedReader(new InputStreamReader(process.getInputStream, StandardCharsets.UTF_8)) + try { + var line = r.readLine() + while (line != null) { + lines.put(Some(line)) + line = r.readLine() + } + } catch { case NonFatal(_) => () } + finally lines.put(None) + }) + t.setDaemon(true) + t.setName("python-worker-stdout") + t.start() + t + } + + private val errThread: Thread = { + val t = new Thread(() => { + val r = + new BufferedReader(new InputStreamReader(process.getErrorStream, StandardCharsets.UTF_8)) + try { + var line = r.readLine() + while (line != null) { + errBuf.synchronized(errBuf.append(line).append('\n')) + line = r.readLine() + } + } catch { case NonFatal(_) => () } + }) + t.setDaemon(true) + t.setName("python-worker-stderr") + t.start() + t + } + + /** Wait for the worker's startup `{"ready": true}`; if it dies first (e.g. an + * import failed) or never gets there, surface its stderr. A worker that + * fails here is killed: it is not in the pool's set yet, so nothing else + * will reap it. + * + * A first line that is not the protocol at all counts as not ready, rather + * than throwing whatever the parser throws: the caller's contract here is + * [[WorkerDiedException]], and an escape past it would leave this + * interpreter with nothing to reap it. The line goes into the message — + * stderr is empty when a script writes its noise to stdout. + */ + def awaitReady(timeoutMillis: Long): Unit = { + val line = nextLine(timeoutMillis, "signal ready") + val ready = + try objectMapper.readTree(line).path("ready").asBoolean(false) + catch { case NonFatal(_) => false } + if (!ready) { + destroy() + throw new WorkerDiedException( + s"python worker [$label] did not signal ready; it wrote: ${abbreviate(line)}." + + s" stderr:\n${drainErr()}" + ) + } + } + + /** An answer without an exit code is the worker breaking protocol, not a job + * that failed, so it leaves as a worker death — the same refusal to guess + * [[awaitReady]] makes about the startup line. Defaulting it would report the + * caller's own input as the thing that failed, with an empty stdout as the + * evidence, where a death routes to the caller's fallback instead. An absent + * stream is a different matter: empty is what it means. + */ + def run(request: ObjectNode, timeoutMillis: Long): Outcome = + try { + send(request, timeoutMillis) + val line = nextLine(timeoutMillis, "answer") + val node = objectMapper.readTree(line) + val exit = node.path("exit") + if (!exit.isNumber) { + throw new WorkerDiedException( + s"python worker [$label] answered without an exit code;" + + s" it wrote: ${abbreviate(line)}" + ) + } + Outcome(exit.asInt(), node.path("stdout").asText(""), node.path("stderr").asText("")) + } catch { + case e: WorkerDiedException => throw e + case NonFatal(e) => + throw new WorkerDiedException( + s"I/O error talking to python worker [$label]: ${e.getMessage}", + e + ) + } + + /** Hand the request over, on the same deadline as the answer. A worker that + * has stopped reading its stdin blocks the write as soon as the payload + * outgrows the pipe buffer, and that write is no more interruptible than the + * read, so it too runs on a thread of its own — a daemon, so a lost one + * cannot keep the JVM alive. Killing the worker closes the pipe, which is + * what releases that thread. + */ + private def send(request: ObjectNode, timeoutMillis: Long): Unit = { + val write = writer.submit(new Callable[Unit] { + override def call(): Unit = { + stdin.write(objectMapper.writeValueAsString(request)) + stdin.write("\n") + stdin.flush() + } + }) + try write.get(timeoutMillis, TimeUnit.MILLISECONDS) + catch { + case _: TimeoutException => + destroy() + throw new WorkerDiedException( + s"python worker [$label] did not read its request within ${timeoutMillis}ms;" + + s" killed it. stderr:\n${drainErr()}" + ) + case e: ExecutionException => + throw new WorkerDiedException( + s"I/O error sending to python worker [$label]: ${e.getCause.getMessage}", + e.getCause + ) + } + } + + /** One protocol line, or a [[WorkerDiedException]]: the worker ended the + * stream, or it went `timeoutMillis` without writing. `what` names what was + * being waited for. A worker that timed out is killed here — that both frees + * the machine of a hung interpreter and unblocks [[outThread]] — and the + * pool discards it, so it never serves another job. + */ + private def nextLine(timeoutMillis: Long, what: String): String = + lines.poll(timeoutMillis, TimeUnit.MILLISECONDS) match { + case null => // poll's own signal that the deadline passed with nothing written + destroy() + throw new WorkerDiedException( + s"python worker [$label] did not $what within ${timeoutMillis}ms; killed it." + + s" stderr:\n${drainErr()}" + ) + case None => // end of stream: the interpreter is gone + throw new WorkerDiedException(s"python worker [$label] crashed. stderr:\n${drainErr()}") + case Some(line) => line + } + + private def drainErr(): String = errBuf.synchronized(errBuf.toString) + + /** Enough of a stray protocol line to recognize it, not a whole traceback. */ + private def abbreviate(line: String): String = + if (line.length <= 200) line else s"${line.take(200)}... (${line.length} chars)" + + /** Idempotent: called on a crash, a timeout, and again on pool shutdown. + * + * The process goes first and the streams after: a write blocked on a full + * pipe only comes back once the pipe has no reader, and closing the buffered + * writer would try to flush into that same pipe — so closing first is itself + * a way to hang. Nothing is flushed on the way out; a worker being destroyed + * has no use for the rest of a request. + */ + def destroy(): Unit = { + writer.shutdownNow() + process.destroy() + try if (!process.waitFor(2, TimeUnit.SECONDS)) process.destroyForcibly() + catch { case NonFatal(_) => process.destroyForcibly() } + try process.getOutputStream.close() + catch { case NonFatal(_) => () } + } + } +} diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/util/python/PythonWorkerPoolSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/util/python/PythonWorkerPoolSpec.scala new file mode 100644 index 0000000000..6187cf8e24 --- /dev/null +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/util/python/PythonWorkerPoolSpec.scala @@ -0,0 +1,266 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.texera.amber.util.python + +import com.fasterxml.jackson.databind.node.ObjectNode +import org.apache.texera.amber.util.JSONUtils.objectMapper +import org.scalatest.funsuite.AnyFunSuite + +import java.util.concurrent.{Executors, TimeUnit} +import scala.concurrent.duration._ +import scala.concurrent.{Await, ExecutionContext, ExecutionContextExecutorService, Future} +import scala.util.Try + +/** + * What the pool owes a caller when a worker misbehaves. An ordinary job failure + * is the other suites' business; this one is about a worker that stays alive and + * stops taking part, which is the case that does not end by itself: a crash + * closes the pipe and the pending read returns, while silence would hold the + * caller forever — neither a read nor a write on a process pipe answers an + * interrupt or a deadline, so no suite-level timeout can release one. + * + * Every wait here is bounded and runs on daemon threads, so a regression fails + * these tests instead of wedging the run: a lost non-daemon thread parked on a + * pipe would keep the JVM, and the build, alive. + * + * The fixture worker is stdlib-only and runs under `-I -S`, so this needs an + * interpreter but none of the operator packages. + */ +final class PythonWorkerPoolSpec extends AnyFunSuite { + + private val HangingWorker = "/python/hanging_worker.py" + + /** Short enough to keep the suite quick, far enough above process startup to + * not be mistaken for one: a fixture asked to hang never answers at all, so + * these deadlines cannot race it. + */ + private val ShortTimeouts: PythonWorkerPool.Timeouts = + PythonWorkerPool.Timeouts(responseMillis = 1500, startupMillis = 1500) + + /** For the case that interrupts its own callers: far enough out that they are + * certainly still waiting on the worker, not already past their deadline. + */ + private val PatientTimeouts: PythonWorkerPool.Timeouts = + PythonWorkerPool.Timeouts(responseMillis = 60000, startupMillis = 60000) + + /** Ceiling on a whole case, well above the deadlines under test. Reaching it + * means something never gave up. + */ + private val Bound: FiniteDuration = 25.seconds + + /** Any interpreter serves — the fixture imports only `json` and `time` — so + * this deliberately skips the configured `python.path` the suites that need + * pandas resolve. A machine without one cancels rather than fails. + */ + private lazy val python: String = { + def isRunnable(exe: String): Boolean = + Try(new ProcessBuilder(exe, "--version").redirectErrorStream(true).start()).toOption + .exists { p => + if (p.waitFor(5, TimeUnit.SECONDS)) p.exitValue() == 0 else { p.destroyForcibly(); false } + } + + List("python3", "python", "py").find(isRunnable).getOrElse(cancel("no runnable python")) + } + + /** Resolves [[python]] here so that the cancel lands on the test thread. Forced + * inside a case, it would be forced inside a `Future`, and by the time it + * reached `intercept` a cancellation is a RuntimeException like any other: the + * cases would fail on a machine without an interpreter instead of cancelling, + * and the one at the cap would record it as the failure it asserts on and pass + * without ever reaching the cap. + */ + override def withFixture(test: NoArgTest): org.scalatest.Outcome = { + val _ = python + super.withFixture(test) + } + + private def onDaemonThreads[T](threads: Int)(body: ExecutionContext => T): T = { + val pool = Executors.newFixedThreadPool( + threads, + (r: Runnable) => { + val t = new Thread(r, "pool-spec-caller") + t.setDaemon(true) + t + } + ) + val ec: ExecutionContextExecutorService = ExecutionContext.fromExecutorService(pool) + try body(ec) + finally pool.shutdownNow() + } + + private def call( + launchArgs: Seq[String], + request: ObjectNode, + timeouts: PythonWorkerPool.Timeouts + ): PythonWorkerPool.Outcome = + PythonWorkerPool.run( + resourcePath = HangingWorker, + launchArgs = launchArgs, + pythonExe = python, + request = request, + interpreterArgs = Seq("-I", "-S"), + timeouts = timeouts + ) + + /** A job the fixture takes and never answers. `hang` travels in the request + * rather than in `launchArgs` so that [[healthyCall]] lands in the same + * sub-pool: `Key` covers the script, its arguments and the environment, and a + * job routed elsewhere would be served by a pool whose queue this suite never + * touched. + */ + private def hangingCall( + launchArgs: Seq[String], + request: ObjectNode = objectMapper.createObjectNode(), + timeouts: PythonWorkerPool.Timeouts = ShortTimeouts + ): PythonWorkerPool.Outcome = + call(launchArgs, request.deepCopy().put("hang", true), timeouts) + + /** A job in that same sub-pool which the fixture does answer. */ + private def healthyCall(): PythonWorkerPool.Outcome = + call(Seq.empty, objectMapper.createObjectNode(), ShortTimeouts) + + /** The call, on a daemon thread and under [[Bound]], expected to give up. */ + private def interceptBounded(call: => Any): PythonWorkerPool.WorkerDiedException = + intercept[PythonWorkerPool.WorkerDiedException] { + onDaemonThreads(1)(ec => Await.result(Future(call)(ec), Bound)) + } + + test("a worker that takes the job and stops answering is killed and reported") { + val startedAt = System.nanoTime() + val thrown = interceptBounded(hangingCall(Seq.empty)) + val elapsedMillis = (System.nanoTime() - startedAt) / 1000000 + + assert(thrown.getMessage.contains("did not answer")) + assert(thrown.getMessage.contains("killed it")) + // Well under the default response budget: what fired is the timeout passed in, + // not a wait that happened to end. + assert(elapsedMillis < PythonWorkerPool.Timeouts.Default.responseMillis / 2) + } + + test("a worker that never signals ready is killed and reported") { + val startedAt = System.nanoTime() + val thrown = interceptBounded(hangingCall(Seq("--hang-before-ready"))) + val elapsedMillis = (System.nanoTime() - startedAt) / 1000000 + + assert(thrown.getMessage.contains("did not signal ready")) + assert(elapsedMillis < PythonWorkerPool.Timeouts.Default.startupMillis / 2) + } + + test("a worker that never reads its request is killed and reported") { + val request = objectMapper.createObjectNode() + // Past any pipe buffer, so the write cannot simply be handed to the kernel and + // left there: it is the blocked write itself that has to be given up on. + request.put("source", "x" * (4 * 1024 * 1024)) + + val thrown = interceptBounded(hangingCall(Seq("--deaf"), request)) + + assert(thrown.getMessage.contains("did not read its request")) + assert(thrown.getMessage.contains("killed it")) + } + + test("a caller waiting at the worker cap is not stranded by a discarded worker") { + // One caller more than there are workers, all onto workers that go quiet: + // the callers that hold one time out, and their workers are discarded, which + // frees a slot without handing anything back, and the caller waiting at the + // cap has to notice that rather than wait for a hand-back that never comes. + // The discards it is waiting on happen on workers other than the one ahead + // of it. + val callers = PythonWorkerPool.maxWorkers + 1 + + val outcomes = onDaemonThreads(callers) { implicit ec => + Await.result(Future.sequence(Seq.fill(callers)(Future(Try(hangingCall(Seq.empty))))), Bound) + } + + assert(outcomes.length == callers) + assert(outcomes.forall(_.isFailure)) + } + + test("the pool still serves jobs after it has discarded a timed-out worker") { + interceptBounded(hangingCall(Seq.empty)) + + // Deliberately the sub-pool the discard happened in — see [[hangingCall]] — + // so what is asserted is that a pool short one worker starts a replacement, + // not that an untouched pool works. + assert(healthyCall().exit == 0) + } + + test("an interpreter that cannot be started reaches the caller as a worker death") { + // Not the IOException ProcessBuilder raises: a caller's fallback is written + // against WorkerDiedException, and a pool that cannot hand out a worker at all + // is the case that fallback exists for. + val thrown = intercept[PythonWorkerPool.WorkerDiedException] { + PythonWorkerPool.run( + resourcePath = HangingWorker, + launchArgs = Seq.empty, + pythonExe = "no-such-python-on-this-machine", + request = objectMapper.createObjectNode(), + interpreterArgs = Seq("-I", "-S"), + timeouts = ShortTimeouts + ) + } + + assert(thrown.getMessage.contains("could not start python worker")) + } + + test("a caller interrupted mid-job does not cost the pool a worker") { + val workers = PythonWorkerPool.maxWorkers + + // Every slot taken by a job nobody will answer, and then the callers are + // interrupted — `shutdownNow` is what an executor does to a fan-out whose test + // has already failed. An interrupt is not a WorkerDiedException, so a worker + // left neither returned nor discarded would cost this sub-pool that slot for + // the rest of the JVM. + onDaemonThreads(workers) { implicit ec => + Seq.fill(workers)(Future(Try(hangingCall(Seq.empty, timeouts = PatientTimeouts)))) + // Enough for the callers to be waiting on a worker rather than starting one; + // the slot has to come back wherever the interrupt lands, so this only + // decides which of the two paths the case exercises. + Thread.sleep(500) + } + + // Serving `workers` jobs again is the whole assertion: a leaked slot leaves + // these at the cap, rechecking it until [[Bound]] runs out. + val outcomes = onDaemonThreads(workers) { implicit ec => + Await.result(Future.sequence(Seq.fill(workers)(Future(healthyCall()))), Bound) + } + + assert(outcomes.forall(_.exit == 0)) + } + + test("a worker that answers without an exit code is reported as a worker death") { + // Not as a job that failed: a default would name the caller's own request as + // what went wrong, with an empty stdout as the evidence. + val request = objectMapper.createObjectNode().put("drop-exit", true) + + val thrown = interceptBounded(call(Seq.empty, request, ShortTimeouts)) + + assert(thrown.getMessage.contains("without an exit code")) + assert(thrown.getMessage.contains("stdout")) + } + + test("a worker whose first line is not the protocol is killed and reported") { + val thrown = interceptBounded(hangingCall(Seq("--babble"))) + + assert(thrown.getMessage.contains("did not signal ready")) + // Named in the message: stderr is empty when a script writes its noise to + // stdout, so the line itself is the only evidence of what went wrong. + assert(thrown.getMessage.contains("not a protocol line")) + } +} diff --git a/project/TestFilters.scala b/project/TestFilters.scala new file mode 100644 index 0000000000..c948f1d45e --- /dev/null +++ b/project/TestFilters.scala @@ -0,0 +1,48 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +import sbt._ + +/** + * Selects a module's tagged tests for the fast-unit job or the integration job: + * skip-integration excludes them, integration-only runs only them, unset runs + * everything, and any other value fails the build rather than quietly running + * everything in a job that selected a subset. Shared because the mapping is + * identical in every module, while the env var and the tag are not — the tag + * annotation has to live somewhere the module's own Test config can see. + */ +object TestFilters { + + /** @param envVar the variable the two CI jobs set to opposite values; it has to + * be the one the workflow already sets on the step that invokes + * this module's tests, or neither subset is selected. + * @param tag fully-qualified name of the tag annotation, as + * `classOf[...].getName` gives it at the test site — ScalaTest + * matches these by string, so a rename that misses one side + * silently stops filtering. + */ + def integrationSplit(envVar: String, tag: String): Seq[TestOption] = + sys.env.get(envVar) match { + case Some("skip-integration") => + Seq(Tests.Argument(TestFrameworks.ScalaTest, "-l", tag)) + case Some("integration-only") => + Seq(Tests.Argument(TestFrameworks.ScalaTest, "-n", tag)) + case Some(other) => sys.error(s"$envVar=$other: use skip-integration or integration-only") + case None => Nil + } +}
