Copilot commented on code in PR #4359:
URL: https://github.com/apache/texera/pull/4359#discussion_r3113678783
##########
frontend/src/app/workspace/component/power-button/computing-unit-selection.component.ts:
##########
@@ -64,6 +85,13 @@ import {
styleUrls: ["./computing-unit-selection.component.scss"],
})
export class ComputingUnitSelectionComponent implements OnInit {
+ // variables for creating a virtual environment
+ pves: PveDraft[] = [this.makeEmptyPve(true)];
+ systemPackages: { name: string; version: string }[] = [];
+ pipModalCloseHandler: (() => void) | null = null;
+ pveModalVisible = false;
+ nextPveId = 1;
Review Comment:
`pves` is initialized via `makeEmptyPve()` before `nextPveId` is initialized
(field initializers run top-to-bottom). This makes `this.nextPveId++` evaluate
to `NaN`, breaking `trackByPveId` and any later ID logic. Move `nextPveId`
above `pves` (or initialize `pves` in `ngOnInit`) so IDs are always numeric and
unique.
```suggestion
nextPveId = 1;
pves: PveDraft[] = [this.makeEmptyPve(true)];
systemPackages: { name: string; version: string }[] = [];
pipModalCloseHandler: (() => void) | null = null;
pveModalVisible = false;
```
##########
frontend/src/app/workspace/component/power-button/computing-unit-selection.component.ts:
##########
@@ -637,4 +670,281 @@ export class ComputingUnitSelectionComponent implements
OnInit {
this.computingUnitStatusService.refreshComputingUnitList();
}
}
+
+ private makeEmptyPve(expanded: boolean): PveDraft {
+ return {
+ id: this.nextPveId++,
+ name: "",
+ userPackages: [],
+ newPackages: [{ name: "", operator: "==", version: "" }],
+ deletingPackages: [],
+ pipOutput: "",
+ prettyPipOutput: "",
+ expanded,
+ isInstalling: false,
+ };
+ }
+
+ addEnvironment(): void {
+ this.pves.push(this.makeEmptyPve(true));
+ }
+
+ trackByPveId(_index: number, pve: PveDraft): number {
+ return pve.id;
+ }
+
+ showPVEmodalVisible(): void {
+ this.pveModalVisible = true;
+ this.getPVEs();
+ }
+
+ getPVEs(): void {
+ const cuId: number | undefined =
this.selectedComputingUnit?.computingUnit.cuid;
+
+ if (cuId == null) {
+ this.notificationService.error("No computing unit selected. Please
select a CU first.");
+ return;
+ }
+
+ this.workflowPveService.setCuid(cuId);
+
+ this.workflowPveService
+ .fetchPVEs(cuId)
+ .pipe(untilDestroyed(this))
+ .subscribe({
+ next: (resp: PvePackageResponse[]) => {
+ this.pves = resp.map((pve, index) => ({
+ id: index,
Review Comment:
`getPVEs()` sets `id: index`, but new environments use `nextPveId++`. This
can create duplicate IDs (e.g., fetched env at index 1 and the next new env
both get ID 1), causing `trackByPveId` collisions and unstable rendering. Use a
monotonically increasing ID source for all PVEs (or a stable unique key like
`pveName`).
```suggestion
this.pves = resp.map(pve => ({
id: this.nextPveId++,
```
##########
amber/src/main/scala/org/apache/texera/web/resource/pythonvirtualenvironment/PveManager.scala:
##########
@@ -0,0 +1,544 @@
+package org.apache.texera.web.resource.pythonvirtualenvironment
+/*
+ * 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
Review Comment:
Apache license header is placed after the `package` declaration in this
file, whereas other Scala files in the repo place the license at the very top.
This can break automated license header checks. Move the license comment block
above the `package` line.
##########
amber/src/main/scala/org/apache/texera/web/resource/pythonvirtualenvironment/PveManager.scala:
##########
@@ -0,0 +1,544 @@
+package org.apache.texera.web.resource.pythonvirtualenvironment
+/*
+ * 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 java.io.{File, RandomAccessFile}
+import java.nio.charset.StandardCharsets
+import java.nio.file.{Files, Path, Paths, StandardOpenOption}
+import java.util.concurrent.BlockingQueue
+import scala.collection.mutable.Map
+import scala.jdk.CollectionConverters._
+import scala.sys.process._
+
+object PveManager {
+
+ private val VenvRoot: Path = Paths.get("/tmp/texera-pve/venvs")
+
+ private def ensureDirExists(path: Path): Unit = {
+ if (!Files.exists(path)) Files.createDirectories(path)
+ }
+
+ private def cuidDir(cuid: Int, pvename: String): Path = {
+ ensureDirExists(VenvRoot)
+ val cuIdDir = VenvRoot.resolve(cuid.toString)
+ ensureDirExists(cuIdDir)
+
+ val dir = cuIdDir.resolve(pvename)
+ ensureDirExists(dir)
+
+ dir
+ }
+
+ private def pveDir(cuid: Int, pveName: String): Path =
+ cuidDir(cuid, pveName).resolve("pve")
+
+ private def pythonBinPath(cuid: Int, pveName: String): Path =
+ pveDir(cuid, pveName).resolve("bin").resolve("python")
+
+ private def pipBinPath(cuid: Int, pveName: String): Path =
+ pveDir(cuid, pveName).resolve("bin").resolve("pip")
+
+ private def metadataDir(cuid: Int, pveName: String): Path =
+ pveDir(cuid, pveName).resolve("metadata")
+
+ private def systemPackagesPath(cuid: Int, pveName: String): Path =
+ metadataDir(cuid, pveName).resolve("system-packages.txt")
+
+ private def userPackagesPath(cuid: Int, pveName: String): Path =
+ metadataDir(cuid, pveName).resolve("user-packages.txt")
+
+ private def ensureParentDir(path: Path): Unit = {
+ val parent = path.getParent
+ if (parent != null && !Files.exists(parent))
Files.createDirectories(parent)
+ }
+
+ private def writeMetadata(path: Path, lines: Seq[String]): Unit = {
+ ensureParentDir(path)
+ Files.write(
+ path,
+ lines.asJava,
+ StandardOpenOption.CREATE,
+ StandardOpenOption.TRUNCATE_EXISTING,
+ StandardOpenOption.WRITE
+ )
+ }
+
+ private def readMetadataList(path: Path): List[String] = {
+ if (!Files.exists(path)) return Nil
+ Files.readAllLines(path).asScala.map(_.trim).filter(_.nonEmpty).toList
+ }
+
+ private def parsePackageName(line: String): String =
+ line.split("==", 2).headOption.getOrElse(line).trim.toLowerCase
+
+ private def pipEnv: Map[String, String] =
+ Map(
+ "PYTHONUNBUFFERED" -> "1",
+ "PIP_PROGRESS_BAR" -> "off",
+ "PIP_DISABLE_PIP_VERSION_CHECK" -> "1",
+ "PIP_NO_INPUT" -> "1"
+ )
+
+ def pythonBin(cuid: Int, pveName: String): String =
+ pythonBinPath(cuid, pveName).toAbsolutePath.toString
+
+ def getSystemAndUserPackages(cuid: Int, pveName: String): (Seq[String],
Seq[String]) = {
+ val sys = readMetadataList(systemPackagesPath(cuid, pveName))
+ val usr = readMetadataList(userPackagesPath(cuid, pveName))
+ (sys, usr)
+ }
+
+ def deletePackages(cuid: Int, packageName: String, pveName: String):
List[String] = {
+ val pipPath = pipBinPath(cuid, pveName).toAbsolutePath
+ val userFile = userPackagesPath(cuid, pveName)
+ val systemFile = systemPackagesPath(cuid, pveName)
+
+ val systemPackages: Set[String] =
+ readMetadataList(systemFile).map(parsePackageName).toSet
+
+ val normalizedName = packageName.toLowerCase
+ if (systemPackages.contains(normalizedName)) {
+ return List(s"ERROR: '$packageName' is a system package and cannot be
deleted.")
+ }
+
+ if (!Files.exists(pipPath)) {
+ val msg = s"[PveManager] No pip found at $pipPath β PVE may not exist or
is not initialized."
+ println(msg)
+ return List(msg)
+ }
+
+ try {
+ val command = Seq(pipPath.toString, "uninstall", "-y", packageName)
+
+ val logger = ProcessLogger(
+ (out: String) => println(s"[pip] $out"),
+ (err: String) => System.err.println(s"[pip][ERR] $err")
+ )
+
+ val exitCode = command.!(logger)
+
+ if (exitCode == 0) {
+ val existing = readMetadataList(userFile)
+ val updated =
+ existing.filterNot(line => parsePackageName(line) ==
normalizedName).sorted
+
+ writeMetadata(userFile, updated)
+
+ List(s"Exit code: $exitCode", s"Uninstalled $packageName successfully")
+ } else {
+ List(s"[PveManager] pip uninstall for '$packageName' failed with exit
code $exitCode")
+ }
+ } catch {
+ case e: Exception =>
+ List(s"[PveManager] Failed to delete package for cuid=$cuid:
${e.getMessage}")
+ }
+ }
+
+ private def tailFileToQueue(
+ file: File,
+ queue: BlockingQueue[String],
+ prefix: String = "[pip] "
+ ): AutoCloseable = {
+ val raf = new RandomAccessFile(file, "r")
+ raf.seek(raf.length())
+ @volatile var running = true
+
+ val thread = new Thread(() => {
+ var buf = new StringBuilder
+ val charset = StandardCharsets.UTF_8
+ try {
+ while (running) {
+ val available = raf.length() - raf.getFilePointer
+ if (available > 0) {
+ val bytes = new Array[Byte](math.min(available, 8192).toInt)
+ val n = raf.read(bytes)
+ if (n > 0) {
+ buf.append(new String(bytes, 0, n, charset))
+ var newlineIndex = buf.indexOf("\n")
+ while (newlineIndex >= 0) {
+ val line = buf.substring(0, newlineIndex).trim
+ if (shouldStream(line)) queue.put(prefix + line)
+ buf = buf.delete(0, newlineIndex + 1)
+ newlineIndex = buf.indexOf("\n")
+ }
+ }
+ } else {
+ Thread.sleep(100)
+ }
+ }
+ val last = buf.result().trim
+ if (shouldStream(last)) queue.put(prefix + last)
+ } catch {
+ case _: InterruptedException => ()
+ case e: Exception => queue.put(s"[pip][ERR] tail exception:
${e.getMessage}")
+ } finally {
+ raf.close()
+ }
+ })
+
+ thread.setDaemon(true)
+ thread.start()
+
+ new AutoCloseable {
+ override def close(): Unit = {
+ running = false
+ thread.interrupt()
+ thread.join(500)
+ }
+ }
+ }
+
+ private def shouldStream(line: String): Boolean = {
+ val s = line.trim
+ if (s.isEmpty) return false
+
+ val lower = s.toLowerCase
+
+ if (lower.contains("found link")) return false
+ if (lower.contains("skipping link")) return false
+ if (lower.contains("cache")) return false
+ if (lower.contains("caching")) return false
+
+ true
+ }
+
+ private def runPipWithLog(
+ cmd: Seq[String],
+ env: Map[String, String],
+ queue: BlockingQueue[String]
+ ): Int = {
+ val logFile = File.createTempFile("pip-live-", ".log")
+ val fullCmd = if (cmd.contains("--log")) cmd else cmd ++ Seq("--log",
logFile.getAbsolutePath)
+
+ val tailer = tailFileToQueue(logFile, queue)
+
+ val logger = ProcessLogger(
+ out => if (shouldStream(out)) queue.put(s"[pip/stdout] $out"),
+ err => if (shouldStream(err)) queue.put(s"[pip/stderr] $err")
+ )
+
+ val proc = Process(fullCmd, None, env.toSeq: _*).run(logger)
+ val exitCode = proc.exitValue()
+
+ try tailer.close()
+ catch { case _: Throwable => () }
+
+ queue.put(s"[pip] (log at ${logFile.getAbsolutePath})")
+ exitCode
+ }
+
+ def createNewPve(cuid: Int, queue: BlockingQueue[String], pveName: String):
Unit = {
+ queue.put(s"[PVE] Creating new PVE for cuid=$cuid with name=$pveName")
+
+ val venvDirPath = pveDir(cuid, pveName).toAbsolutePath
+ ensureDirExists(cuidDir(cuid, pveName))
+
+ val python = pythonBinPath(cuid, pveName).toAbsolutePath.toString
+ val envVars = pipEnv
+
+ val Requirements: String =
+ """wheel==0.41.2
+ |setuptools==80.10.2
+ |numpy==2.1.0
+ |pandas==2.2.3
+ |ruff==0.14.7
+ |iniconfig==1.1.1
+ |loguru==0.7.0
+ |pyarrow==21.0.0
+ |pytest==7.4.0
+ |python-dateutil==2.8.2
+ |pytest-timeout==2.2.0
+ |protobuf==4.25.8
+ |betterproto==2.0.0b7
+ |typing==3.7.4.3
+ |pampy==0.3.0
+ |overrides==7.4.0
+ |typing_extensions==4.10.0
+ |pytest-reraise==2.1.2
+ |dataclasses==0.6
+ |Deprecated==1.2.14
+ |fs==2.4.16
+ |python-lsp-server[all]==1.12.0
+ |python-lsp-server[websockets]==1.12.0
+ |bidict==0.22.0
+ |cached_property==1.5.2
+ |psutil==5.9.0
+ |tzlocal==2.1
+ |pyiceberg==0.8.1
+ |readerwriterlock==1.0.9
+ |tenacity==8.5.0
+ |SQLAlchemy==2.0.37
+ |pg8000==1.31.5
+ |pympler==1.1
+ |""".stripMargin
+
+ val OperatorRequirements: String =
+ """|wordcloud==1.9.3
+ |plotly==5.24.1
+ |praw==7.6.1
+ |pillow==10.2.0
+ |pybase64==1.3.2
+ |torch==2.8.0
+ |scikit-learn==1.5.0
+ |transformers==4.57.3
+ |boto3==1.40.53
+ |""".stripMargin
+
+ val pveBase = sys.env.getOrElse("PVE_BASE", "/opt/pve-base")
+ val basePython = Paths.get(pveBase).resolve("bin").resolve("python")
+
+ val hasBasePve = Files.exists(basePython)
+
+ if (!hasBasePve) {
+ if (Files.exists(venvDirPath)) {
+ val rmCode = Process(Seq("bash", "-lc", s"rm -rf
'${venvDirPath.toString}'")).!(
+ ProcessLogger(
+ out => queue.put(s"[pve] $out"),
+ err => queue.put(s"[pve][ERR] $err")
+ )
+ )
+ queue.put(s"[pve] removed existing venv with exit code $rmCode")
Review Comment:
This uses `bash -lc` with an interpolated path inside single quotes (`rm -rf
'${venvDirPath}'`). If `pveName` contains a single quote or shell
metacharacters, this becomes command injection. Avoid invoking a shell here
(use `java.nio.file` deletion or `Process(Seq("rm","-rf", ...))`), and validate
`pveName`.
##########
frontend/src/app/workspace/component/power-button/computing-unit-selection.component.html:
##########
@@ -427,3 +435,177 @@
</div>
</div>
</ng-template>
+
+<!-- Modal for adding packages -->
+<nz-modal
+ nzWrapClassName="pve-modal"
+ nzClassName="pve-modal"
+ [nzVisible]="pveModalVisible"
+ nzTitle="Python Environments"
+ (nzOnCancel)="pveModalVisible = false"
+ [nzFooter]="customFooter"
+>
+ <ng-template #customFooter>
+ <div class="footer-all">
+ <button nz-button nzType="default" (click)="pveModalVisible =
false">Close</button>
Review Comment:
Closing the PVE modal via `nzOnCancel` only sets `pveModalVisible=false` and
does not close any active `EventSource` streams. This can leave pip installs
streaming in the background and leak browser/server resources. Call a close
handler on cancel that iterates over `pves` and closes any `pve.source` (and
resets `isInstalling`).
```suggestion
(nzOnCancel)="closePveModal()"
[nzFooter]="customFooter"
>
<ng-template #customFooter>
<div class="footer-all">
<button nz-button nzType="default"
(click)="closePveModal()">Close</button>
```
##########
amber/src/test/scala/org/apache/texera/web/resource/pythonvirtualenvironment/PveResourceSpec.scala:
##########
@@ -0,0 +1,116 @@
+/*
+ * 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.web.resource.pythonvirtualenvironment
+
+import org.scalatest.BeforeAndAfterEach
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+import java.nio.file.{Files, Path, Paths}
+import java.util.concurrent.LinkedBlockingQueue
+import scala.jdk.CollectionConverters._
+
+class PveManagerSpec extends AnyFlatSpec with Matchers with BeforeAndAfterEach
{
+
+ private val testCuid = 256
+ private var testPveName: String = _
+ private var testRoot: Path = _
+ private var queue: LinkedBlockingQueue[String] = _
+
+ override protected def beforeEach(): Unit = {
+ testPveName = s"test-env-${System.currentTimeMillis()}"
+ testRoot = Paths.get("/tmp/texera-pve/venvs").resolve(testCuid.toString)
+ queue = new LinkedBlockingQueue[String]()
+ }
+
+ override protected def afterEach(): Unit = {
+ deleteRecursively(testRoot)
+ }
+
+ private def deleteRecursively(path: Path): Unit = {
+ if (path != null && Files.exists(path)) {
+ Files
+ .walk(path)
+ .iterator()
+ .asScala
+ .toList
+ .sortBy(_.toString.length)
+ .reverse
+ .foreach(Files.deleteIfExists)
+ }
+ }
+
+ private def queueMessages(): List[String] = {
+ queue.iterator().asScala.toList
+ }
+
+ private def queueText(): String = {
+ queueMessages().mkString("\n")
+ }
+
+ "PveManager" should "create a real pve, install a package, and uninstall it"
in {
+ PveManager.createNewPve(testCuid, queue, testPveName)
+
+ val createLogs = queueText()
+
+ createLogs should not include "[PVE][ERR]"
+ PveManager.pveExists(testCuid, testPveName) shouldBe true
+
+ val pythonPath = Paths.get(PveManager.pythonBin(testCuid, testPveName))
+ val pipPath =
testRoot.resolve(testPveName).resolve("pve").resolve("bin").resolve("pip")
+ val metadataDir =
testRoot.resolve(testPveName).resolve("pve").resolve("metadata")
+
+ Files.exists(pythonPath) shouldBe true
+ Files.exists(pipPath) shouldBe true
+ Files.exists(metadataDir.resolve("system-packages.txt")) shouldBe true
+ Files.exists(metadataDir.resolve("user-packages.txt")) shouldBe true
+
+ PveManager.getEnvironments(testCuid) should contain(testPveName)
+
+ val (_, userPackagesBeforeInstall) =
+ PveManager.getSystemAndUserPackages(testCuid, testPveName)
+
+ userPackagesBeforeInstall shouldBe empty
+
+ val packageToInstall = "scanpy==1.11.1"
+
+ PveManager.installPackages(
Review Comment:
This test installs `scanpy==1.11.1`, which is very large and will make CI
slow/flaky (network, build deps, long install times). Use a lightweight
pure-Python package (or mock/stub pip), and/or mark this as an integration test
excluded from the default unit-test suite.
##########
frontend/src/app/workspace/component/power-button/computing-unit-selection.component.ts:
##########
@@ -637,4 +670,281 @@ export class ComputingUnitSelectionComponent implements
OnInit {
this.computingUnitStatusService.refreshComputingUnitList();
}
}
+
+ private makeEmptyPve(expanded: boolean): PveDraft {
+ return {
+ id: this.nextPveId++,
+ name: "",
+ userPackages: [],
+ newPackages: [{ name: "", operator: "==", version: "" }],
+ deletingPackages: [],
+ pipOutput: "",
+ prettyPipOutput: "",
+ expanded,
+ isInstalling: false,
+ };
+ }
+
+ addEnvironment(): void {
+ this.pves.push(this.makeEmptyPve(true));
+ }
+
+ trackByPveId(_index: number, pve: PveDraft): number {
+ return pve.id;
+ }
+
+ showPVEmodalVisible(): void {
+ this.pveModalVisible = true;
+ this.getPVEs();
+ }
+
+ getPVEs(): void {
+ const cuId: number | undefined =
this.selectedComputingUnit?.computingUnit.cuid;
+
+ if (cuId == null) {
+ this.notificationService.error("No computing unit selected. Please
select a CU first.");
+ return;
+ }
+
+ this.workflowPveService.setCuid(cuId);
+
+ this.workflowPveService
+ .fetchPVEs(cuId)
+ .pipe(untilDestroyed(this))
+ .subscribe({
+ next: (resp: PvePackageResponse[]) => {
+ this.pves = resp.map((pve, index) => ({
+ id: index,
+ name: pve.pveName,
+ expanded: false,
+ isInstalling: false,
+ pipOutput: "",
+ prettyPipOutput: "",
+ userPackages: (pve.userPackages ?? []).map(pkg => {
+ const [name, version] = pkg.split("==");
+ return {
+ name: name.trim(),
+ version: (version ?? "").trim(),
+ isHighlighted: false,
+ };
+ }),
+ newPackages: [],
+ deletingPackages: [],
+ }));
+ },
+ error: (err: unknown) => {
+ console.error("Failed to fetch PVEs:", err);
+ this.pves = [];
+ },
+ });
+ }
+
+ addPackage(index: number): void {
+ const env = this.pves[index];
+ env.newPackages.push({ name: "", version: "", operator: undefined,
isHighlighted: false });
+ }
+
+ togglePackageDelete(index: number, pkg: PackageRow): void {
+ const env = this.pves[index];
+
+ pkg.isHighlighted = !pkg.isHighlighted;
+
+ const version = pkg.version ?? "";
+
+ if (pkg.isHighlighted) {
+ const exists = env.deletingPackages.some(p => p.name === pkg.name &&
(p.version ?? "") === version);
+ if (!exists) {
+ env.deletingPackages.push({ name: pkg.name, version });
+ }
+ } else {
+ env.deletingPackages = env.deletingPackages.filter(p => !(p.name ===
pkg.name && (p.version ?? "") === version));
+ }
+ }
+
+ scrollToBottomOfPipModal(index: number) {
+ setTimeout(() => {
+ const pre = document.getElementById(`pip-log-${index}`) as HTMLElement |
null;
+ if (pre) {
+ pre.scrollTop = pre.scrollHeight;
+ }
+ }, 50);
+ }
+
+ updatePrettyPipOutput(index: number) {
+ const env = this.pves[index];
+
+ const escapeHtml = (s: string) =>
+ s
+ .replace(/&/g, "&")
+ .replace(/</g, "<")
+ .replace(/>/g, ">")
+ .replace(/"/g, """)
+ .replace(/'/g, "'");
+
+ const raw = env.pipOutput ?? "";
+ const safe = escapeHtml(raw);
+
+ env.prettyPipOutput = safe
+ .replace(
+ /^(\[pip\].*finished with exit code\s+0.*)$/gm,
+ "<span class=\"pip-exit ok\"><strong>$1</strong></span>"
+ )
+ .replace(
+ /^(\[pip\].*finished with exit code\s+1.*)$/gm,
+ "<span class=\"pip-exit err\"><strong>$1</strong></span>"
+ )
+ .replace(
+ /^(\[pip\].*finished with exit code\s+([2-9]\d*).*)$/gm,
+ "<span class=\"pip-exit err\"><strong>$1</strong></span>"
+ )
+ .replace(/ERROR/g, "<span class=\"error\">ERROR</span>")
+ .replace(/WARNING/g, "<span class=\"warning\">WARNING</span>")
+ .replace(/already satisfied/g, "<span class=\"success\">already
satisfied</span>")
+ .replace(/\n/g, "<br/>");
+ }
+
+ createVirtualEnvironment(index: number): void {
+ const cuId = this.selectedComputingUnit?.computingUnit.cuid;
+ const env = this.pves[index];
+
+ if (cuId == null) {
+ this.notificationService.error("No computing unit selected. Please
select a CU first.");
+ return;
+ }
+
+ if (!env.name?.trim()) {
+ this.notificationService.error("Environment name cannot be empty.");
+ return;
+ }
+
+ env.isInstalling = true;
+
+ this.workflowPveService.setCuid(cuId);
+ this.workflowPveService.setPveName(env.name);
+
+ for (const pkg of env.deletingPackages) {
+ this.workflowPveService
+ .deletePackage(pkg.name)
+ .pipe(untilDestroyed(this))
+ .subscribe({
+ next: () => {
+ env.pipOutput += `Starting ...\nSuccessfully uninstalled package
${pkg.name}`;
+ this.updatePrettyPipOutput(index);
+ this.scrollToBottomOfPipModal(index);
+
+ this.workflowPveService
+ .getInstalledPackages()
+ .pipe(untilDestroyed(this))
+ .subscribe({
+ next: resp => {
+ this.systemPackages = resp.system.map(pkgStr => {
+ const [name, version] = pkgStr.split("==");
+ return { name: name.trim(), version: (version ?? "").trim() };
+ });
+
+ env.userPackages = resp.user.map(pkgStr => {
+ const [name, version] = pkgStr.split("==");
+ return { name: name.trim(), version: (version ?? "").trim(),
isHighlighted: false };
+ });
+
+ env.deletingPackages = [];
+ },
+ error: (e: unknown) => console.error("Failed to refresh packages
after delete", e),
+ });
+ },
+ error: (err: unknown) => {
+ console.error("Error deleting package:", err);
+ env.pipOutput += `\nError uninstalling ${pkg.name}`;
+ this.updatePrettyPipOutput(index);
+ this.scrollToBottomOfPipModal(index);
+ },
+ });
+ }
Review Comment:
Uninstalls are fired asynchronously in a `for` loop and the method continues
immediately to start the install SSE. This can overlap uninstall/install and
produces inconsistent UI state. Consider sequencing these operations (e.g.,
`concatMap`/`forkJoin` uninstall observables before starting the SSE install).
##########
frontend/src/app/workspace/component/power-button/computing-unit-selection.component.html:
##########
@@ -160,6 +160,14 @@
(click)="startEditingUnitName(unit); $event.stopPropagation()"
role="button">
</i>
+ <i
+ nz-icon
+ nzType="plus"
+ nz-tooltip
+ *ngIf="unit.isOwner"
+ [nzTooltipTitle]="'Python Environment'"
+ (click)="showPVEmodalVisible()">
Review Comment:
Clicking the Python Environment icon will bubble to the `<li>` click
handler, but this icon's handler runs first. Since `showPVEmodalVisible()`
relies on `selectedComputingUnit`, the modal can open/fetch PVEs for the
previously-selected CU. Pass the `unit` into the handler and select/set the CU
before opening the modal (or stop propagation and explicitly select).
```suggestion
(click)="selectedComputingUnit = unit; showPVEmodalVisible();
$event.stopPropagation()">
```
##########
frontend/src/app/workspace/component/power-button/computing-unit-selection.component.html:
##########
@@ -427,3 +435,177 @@
</div>
</div>
</ng-template>
+
+<!-- Modal for adding packages -->
+<nz-modal
+ nzWrapClassName="pve-modal"
+ nzClassName="pve-modal"
+ [nzVisible]="pveModalVisible"
+ nzTitle="Python Environments"
+ (nzOnCancel)="pveModalVisible = false"
+ [nzFooter]="customFooter"
+>
+ <ng-template #customFooter>
+ <div class="footer-all">
+ <button nz-button nzType="default" (click)="pveModalVisible =
false">Close</button>
+ <button nz-button nzType="primary" (click)="addEnvironment()">Add
Environment</button>
+ </div>
+ </ng-template>
+
+ <ng-container *nzModalContent>
+
+ <!-- Shared system packages -->
+ <div class="system-section">
+ <nz-collapse>
+ <nz-collapse-panel nzHeader="System Packages (read-only)">
+ <div class="system-panel-inner">
+ <div *ngFor="let pkg of systemPackages" class="package-row
system-row">
+ <div class="package-inputs">
+ <div class="field">
+ <label>Package</label>
+ <input nz-input class="system-input" [disabled]="true"
[ngModel]="pkg.name" />
+ </div>
+ <div class="field">
+ <label>Version</label>
+ <input nz-input class="system-input" [disabled]="true"
[ngModel]="pkg.version" />
+ </div>
+ </div>
+ </div>
+ </div>
+ </nz-collapse-panel>
+ </nz-collapse>
+ </div>
+
+ <!-- Environments -->
+ <nz-collapse>
+ <nz-collapse-panel
+ *ngFor="let pve of pves; let envIndex = index; trackBy: trackByPveId"
+ [nzActive]="pve.expanded"
+ (nzActiveChange)="pve.expanded = $event"
+ [nzHeader]="headerTpl"
+ >
+ <!-- Custom header -->
+ <ng-template #headerTpl>
+ <div class="env-header">
+ <span class="env-title">
+ {{ pve.name }}
+ </span>
+ <span class="env-actions" (click)="$event.stopPropagation()">
+ </span>
+ </div>
+ </ng-template>
+
+ <!-- Panel body -->
+ <div class="ve-form">
+
+ <div class="fieldRow">
+ <label class="fieldLabel">Virtual Environment Name</label>
+ <input nz-input class="fieldInput" placeholder="Environment Name"
[(ngModel)]="pve.name" />
+ </div>
+
+ <!-- USER PACKAGES -->
+ <div class="section-title" *ngIf="pve.userPackages.length > 0">
+ Installed User Packages
+ </div>
+
+ <div *ngFor="let pkg of pve.userPackages; let i = index"
class="package-row">
+ <div class="package-inputs">
+ <div class="field">
+ <label>Package</label>
+ <input nz-input [ngModel]="pkg.name" [disabled]="true" />
+ </div>
+
+ <div class="field">
+ <label>Version</label>
+ <input nz-input [ngModel]="pkg.version" [disabled]="true" />
+ </div>
+ </div>
+
+ <button
+ nz-button
+ nzType="default"
+ nzShape="circle"
+ nzDanger
+ [ngClass]="{ 'highlighted-btn': pkg.isHighlighted }"
+ (click)="togglePackageDelete(envIndex, pkg)"
+ >
+ <i nz-icon nzType="delete"></i>
+ </button>
+ </div>
+
+ <!-- NEW PACKAGES -->
+ <div *ngFor="let pkg of pve.newPackages; let i = index"
class="package-row">
+ <div class="package-inputs">
+ <div class="field">
+ <label>Package</label>
+ <input nz-input placeholder="Package Name"
[(ngModel)]="pve.newPackages[i].name" />
+ </div>
+
+ <div class="field operator operator-select">
+ <label style="visibility: hidden;">Op</label>
+ <nz-select nzPlaceHolder="Select" nzCentered
[(ngModel)]="pve.newPackages[i].operator">
+ <nz-option nzValue="==" nzLabel="=="></nz-option>
+ <nz-option nzValue=">=" nzLabel=">="></nz-option>
+ <nz-option nzValue="<=" nzLabel="<="></nz-option>
+ </nz-select>
+ </div>
+
+ <div class="field">
+ <label>Version</label>
+ <input nz-input placeholder="Package Version (optional)"
[(ngModel)]="pve.newPackages[i].version" />
+ </div>
+ </div>
+
+ <button
+ nz-button
+ nzType="default"
+ nzShape="circle"
+ nzDanger
+ [ngClass]="{ 'highlighted-btn': pkg.isHighlighted }"
+ (click)="togglePackageDelete(envIndex, pkg)"
Review Comment:
The delete button for *new package* rows calls
`togglePackageDelete(envIndex, pkg)`, which adds the row to `deletingPackages`
instead of removing the input row. This is likely unintended UX and can trigger
uninstall calls for packages the user only meant to remove from the form. Add a
separate handler to remove `pve.newPackages[i]`.
```suggestion
(click)="pve.newPackages.splice(i, 1)"
```
##########
amber/src/test/scala/org/apache/texera/web/resource/pythonvirtualenvironment/PveResourceSpec.scala:
##########
@@ -0,0 +1,116 @@
+/*
+ * 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.web.resource.pythonvirtualenvironment
+
+import org.scalatest.BeforeAndAfterEach
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+import java.nio.file.{Files, Path, Paths}
+import java.util.concurrent.LinkedBlockingQueue
+import scala.jdk.CollectionConverters._
+
+class PveManagerSpec extends AnyFlatSpec with Matchers with BeforeAndAfterEach
{
Review Comment:
File name is `PveResourceSpec.scala` but the test class is `PveManagerSpec`.
This mismatch makes it harder to find tests and is inconsistent with typical
Scala test naming. Consider renaming the file or the class so they match.
```suggestion
class PveResourceSpec extends AnyFlatSpec with Matchers with
BeforeAndAfterEach {
```
##########
amber/src/main/scala/org/apache/texera/web/resource/pythonvirtualenvironment/PveManager.scala:
##########
@@ -0,0 +1,544 @@
+package org.apache.texera.web.resource.pythonvirtualenvironment
+/*
+ * 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 java.io.{File, RandomAccessFile}
+import java.nio.charset.StandardCharsets
+import java.nio.file.{Files, Path, Paths, StandardOpenOption}
+import java.util.concurrent.BlockingQueue
+import scala.collection.mutable.Map
+import scala.jdk.CollectionConverters._
+import scala.sys.process._
+
+object PveManager {
+
+ private val VenvRoot: Path = Paths.get("/tmp/texera-pve/venvs")
+
+ private def ensureDirExists(path: Path): Unit = {
+ if (!Files.exists(path)) Files.createDirectories(path)
+ }
+
+ private def cuidDir(cuid: Int, pvename: String): Path = {
+ ensureDirExists(VenvRoot)
+ val cuIdDir = VenvRoot.resolve(cuid.toString)
+ ensureDirExists(cuIdDir)
+
+ val dir = cuIdDir.resolve(pvename)
Review Comment:
`pveName` is used directly in `cuIdDir.resolve(pvename)` and directories are
created. Without validation, values like `../...` or path separators can escape
the intended root and write/delete files outside the PVE directory. Validate
`pveName` against a strict allowlist and reject `/`, `\\`, or `..` segments
before using it in filesystem paths.
```suggestion
private val ValidPveNamePattern = "^[A-Za-z0-9._-]+$"
private def ensureDirExists(path: Path): Unit = {
if (!Files.exists(path)) Files.createDirectories(path)
}
private def validatePveName(pveName: String): String = {
require(
pveName != null &&
pveName.nonEmpty &&
pveName.matches(ValidPveNamePattern) &&
!pveName.contains("/") &&
!pveName.contains("\\") &&
pveName != "." &&
pveName != ".." &&
!pveName.contains(".."),
s"Invalid pveName: $pveName"
)
pveName
}
private def cuidDir(cuid: Int, pvename: String): Path = {
val validatedPveName = validatePveName(pvename)
ensureDirExists(VenvRoot)
val cuIdDir = VenvRoot.resolve(cuid.toString)
ensureDirExists(cuIdDir)
val dir = cuIdDir.resolve(validatedPveName)
```
##########
frontend/src/app/workspace/component/power-button/computing-unit-selection.component.html:
##########
@@ -427,3 +435,177 @@
</div>
</div>
</ng-template>
+
+<!-- Modal for adding packages -->
+<nz-modal
+ nzWrapClassName="pve-modal"
+ nzClassName="pve-modal"
+ [nzVisible]="pveModalVisible"
+ nzTitle="Python Environments"
+ (nzOnCancel)="pveModalVisible = false"
+ [nzFooter]="customFooter"
+>
+ <ng-template #customFooter>
+ <div class="footer-all">
+ <button nz-button nzType="default" (click)="pveModalVisible =
false">Close</button>
+ <button nz-button nzType="primary" (click)="addEnvironment()">Add
Environment</button>
Review Comment:
The footer "Close" button also just sets `pveModalVisible=false` without
cleaning up active `EventSource` streams. Reuse the same cleanup handler as
`nzOnCancel` so closing the modal always stops streaming.
##########
frontend/src/styles.scss:
##########
@@ -107,3 +107,253 @@ hr {
position: relative;
left: 0;
}
+
+
+// pip modal
+.pve-modal {
+ .ant-modal {
+ max-width: 980px;
+ width: 92vw !important;
+ }
+
+ .ant-modal-body {
+ padding: 16px 20px 18px;
+ background: #fafafa;
+ }
+
+ .ant-modal-header {
+ padding: 14px 20px;
+ }
+
+ .ant-modal-title {
+ font-weight: 600;
+ letter-spacing: 0.2px;
+ }
+
+ .footer-all {
+ display: flex;
+ justify-content: space-between;
+ align-items: center;
+ width: 100%;
+ gap: 10px;
+ padding: 8px 0;
+ }
+
+ nz-collapse {
+ display: block;
+ }
+
+ .ant-collapse {
+ border-radius: 12px;
+ overflow: hidden;
+ background: transparent;
+ }
+
+ .system-section {
+ margin-bottom: 14px;
+ }
+
+ .system-section .ant-collapse-item {
+ //border-radius: 12px;
+ overflow: hidden;
+ background: #ffffff;
+ border: 1px solid #eef0f3;
+ box-shadow: 0 1px 2px rgba(0, 0, 0, 0.03);
+ }
+
+ .system-section .ant-collapse-header {
+ font-weight: 600;
+ }
+
+ .system-panel-inner {
+ padding: 6px 0;
+ }
+
+ .system-row {
+ opacity: 0.9;
+ }
+
+ .system-input {
+ background: #f5f6f8 !important;
+ border-color: #e6e8ec !important;
+ color: #5a667a;
+ cursor: not-allowed;
+ }
+
+ .env-header {
+ width: 100%;
+ display: flex;
+ align-items: center;
+ justify-content: space-between;
+ gap: 12px;
+ }
+
+ .env-title {
+ font-weight: 600;
+ font-size: 14px;
+ color: #1f2a37;
+ min-width: 0;
+ overflow: hidden;
+ text-overflow: ellipsis;
+ white-space: nowrap;
+ }
+
+ .env-actions {
+ display: inline-flex;
+ align-items: center;
+ gap: 8px;
+ flex-shrink: 0;
+ }
+
+ .ant-collapse-item {
+ background: #ffffff;
+ border: 1px solid #eef0f3;
+ //border-radius: 12px;
+ overflow: hidden;
+ margin-bottom: 12px;
+ box-shadow: 0 1px 2px rgba(0, 0, 0, 0.03);
+ }
+
+ .ant-collapse-header {
+ padding: 12px 14px !important;
+ align-items: center !important;
+ }
+
+ .ant-collapse-content-box {
+ padding: 14px !important;
+ }
+
+ .ve-form {
+ display: flex;
+ flex-direction: column;
+ gap: 14px;
+ }
+
+ .fieldRow {
+ display: flex !important;
+ align-items: center !important;
+ gap: 12px !important;
+ }
+
+ .fieldLabel {
+ width: 220px;
+ margin: 0;
+ font-weight: 700;
+ white-space: nowrap;
+ }
+
+ .fieldInput {
+ flex: 1;
+ min-width: 0;
+ }
+
+ .package-row {
+ display: flex;
+ align-items: flex-end;
+ justify-content: space-between;
+ gap: 10px;
+ padding: 10px 10px;
+ border: 1px solid #eef0f3;
+ //border-radius: 12px;
+ background: #ffffff;
+ }
+
+ .package-inputs {
+ display: grid;
+ grid-template-columns: 1fr 140px 1fr;
+ gap: 10px;
+ flex: 1;
+ min-width: 0;
+ }
+
+ .field {
+ display: flex;
+ flex-direction: column;
+ gap: 6px;
+ min-width: 0;
+ }
+
+ .field label {
+ font-size: 11px;
+ font-weight: 600;
+ color: #6b7280;
+ line-height: 1;
+ }
+
+ .operator-select .ant-select {
+ width: 100%;
+ }
+
+ .ant-input,
+ .ant-select-selector {
+ //border-radius: 10px !important;
+ }
+
+ .ant-input[disabled] {
+ background: #f5f6f8 !important;
+ border-color: #e6e8ec !important;
+ color: #5a667a;
+ }
+
+ .highlighted-btn {
+ background-color: #ff4d4f !important; /* Ant Design red */
+ border-color: #ff4d4f !important;
+ color: white !important;
+ }
+
+ .add-btn {
+ display: flex;
+ justify-content: flex-start;
+ margin-top: -6px;
+ }
+
+ .env-footer {
+ display: flex;
+ justify-content: flex-end;
+ padding-top: 6px;
+ }
+
+ .pip-panel {
+ margin-top: 16px;
+ border: 1px solid #d9d9d9;
+ //border-radius: 8px;
+ background: #f2f2f2;
+ overflow: hidden;
+ }
+
+ .pip-panel-header {
+ display: flex;
+ justify-content: space-between;
+ align-items: baseline;
+ padding: 10px 14px;
+ background: #e9e9e9;
+ border-bottom: 1px solid #d9d9d9;
+ }
+
+ .pip-panel-title {
+ font-weight: 600;
+ color: #222;
+ }
+
+ .pip-panel-subtitle {
+ font-size: 12px;
+ color: #666;
+ }
+
+ .pip-panel-body {
+ padding: 0;
+ }
+
+ .pip-fullscreen-log {
+ color: #333;
+ font-family: "JetBrains Mono", monospace;
+ font-size: 13px;
+ line-height: 1.6;
Review Comment:
The modal log output injects spans with classes like `pip-exit ok/err`,
`error`, `warning`, and `success` (from `updatePrettyPipOutput`), but no styles
for these classes exist under `.pve-modal`. If the intent is to highlight log
levels, add scoped styles (e.g., colors/weights) for those classes.
##########
amber/src/main/scala/org/apache/texera/web/resource/pythonvirtualenvironment/PveManager.scala:
##########
@@ -0,0 +1,544 @@
+package org.apache.texera.web.resource.pythonvirtualenvironment
+/*
+ * 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 java.io.{File, RandomAccessFile}
+import java.nio.charset.StandardCharsets
+import java.nio.file.{Files, Path, Paths, StandardOpenOption}
+import java.util.concurrent.BlockingQueue
+import scala.collection.mutable.Map
+import scala.jdk.CollectionConverters._
+import scala.sys.process._
+
+object PveManager {
+
+ private val VenvRoot: Path = Paths.get("/tmp/texera-pve/venvs")
+
+ private def ensureDirExists(path: Path): Unit = {
+ if (!Files.exists(path)) Files.createDirectories(path)
+ }
+
+ private def cuidDir(cuid: Int, pvename: String): Path = {
+ ensureDirExists(VenvRoot)
+ val cuIdDir = VenvRoot.resolve(cuid.toString)
+ ensureDirExists(cuIdDir)
+
+ val dir = cuIdDir.resolve(pvename)
+ ensureDirExists(dir)
+
+ dir
+ }
+
+ private def pveDir(cuid: Int, pveName: String): Path =
+ cuidDir(cuid, pveName).resolve("pve")
+
+ private def pythonBinPath(cuid: Int, pveName: String): Path =
+ pveDir(cuid, pveName).resolve("bin").resolve("python")
+
+ private def pipBinPath(cuid: Int, pveName: String): Path =
+ pveDir(cuid, pveName).resolve("bin").resolve("pip")
+
+ private def metadataDir(cuid: Int, pveName: String): Path =
+ pveDir(cuid, pveName).resolve("metadata")
+
+ private def systemPackagesPath(cuid: Int, pveName: String): Path =
+ metadataDir(cuid, pveName).resolve("system-packages.txt")
+
+ private def userPackagesPath(cuid: Int, pveName: String): Path =
+ metadataDir(cuid, pveName).resolve("user-packages.txt")
+
+ private def ensureParentDir(path: Path): Unit = {
+ val parent = path.getParent
+ if (parent != null && !Files.exists(parent))
Files.createDirectories(parent)
+ }
+
+ private def writeMetadata(path: Path, lines: Seq[String]): Unit = {
+ ensureParentDir(path)
+ Files.write(
+ path,
+ lines.asJava,
+ StandardOpenOption.CREATE,
+ StandardOpenOption.TRUNCATE_EXISTING,
+ StandardOpenOption.WRITE
+ )
+ }
+
+ private def readMetadataList(path: Path): List[String] = {
+ if (!Files.exists(path)) return Nil
+ Files.readAllLines(path).asScala.map(_.trim).filter(_.nonEmpty).toList
+ }
+
+ private def parsePackageName(line: String): String =
+ line.split("==", 2).headOption.getOrElse(line).trim.toLowerCase
+
+ private def pipEnv: Map[String, String] =
+ Map(
+ "PYTHONUNBUFFERED" -> "1",
+ "PIP_PROGRESS_BAR" -> "off",
+ "PIP_DISABLE_PIP_VERSION_CHECK" -> "1",
+ "PIP_NO_INPUT" -> "1"
+ )
+
+ def pythonBin(cuid: Int, pveName: String): String =
+ pythonBinPath(cuid, pveName).toAbsolutePath.toString
+
+ def getSystemAndUserPackages(cuid: Int, pveName: String): (Seq[String],
Seq[String]) = {
+ val sys = readMetadataList(systemPackagesPath(cuid, pveName))
+ val usr = readMetadataList(userPackagesPath(cuid, pveName))
+ (sys, usr)
+ }
+
+ def deletePackages(cuid: Int, packageName: String, pveName: String):
List[String] = {
+ val pipPath = pipBinPath(cuid, pveName).toAbsolutePath
+ val userFile = userPackagesPath(cuid, pveName)
+ val systemFile = systemPackagesPath(cuid, pveName)
+
+ val systemPackages: Set[String] =
+ readMetadataList(systemFile).map(parsePackageName).toSet
+
+ val normalizedName = packageName.toLowerCase
+ if (systemPackages.contains(normalizedName)) {
+ return List(s"ERROR: '$packageName' is a system package and cannot be
deleted.")
+ }
+
+ if (!Files.exists(pipPath)) {
+ val msg = s"[PveManager] No pip found at $pipPath β PVE may not exist or
is not initialized."
+ println(msg)
+ return List(msg)
+ }
+
+ try {
+ val command = Seq(pipPath.toString, "uninstall", "-y", packageName)
+
+ val logger = ProcessLogger(
+ (out: String) => println(s"[pip] $out"),
+ (err: String) => System.err.println(s"[pip][ERR] $err")
+ )
+
+ val exitCode = command.!(logger)
+
+ if (exitCode == 0) {
+ val existing = readMetadataList(userFile)
+ val updated =
+ existing.filterNot(line => parsePackageName(line) ==
normalizedName).sorted
+
+ writeMetadata(userFile, updated)
+
+ List(s"Exit code: $exitCode", s"Uninstalled $packageName successfully")
+ } else {
+ List(s"[PveManager] pip uninstall for '$packageName' failed with exit
code $exitCode")
+ }
+ } catch {
+ case e: Exception =>
+ List(s"[PveManager] Failed to delete package for cuid=$cuid:
${e.getMessage}")
+ }
+ }
+
+ private def tailFileToQueue(
+ file: File,
+ queue: BlockingQueue[String],
+ prefix: String = "[pip] "
+ ): AutoCloseable = {
+ val raf = new RandomAccessFile(file, "r")
+ raf.seek(raf.length())
+ @volatile var running = true
+
+ val thread = new Thread(() => {
+ var buf = new StringBuilder
+ val charset = StandardCharsets.UTF_8
+ try {
+ while (running) {
+ val available = raf.length() - raf.getFilePointer
+ if (available > 0) {
+ val bytes = new Array[Byte](math.min(available, 8192).toInt)
+ val n = raf.read(bytes)
+ if (n > 0) {
+ buf.append(new String(bytes, 0, n, charset))
+ var newlineIndex = buf.indexOf("\n")
+ while (newlineIndex >= 0) {
+ val line = buf.substring(0, newlineIndex).trim
+ if (shouldStream(line)) queue.put(prefix + line)
+ buf = buf.delete(0, newlineIndex + 1)
+ newlineIndex = buf.indexOf("\n")
+ }
+ }
+ } else {
+ Thread.sleep(100)
+ }
+ }
+ val last = buf.result().trim
+ if (shouldStream(last)) queue.put(prefix + last)
+ } catch {
+ case _: InterruptedException => ()
+ case e: Exception => queue.put(s"[pip][ERR] tail exception:
${e.getMessage}")
+ } finally {
+ raf.close()
+ }
+ })
+
+ thread.setDaemon(true)
+ thread.start()
+
+ new AutoCloseable {
+ override def close(): Unit = {
+ running = false
+ thread.interrupt()
+ thread.join(500)
+ }
+ }
+ }
+
+ private def shouldStream(line: String): Boolean = {
+ val s = line.trim
+ if (s.isEmpty) return false
+
+ val lower = s.toLowerCase
+
+ if (lower.contains("found link")) return false
+ if (lower.contains("skipping link")) return false
+ if (lower.contains("cache")) return false
+ if (lower.contains("caching")) return false
+
+ true
+ }
+
+ private def runPipWithLog(
+ cmd: Seq[String],
+ env: Map[String, String],
+ queue: BlockingQueue[String]
+ ): Int = {
+ val logFile = File.createTempFile("pip-live-", ".log")
+ val fullCmd = if (cmd.contains("--log")) cmd else cmd ++ Seq("--log",
logFile.getAbsolutePath)
+
+ val tailer = tailFileToQueue(logFile, queue)
+
+ val logger = ProcessLogger(
+ out => if (shouldStream(out)) queue.put(s"[pip/stdout] $out"),
+ err => if (shouldStream(err)) queue.put(s"[pip/stderr] $err")
+ )
+
+ val proc = Process(fullCmd, None, env.toSeq: _*).run(logger)
+ val exitCode = proc.exitValue()
+
+ try tailer.close()
+ catch { case _: Throwable => () }
+
+ queue.put(s"[pip] (log at ${logFile.getAbsolutePath})")
+ exitCode
+ }
+
+ def createNewPve(cuid: Int, queue: BlockingQueue[String], pveName: String):
Unit = {
+ queue.put(s"[PVE] Creating new PVE for cuid=$cuid with name=$pveName")
+
+ val venvDirPath = pveDir(cuid, pveName).toAbsolutePath
+ ensureDirExists(cuidDir(cuid, pveName))
+
+ val python = pythonBinPath(cuid, pveName).toAbsolutePath.toString
+ val envVars = pipEnv
+
+ val Requirements: String =
+ """wheel==0.41.2
+ |setuptools==80.10.2
+ |numpy==2.1.0
+ |pandas==2.2.3
+ |ruff==0.14.7
+ |iniconfig==1.1.1
+ |loguru==0.7.0
+ |pyarrow==21.0.0
+ |pytest==7.4.0
+ |python-dateutil==2.8.2
+ |pytest-timeout==2.2.0
+ |protobuf==4.25.8
+ |betterproto==2.0.0b7
+ |typing==3.7.4.3
+ |pampy==0.3.0
+ |overrides==7.4.0
+ |typing_extensions==4.10.0
+ |pytest-reraise==2.1.2
+ |dataclasses==0.6
+ |Deprecated==1.2.14
+ |fs==2.4.16
+ |python-lsp-server[all]==1.12.0
+ |python-lsp-server[websockets]==1.12.0
+ |bidict==0.22.0
+ |cached_property==1.5.2
+ |psutil==5.9.0
+ |tzlocal==2.1
+ |pyiceberg==0.8.1
+ |readerwriterlock==1.0.9
+ |tenacity==8.5.0
+ |SQLAlchemy==2.0.37
+ |pg8000==1.31.5
+ |pympler==1.1
+ |""".stripMargin
+
+ val OperatorRequirements: String =
+ """|wordcloud==1.9.3
+ |plotly==5.24.1
+ |praw==7.6.1
+ |pillow==10.2.0
+ |pybase64==1.3.2
+ |torch==2.8.0
+ |scikit-learn==1.5.0
+ |transformers==4.57.3
+ |boto3==1.40.53
+ |""".stripMargin
+
+ val pveBase = sys.env.getOrElse("PVE_BASE", "/opt/pve-base")
+ val basePython = Paths.get(pveBase).resolve("bin").resolve("python")
+
+ val hasBasePve = Files.exists(basePython)
+
+ if (!hasBasePve) {
+ if (Files.exists(venvDirPath)) {
+ val rmCode = Process(Seq("bash", "-lc", s"rm -rf
'${venvDirPath.toString}'")).!(
+ ProcessLogger(
+ out => queue.put(s"[pve] $out"),
+ err => queue.put(s"[pve][ERR] $err")
+ )
+ )
+ queue.put(s"[pve] removed existing venv with exit code $rmCode")
+ }
+
+ ensureDirExists(venvDirPath.getParent)
+ queue.put(s"[PVE] Creating fresh local venv at ${venvDirPath.toString}")
+
+ val createCode = Process(Seq("python3", "-m", "venv",
venvDirPath.toString)).!(
+ ProcessLogger(
+ out => queue.put(s"[pve] $out"),
+ err => queue.put(s"[pve][ERR] $err")
+ )
+ )
+ queue.put(s"[pve] local venv creation finished with exit code
$createCode")
+
+ if (createCode != 0) {
+ queue.put(s"[PVE][ERR] Failed to create local venv (exit=$createCode)")
+ return
+ }
+
+ ensureDirExists(metadataDir(cuid, pveName))
+ val reqFile1 = metadataDir(cuid, pveName).resolve("requirements.txt")
+ val reqFile2 = metadataDir(cuid,
pveName).resolve("operator-requirements.txt")
+ Files.write(reqFile1, Requirements.getBytes(StandardCharsets.UTF_8))
+ Files.write(reqFile2,
OperatorRequirements.getBytes(StandardCharsets.UTF_8))
+
+ queue.put("[PVE] Installing local base requirements")
+
+ val installReqCode = Process(
+ Seq(python, "-m", "pip", "install", "-r", reqFile1.toString),
+ None,
+ envVars.toSeq: _*
+ ).!(ProcessLogger(_ => (), _ => ()))
+
+ if (installReqCode != 0) {
+ queue.put(s"[PVE][ERR] Failed to install requirements.txt
(exit=$installReqCode)")
+ return
+ }
+
+ queue.put("[PVE] Installing local operator requirements")
+
+ val installOperatorReqCode = Process(
+ Seq(python, "-m", "pip", "install", "-r", reqFile2.toString),
+ None,
+ envVars.toSeq: _*
+ ).!(ProcessLogger(_ => (), _ => ()))
+
+ if (installOperatorReqCode != 0) {
+ queue.put(
+ s"[PVE][ERR] Failed to install operator-requirements.txt
(exit=$installOperatorReqCode)"
+ )
+ return
+ }
+
+ val freezeOutput = Process(Seq(python, "-m", "pip", "freeze"), None,
envVars.toSeq: _*).!!
+ val installedLines =
freezeOutput.split("\n").map(_.trim).filter(_.nonEmpty).toSeq
+
+ writeMetadata(systemPackagesPath(cuid, pveName), installedLines)
+ writeMetadata(userPackagesPath(cuid, pveName), Seq.empty)
+
+ queue.put(s"[PVE] Created new local environment for cuid=$cuid")
+ return
+ }
+
+ if (Files.exists(venvDirPath)) {
+ val rmCode = Process(Seq("bash", "-lc", s"rm -rf
'${venvDirPath.toString}'")).!(
+ ProcessLogger(
+ out => queue.put(s"[pve] $out"),
+ err => queue.put(s"[pve][ERR] $err")
+ )
+ )
+ queue.put(s"[pve] removed existing venv with exit code $rmCode")
+ }
+
+ ensureDirExists(venvDirPath.getParent)
+ queue.put(s"[PVE] Copying base venv from $pveBase to
${venvDirPath.toString}")
+
+ val copyCode = Process(Seq("bash", "-lc", s"cp -a '${pveBase}'
'${venvDirPath.toString}'")).!(
+ ProcessLogger(
+ out => queue.put(s"[pve] $out"),
+ err => queue.put(s"[pve][ERR] $err")
+ )
+ )
+ queue.put(s"[pve] base copy finished with exit code $copyCode")
+
+ if (copyCode != 0) {
+ queue.put(s"[PVE][ERR] Failed to copy base venv (exit=$copyCode)")
+ return
+ }
+
+ val fixCode = Process(
+ Seq(
+ "bash",
+ "-lc",
+ s"""
+ |set -e
+ |PY='${python}'
+ |BIN='${venvDirPath.toString}/bin'
+ |for f in "$$BIN"/*; do
+ | [ -f "$$f" ] || continue
+ | head -n 1 "$$f" | grep -q '^#!' || continue
+ | head -n 1 "$$f" | grep -qi 'python' || continue
+ | sed -i.bak "1s|^#!.*python.*|#!$$PY|" "$$f" || true
+ | rm -f "$$f.bak" || true
+ |done
+ |""".stripMargin
+ )
Review Comment:
The embedded `bash -lc` script interpolates `python`/`venvDirPath` into a
shell heredoc. This is also subject to shell injection if `pveName` is not
strictly validated, and itβs hard to audit. Prefer a non-shell implementation
for rewriting shebangs (or at minimum strictly validate/escape inputs).
##########
amber/src/main/scala/org/apache/texera/web/resource/pythonvirtualenvironment/PveManager.scala:
##########
@@ -0,0 +1,544 @@
+package org.apache.texera.web.resource.pythonvirtualenvironment
+/*
+ * 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 java.io.{File, RandomAccessFile}
+import java.nio.charset.StandardCharsets
+import java.nio.file.{Files, Path, Paths, StandardOpenOption}
+import java.util.concurrent.BlockingQueue
+import scala.collection.mutable.Map
+import scala.jdk.CollectionConverters._
+import scala.sys.process._
+
+object PveManager {
+
+ private val VenvRoot: Path = Paths.get("/tmp/texera-pve/venvs")
+
+ private def ensureDirExists(path: Path): Unit = {
+ if (!Files.exists(path)) Files.createDirectories(path)
+ }
+
+ private def cuidDir(cuid: Int, pvename: String): Path = {
+ ensureDirExists(VenvRoot)
+ val cuIdDir = VenvRoot.resolve(cuid.toString)
+ ensureDirExists(cuIdDir)
+
+ val dir = cuIdDir.resolve(pvename)
+ ensureDirExists(dir)
+
+ dir
+ }
+
+ private def pveDir(cuid: Int, pveName: String): Path =
+ cuidDir(cuid, pveName).resolve("pve")
+
+ private def pythonBinPath(cuid: Int, pveName: String): Path =
+ pveDir(cuid, pveName).resolve("bin").resolve("python")
+
+ private def pipBinPath(cuid: Int, pveName: String): Path =
+ pveDir(cuid, pveName).resolve("bin").resolve("pip")
+
+ private def metadataDir(cuid: Int, pveName: String): Path =
+ pveDir(cuid, pveName).resolve("metadata")
+
+ private def systemPackagesPath(cuid: Int, pveName: String): Path =
+ metadataDir(cuid, pveName).resolve("system-packages.txt")
+
+ private def userPackagesPath(cuid: Int, pveName: String): Path =
+ metadataDir(cuid, pveName).resolve("user-packages.txt")
+
+ private def ensureParentDir(path: Path): Unit = {
+ val parent = path.getParent
+ if (parent != null && !Files.exists(parent))
Files.createDirectories(parent)
+ }
+
+ private def writeMetadata(path: Path, lines: Seq[String]): Unit = {
+ ensureParentDir(path)
+ Files.write(
+ path,
+ lines.asJava,
+ StandardOpenOption.CREATE,
+ StandardOpenOption.TRUNCATE_EXISTING,
+ StandardOpenOption.WRITE
+ )
+ }
+
+ private def readMetadataList(path: Path): List[String] = {
+ if (!Files.exists(path)) return Nil
+ Files.readAllLines(path).asScala.map(_.trim).filter(_.nonEmpty).toList
+ }
+
+ private def parsePackageName(line: String): String =
+ line.split("==", 2).headOption.getOrElse(line).trim.toLowerCase
+
+ private def pipEnv: Map[String, String] =
+ Map(
+ "PYTHONUNBUFFERED" -> "1",
+ "PIP_PROGRESS_BAR" -> "off",
+ "PIP_DISABLE_PIP_VERSION_CHECK" -> "1",
+ "PIP_NO_INPUT" -> "1"
+ )
+
+ def pythonBin(cuid: Int, pveName: String): String =
+ pythonBinPath(cuid, pveName).toAbsolutePath.toString
+
+ def getSystemAndUserPackages(cuid: Int, pveName: String): (Seq[String],
Seq[String]) = {
+ val sys = readMetadataList(systemPackagesPath(cuid, pveName))
+ val usr = readMetadataList(userPackagesPath(cuid, pveName))
+ (sys, usr)
+ }
+
+ def deletePackages(cuid: Int, packageName: String, pveName: String):
List[String] = {
+ val pipPath = pipBinPath(cuid, pveName).toAbsolutePath
+ val userFile = userPackagesPath(cuid, pveName)
+ val systemFile = systemPackagesPath(cuid, pveName)
+
+ val systemPackages: Set[String] =
+ readMetadataList(systemFile).map(parsePackageName).toSet
+
+ val normalizedName = packageName.toLowerCase
+ if (systemPackages.contains(normalizedName)) {
+ return List(s"ERROR: '$packageName' is a system package and cannot be
deleted.")
+ }
+
+ if (!Files.exists(pipPath)) {
+ val msg = s"[PveManager] No pip found at $pipPath β PVE may not exist or
is not initialized."
+ println(msg)
+ return List(msg)
+ }
+
+ try {
+ val command = Seq(pipPath.toString, "uninstall", "-y", packageName)
+
+ val logger = ProcessLogger(
+ (out: String) => println(s"[pip] $out"),
+ (err: String) => System.err.println(s"[pip][ERR] $err")
+ )
+
+ val exitCode = command.!(logger)
+
+ if (exitCode == 0) {
+ val existing = readMetadataList(userFile)
+ val updated =
+ existing.filterNot(line => parsePackageName(line) ==
normalizedName).sorted
+
+ writeMetadata(userFile, updated)
+
+ List(s"Exit code: $exitCode", s"Uninstalled $packageName successfully")
+ } else {
+ List(s"[PveManager] pip uninstall for '$packageName' failed with exit
code $exitCode")
+ }
+ } catch {
+ case e: Exception =>
+ List(s"[PveManager] Failed to delete package for cuid=$cuid:
${e.getMessage}")
+ }
+ }
+
+ private def tailFileToQueue(
+ file: File,
+ queue: BlockingQueue[String],
+ prefix: String = "[pip] "
+ ): AutoCloseable = {
+ val raf = new RandomAccessFile(file, "r")
+ raf.seek(raf.length())
+ @volatile var running = true
+
+ val thread = new Thread(() => {
+ var buf = new StringBuilder
+ val charset = StandardCharsets.UTF_8
+ try {
+ while (running) {
+ val available = raf.length() - raf.getFilePointer
+ if (available > 0) {
+ val bytes = new Array[Byte](math.min(available, 8192).toInt)
+ val n = raf.read(bytes)
+ if (n > 0) {
+ buf.append(new String(bytes, 0, n, charset))
+ var newlineIndex = buf.indexOf("\n")
+ while (newlineIndex >= 0) {
+ val line = buf.substring(0, newlineIndex).trim
+ if (shouldStream(line)) queue.put(prefix + line)
+ buf = buf.delete(0, newlineIndex + 1)
+ newlineIndex = buf.indexOf("\n")
+ }
+ }
+ } else {
+ Thread.sleep(100)
+ }
+ }
+ val last = buf.result().trim
+ if (shouldStream(last)) queue.put(prefix + last)
+ } catch {
+ case _: InterruptedException => ()
+ case e: Exception => queue.put(s"[pip][ERR] tail exception:
${e.getMessage}")
+ } finally {
+ raf.close()
+ }
+ })
+
+ thread.setDaemon(true)
+ thread.start()
+
+ new AutoCloseable {
+ override def close(): Unit = {
+ running = false
+ thread.interrupt()
+ thread.join(500)
+ }
+ }
+ }
+
+ private def shouldStream(line: String): Boolean = {
+ val s = line.trim
+ if (s.isEmpty) return false
+
+ val lower = s.toLowerCase
+
+ if (lower.contains("found link")) return false
+ if (lower.contains("skipping link")) return false
+ if (lower.contains("cache")) return false
+ if (lower.contains("caching")) return false
+
+ true
+ }
+
+ private def runPipWithLog(
+ cmd: Seq[String],
+ env: Map[String, String],
+ queue: BlockingQueue[String]
+ ): Int = {
+ val logFile = File.createTempFile("pip-live-", ".log")
+ val fullCmd = if (cmd.contains("--log")) cmd else cmd ++ Seq("--log",
logFile.getAbsolutePath)
+
+ val tailer = tailFileToQueue(logFile, queue)
+
+ val logger = ProcessLogger(
+ out => if (shouldStream(out)) queue.put(s"[pip/stdout] $out"),
+ err => if (shouldStream(err)) queue.put(s"[pip/stderr] $err")
+ )
+
+ val proc = Process(fullCmd, None, env.toSeq: _*).run(logger)
+ val exitCode = proc.exitValue()
+
+ try tailer.close()
+ catch { case _: Throwable => () }
+
+ queue.put(s"[pip] (log at ${logFile.getAbsolutePath})")
+ exitCode
+ }
+
+ def createNewPve(cuid: Int, queue: BlockingQueue[String], pveName: String):
Unit = {
+ queue.put(s"[PVE] Creating new PVE for cuid=$cuid with name=$pveName")
+
+ val venvDirPath = pveDir(cuid, pveName).toAbsolutePath
+ ensureDirExists(cuidDir(cuid, pveName))
+
+ val python = pythonBinPath(cuid, pveName).toAbsolutePath.toString
+ val envVars = pipEnv
+
+ val Requirements: String =
+ """wheel==0.41.2
+ |setuptools==80.10.2
+ |numpy==2.1.0
+ |pandas==2.2.3
+ |ruff==0.14.7
+ |iniconfig==1.1.1
+ |loguru==0.7.0
+ |pyarrow==21.0.0
+ |pytest==7.4.0
+ |python-dateutil==2.8.2
+ |pytest-timeout==2.2.0
+ |protobuf==4.25.8
+ |betterproto==2.0.0b7
+ |typing==3.7.4.3
+ |pampy==0.3.0
+ |overrides==7.4.0
+ |typing_extensions==4.10.0
+ |pytest-reraise==2.1.2
+ |dataclasses==0.6
+ |Deprecated==1.2.14
+ |fs==2.4.16
+ |python-lsp-server[all]==1.12.0
+ |python-lsp-server[websockets]==1.12.0
+ |bidict==0.22.0
+ |cached_property==1.5.2
+ |psutil==5.9.0
+ |tzlocal==2.1
+ |pyiceberg==0.8.1
+ |readerwriterlock==1.0.9
+ |tenacity==8.5.0
+ |SQLAlchemy==2.0.37
+ |pg8000==1.31.5
+ |pympler==1.1
+ |""".stripMargin
+
+ val OperatorRequirements: String =
+ """|wordcloud==1.9.3
+ |plotly==5.24.1
+ |praw==7.6.1
+ |pillow==10.2.0
+ |pybase64==1.3.2
+ |torch==2.8.0
+ |scikit-learn==1.5.0
+ |transformers==4.57.3
+ |boto3==1.40.53
+ |""".stripMargin
+
+ val pveBase = sys.env.getOrElse("PVE_BASE", "/opt/pve-base")
+ val basePython = Paths.get(pveBase).resolve("bin").resolve("python")
+
+ val hasBasePve = Files.exists(basePython)
+
+ if (!hasBasePve) {
+ if (Files.exists(venvDirPath)) {
+ val rmCode = Process(Seq("bash", "-lc", s"rm -rf
'${venvDirPath.toString}'")).!(
+ ProcessLogger(
+ out => queue.put(s"[pve] $out"),
+ err => queue.put(s"[pve][ERR] $err")
+ )
+ )
+ queue.put(s"[pve] removed existing venv with exit code $rmCode")
+ }
+
+ ensureDirExists(venvDirPath.getParent)
+ queue.put(s"[PVE] Creating fresh local venv at ${venvDirPath.toString}")
+
+ val createCode = Process(Seq("python3", "-m", "venv",
venvDirPath.toString)).!(
+ ProcessLogger(
+ out => queue.put(s"[pve] $out"),
+ err => queue.put(s"[pve][ERR] $err")
+ )
+ )
+ queue.put(s"[pve] local venv creation finished with exit code
$createCode")
+
+ if (createCode != 0) {
+ queue.put(s"[PVE][ERR] Failed to create local venv (exit=$createCode)")
+ return
+ }
+
+ ensureDirExists(metadataDir(cuid, pveName))
+ val reqFile1 = metadataDir(cuid, pveName).resolve("requirements.txt")
+ val reqFile2 = metadataDir(cuid,
pveName).resolve("operator-requirements.txt")
+ Files.write(reqFile1, Requirements.getBytes(StandardCharsets.UTF_8))
+ Files.write(reqFile2,
OperatorRequirements.getBytes(StandardCharsets.UTF_8))
+
+ queue.put("[PVE] Installing local base requirements")
+
+ val installReqCode = Process(
+ Seq(python, "-m", "pip", "install", "-r", reqFile1.toString),
+ None,
+ envVars.toSeq: _*
+ ).!(ProcessLogger(_ => (), _ => ()))
+
+ if (installReqCode != 0) {
+ queue.put(s"[PVE][ERR] Failed to install requirements.txt
(exit=$installReqCode)")
+ return
+ }
+
+ queue.put("[PVE] Installing local operator requirements")
+
+ val installOperatorReqCode = Process(
+ Seq(python, "-m", "pip", "install", "-r", reqFile2.toString),
+ None,
+ envVars.toSeq: _*
+ ).!(ProcessLogger(_ => (), _ => ()))
+
+ if (installOperatorReqCode != 0) {
+ queue.put(
+ s"[PVE][ERR] Failed to install operator-requirements.txt
(exit=$installOperatorReqCode)"
+ )
+ return
+ }
+
+ val freezeOutput = Process(Seq(python, "-m", "pip", "freeze"), None,
envVars.toSeq: _*).!!
+ val installedLines =
freezeOutput.split("\n").map(_.trim).filter(_.nonEmpty).toSeq
+
+ writeMetadata(systemPackagesPath(cuid, pveName), installedLines)
+ writeMetadata(userPackagesPath(cuid, pveName), Seq.empty)
+
+ queue.put(s"[PVE] Created new local environment for cuid=$cuid")
+ return
+ }
+
+ if (Files.exists(venvDirPath)) {
+ val rmCode = Process(Seq("bash", "-lc", s"rm -rf
'${venvDirPath.toString}'")).!(
+ ProcessLogger(
+ out => queue.put(s"[pve] $out"),
+ err => queue.put(s"[pve][ERR] $err")
+ )
+ )
+ queue.put(s"[pve] removed existing venv with exit code $rmCode")
+ }
+
+ ensureDirExists(venvDirPath.getParent)
+ queue.put(s"[PVE] Copying base venv from $pveBase to
${venvDirPath.toString}")
+
+ val copyCode = Process(Seq("bash", "-lc", s"cp -a '${pveBase}'
'${venvDirPath.toString}'")).!(
+ ProcessLogger(
+ out => queue.put(s"[pve] $out"),
+ err => queue.put(s"[pve][ERR] $err")
+ )
+ )
Review Comment:
This uses `bash -lc` with interpolated `pveBase` / `venvDirPath` (`cp -a
'...')`, which is vulnerable to shell injection if `pveName` contains
quotes/metacharacters. Prefer filesystem APIs for copying (or `Process` with
arg lists and no shell) and validate `pveName`.
##########
frontend/src/app/workspace/component/power-button/computing-unit-selection.component.ts:
##########
@@ -637,4 +670,281 @@ export class ComputingUnitSelectionComponent implements
OnInit {
this.computingUnitStatusService.refreshComputingUnitList();
}
}
+
+ private makeEmptyPve(expanded: boolean): PveDraft {
+ return {
+ id: this.nextPveId++,
+ name: "",
+ userPackages: [],
+ newPackages: [{ name: "", operator: "==", version: "" }],
+ deletingPackages: [],
+ pipOutput: "",
+ prettyPipOutput: "",
+ expanded,
+ isInstalling: false,
+ };
+ }
+
+ addEnvironment(): void {
+ this.pves.push(this.makeEmptyPve(true));
+ }
+
+ trackByPveId(_index: number, pve: PveDraft): number {
+ return pve.id;
+ }
+
+ showPVEmodalVisible(): void {
+ this.pveModalVisible = true;
+ this.getPVEs();
+ }
+
+ getPVEs(): void {
+ const cuId: number | undefined =
this.selectedComputingUnit?.computingUnit.cuid;
+
+ if (cuId == null) {
+ this.notificationService.error("No computing unit selected. Please
select a CU first.");
+ return;
+ }
+
+ this.workflowPveService.setCuid(cuId);
+
+ this.workflowPveService
+ .fetchPVEs(cuId)
+ .pipe(untilDestroyed(this))
+ .subscribe({
+ next: (resp: PvePackageResponse[]) => {
+ this.pves = resp.map((pve, index) => ({
+ id: index,
+ name: pve.pveName,
+ expanded: false,
+ isInstalling: false,
+ pipOutput: "",
+ prettyPipOutput: "",
+ userPackages: (pve.userPackages ?? []).map(pkg => {
+ const [name, version] = pkg.split("==");
+ return {
+ name: name.trim(),
+ version: (version ?? "").trim(),
+ isHighlighted: false,
+ };
+ }),
+ newPackages: [],
+ deletingPackages: [],
+ }));
+ },
+ error: (err: unknown) => {
+ console.error("Failed to fetch PVEs:", err);
+ this.pves = [];
+ },
+ });
+ }
+
+ addPackage(index: number): void {
+ const env = this.pves[index];
+ env.newPackages.push({ name: "", version: "", operator: undefined,
isHighlighted: false });
+ }
+
+ togglePackageDelete(index: number, pkg: PackageRow): void {
+ const env = this.pves[index];
+
+ pkg.isHighlighted = !pkg.isHighlighted;
+
+ const version = pkg.version ?? "";
+
+ if (pkg.isHighlighted) {
+ const exists = env.deletingPackages.some(p => p.name === pkg.name &&
(p.version ?? "") === version);
+ if (!exists) {
+ env.deletingPackages.push({ name: pkg.name, version });
+ }
+ } else {
+ env.deletingPackages = env.deletingPackages.filter(p => !(p.name ===
pkg.name && (p.version ?? "") === version));
+ }
+ }
+
+ scrollToBottomOfPipModal(index: number) {
+ setTimeout(() => {
+ const pre = document.getElementById(`pip-log-${index}`) as HTMLElement |
null;
+ if (pre) {
+ pre.scrollTop = pre.scrollHeight;
+ }
+ }, 50);
+ }
+
+ updatePrettyPipOutput(index: number) {
+ const env = this.pves[index];
+
+ const escapeHtml = (s: string) =>
+ s
+ .replace(/&/g, "&")
+ .replace(/</g, "<")
+ .replace(/>/g, ">")
+ .replace(/"/g, """)
+ .replace(/'/g, "'");
+
+ const raw = env.pipOutput ?? "";
+ const safe = escapeHtml(raw);
+
+ env.prettyPipOutput = safe
+ .replace(
+ /^(\[pip\].*finished with exit code\s+0.*)$/gm,
+ "<span class=\"pip-exit ok\"><strong>$1</strong></span>"
+ )
+ .replace(
+ /^(\[pip\].*finished with exit code\s+1.*)$/gm,
+ "<span class=\"pip-exit err\"><strong>$1</strong></span>"
+ )
+ .replace(
+ /^(\[pip\].*finished with exit code\s+([2-9]\d*).*)$/gm,
+ "<span class=\"pip-exit err\"><strong>$1</strong></span>"
+ )
+ .replace(/ERROR/g, "<span class=\"error\">ERROR</span>")
+ .replace(/WARNING/g, "<span class=\"warning\">WARNING</span>")
+ .replace(/already satisfied/g, "<span class=\"success\">already
satisfied</span>")
+ .replace(/\n/g, "<br/>");
+ }
+
+ createVirtualEnvironment(index: number): void {
+ const cuId = this.selectedComputingUnit?.computingUnit.cuid;
+ const env = this.pves[index];
+
+ if (cuId == null) {
+ this.notificationService.error("No computing unit selected. Please
select a CU first.");
+ return;
+ }
+
+ if (!env.name?.trim()) {
+ this.notificationService.error("Environment name cannot be empty.");
+ return;
+ }
+
+ env.isInstalling = true;
+
+ this.workflowPveService.setCuid(cuId);
+ this.workflowPveService.setPveName(env.name);
+
+ for (const pkg of env.deletingPackages) {
+ this.workflowPveService
+ .deletePackage(pkg.name)
+ .pipe(untilDestroyed(this))
+ .subscribe({
+ next: () => {
+ env.pipOutput += `Starting ...\nSuccessfully uninstalled package
${pkg.name}`;
+ this.updatePrettyPipOutput(index);
+ this.scrollToBottomOfPipModal(index);
+
+ this.workflowPveService
+ .getInstalledPackages()
+ .pipe(untilDestroyed(this))
+ .subscribe({
+ next: resp => {
+ this.systemPackages = resp.system.map(pkgStr => {
+ const [name, version] = pkgStr.split("==");
+ return { name: name.trim(), version: (version ?? "").trim() };
+ });
+
+ env.userPackages = resp.user.map(pkgStr => {
+ const [name, version] = pkgStr.split("==");
+ return { name: name.trim(), version: (version ?? "").trim(),
isHighlighted: false };
+ });
+
+ env.deletingPackages = [];
+ },
+ error: (e: unknown) => console.error("Failed to refresh packages
after delete", e),
+ });
+ },
+ error: (err: unknown) => {
+ console.error("Error deleting package:", err);
+ env.pipOutput += `\nError uninstalling ${pkg.name}`;
+ this.updatePrettyPipOutput(index);
+ this.scrollToBottomOfPipModal(index);
+ },
+ });
+ }
+
+ const packageArray: string[] = [];
+
+ for (const p of env.newPackages) {
+ const name = (p.name ?? "").trim();
+ const version = (p.version ?? "").trim();
+ const op = (p.operator ?? "==").trim() || "==";
+
+ if (name !== "") {
+ const formatted = version !== "" ? `${name}${op}${version}` : name;
+ packageArray.push(formatted);
+ }
+ }
+
+ if (packageArray.length === 0) {
+ env.isInstalling = false;
+ return;
+ }
Review Comment:
If the user only selected packages to uninstall, `packageArray.length === 0`
causes an early return and sets `isInstalling=false` even though uninstall
requests may still be in-flight. Handle the uninstall-only case explicitly
(wait for uninstall completion, refresh packages, then clear `isInstalling`).
##########
frontend/src/app/workspace/component/power-button/computing-unit-selection.component.scss:
##########
@@ -388,3 +388,109 @@
outline: none;
}
}
+
+.package-row {
+ display: flex;
+ align-items: flex-end;
+ justify-content: space-between;
+ gap: 8px;
+ margin-bottom: 10px;
+}
+
+.package-inputs {
+ display: flex;
+ flex: 1;
+ gap: 10px; /* space between name and version */
+}
+
+.field {
+ display: flex;
+ flex-direction: column;
+ flex: 1; /* ensures both inputs are equal width */
+}
+
+.add-btn {
+ display: flex;
+ justify-content: flex-start;
+ margin-top: 10px;
+}
+
+.footer-all{
+ display: flex;
+ justify-content: space-between;
+ align-items: center;
+ width: 100%;
+}
+
+.footer-right{
+ display: flex;
+ gap: 8px;
+}
+
+.system-section {
+ margin-top: 20px;
+ margin-bottom: 20px;
+}
+
+.system-panel-inner {
+ padding: 16px;
+}
+
+.system-row {
+ margin-bottom: 14px;
+}
+
+.system-input {
+ background: #f5f5f5 !important;
+ border-radius: 4px !important;
+}
+
+.package-row {
+ display: flex;
+ align-items: flex-end;
+ margin-bottom: 16px;
+}
+
+.package-inputs {
+ flex: 1;
+ display: flex;
+ gap: 14px;
+}
+
+.field {
+ flex: 1;
+ display: flex;
+ flex-direction: column;
+}
+
+.package-inputs .field.operator {
+ width: 50px;
+ text-align: center;
+}
+
+.operator-select ::ng-deep .ant-select-selector {
+ display: flex;
+ align-items: center;
+}
+
+.operator-select ::ng-deep .ant-select-selection-placeholder,
+.operator-select ::ng-deep .ant-select-selection-item {
+ width: 100%;
+ text-align: center;
+}
+
+.operator-select {
+ width: 120px;
+}
+
+.operator-select ::ng-deep .ant-select {
+ width: 100%;
Review Comment:
This SCSS section redefines `.package-row`, `.package-inputs`, and `.field`
that were already defined a few lines above with different gap/margin values.
The later definitions override earlier ones, making layout changes brittle.
Consolidate into a single definition (and ideally scope under the modal
wrapper) to avoid accidental overrides.
```suggestion
.package-row {
display: flex;
align-items: flex-end;
margin-bottom: 16px;
}
.package-inputs {
flex: 1;
display: flex;
gap: 14px;
.field {
flex: 1;
display: flex;
flex-direction: column;
&.operator {
width: 50px;
text-align: center;
}
}
}
.operator-select {
width: 120px;
::ng-deep .ant-select-selector {
display: flex;
align-items: center;
}
::ng-deep .ant-select-selection-placeholder,
::ng-deep .ant-select-selection-item {
width: 100%;
text-align: center;
}
::ng-deep .ant-select {
width: 100%;
}
}
}
.system-panel-inner {
padding: 16px;
}
.system-row {
margin-bottom: 14px;
}
.system-input {
background: #f5f5f5 !important;
border-radius: 4px !important;
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
