You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@texera.apache.org by "SarahAsad23 (via GitHub)" <gi...@apache.org> on 2026/04/10 19:38:08 UTC
[PR] feat: Add Python Virtual Environment Support [texera]
SarahAsad23 opened a new pull request, #4359:
URL: https://github.com/apache/texera/pull/4359
<!--
Thanks for sending a pull request (PR)! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines:
[Contributing to Texera](https://github.com/apache/texera/blob/main/CONTRIBUTING.md)
2. Ensure you have added or run the appropriate tests for your PR
3. If the PR is work in progress, mark it a draft on GitHub.
4. Please write your PR title to summarize what this PR proposes, we
are following Conventional Commits style for PR titles as well.
5. Be sure to keep the PR description updated to reflect all changes.
-->
### What changes were proposed in this PR?
<!--
Please clarify what changes you are proposing. The purpose of this section
is to outline the changes. Here are some tips for you:
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
3. If it is a refactoring, clarify what has been changed.
3. It would be helpful to include a before-and-after comparison using
screenshots or GIFs.
4. Please consider writing useful notes for better and faster reviews.
-->
### Any related issues, documentation, discussions?
<!--
Please use this section to link other resources if not mentioned already.
1. If this PR fixes an issue, please include `Fixes #1234`, `Resolves #1234`
or `Closes #1234`. If it is only related, simply mention the issue number.
2. If there is design documentation, please add the link.
3. If there is a discussion in the mailing list, please add the link.
-->
### How was this PR tested?
<!--
If tests were added, say they were added here. Or simply mention that if the PR
is tested with existing test cases. Make sure to include/update test cases that
check the changes thoroughly including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please clarify how
you tested step by step, ideally copy and paste-able, so that other reviewers can
test and check, and descendants can verify in the future. If tests were not added,
please describe why they were not added and/or why it was difficult to add.
-->
### Was this PR authored or co-authored using generative AI tooling?
<!--
If generative AI tooling has been used in the process of authoring this PR,
please include the phrase: 'Generated-by: ' followed by the name of the tool
and its version. If no, write 'No'.
Please refer to the [ASF Generative Tooling Guidance](https://www.apache.org/legal/generative-tooling.html) for details.
-->
--
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: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add Python Virtual Environment Support [texera]
Posted by "SarahAsad23 (via GitHub)" <gi...@apache.org>.
SarahAsad23 commented on code in PR #4359:
URL: https://github.com/apache/texera/pull/4359#discussion_r3114316643
##########
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)
Review Comment:
The directories introduced in this PR are PVE-scoped, not CU-scoped. They are created lazily only when a user creates a PVE, so they do not exist at CU creation time. Their lifecycle is therefore tied to PVE creation and deletion rather than general CU initialization.
--
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: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add Python Virtual Environment Support [texera]
Posted by "chenlica (via GitHub)" <gi...@apache.org>.
chenlica commented on PR #4359:
URL: https://github.com/apache/texera/pull/4359#issuecomment-4275160921
@SarahAsad23 There are still conflicts. Please resolve them.
--
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: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add Python Virtual Environment Support [texera]
Posted by "SarahAsad23 (via GitHub)" <gi...@apache.org>.
SarahAsad23 commented on code in PR #4359:
URL: https://github.com/apache/texera/pull/4359#discussion_r3114341320
##########
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()) {
Review Comment:
This check ensures the name is not empty or whitespace only.
--
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: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add Python Virtual Environment Support [texera]
Posted by "SarahAsad23 (via GitHub)" <gi...@apache.org>.
SarahAsad23 commented on code in PR #4359:
URL: https://github.com/apache/texera/pull/4359#discussion_r3114352251
##########
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 {
Review Comment:
This is used by the frontend to track each PVE by its ID when rendering the list, so existing items are preserved instead of being re-rendered unnecessarily.
--
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: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add Python Virtual Environment Support [texera]
Posted by "SarahAsad23 (via GitHub)" <gi...@apache.org>.
SarahAsad23 closed pull request #4359: feat: Add Python Virtual Environment Support
URL: https://github.com/apache/texera/pull/4359
--
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: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add Python Virtual Environment Support [texera]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
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: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add Python Virtual Environment Support [texera]
Posted by "SarahAsad23 (via GitHub)" <gi...@apache.org>.
SarahAsad23 commented on code in PR #4359:
URL: https://github.com/apache/texera/pull/4359#discussion_r3114357173
##########
frontend/src/app/workspace/component/power-button/computing-unit-selection.component.ts:
##########
@@ -56,6 +56,27 @@ import {
isComputingUnitShmTooLarge,
getJvmMemorySliderConfig,
} from "../../../common/util/computing-unit.util";
+import { PvePackageResponse, WorkflowPveService } from "../../service/virtual-environment/virtual-environment.service";
+
+type PackageRow = {
+ name: string;
+ operator?: "==" | ">=" | "<=";
+ version?: string;
+ isHighlighted?: boolean;
+};
+
+type PveDraft = {
+ id: number;
Review Comment:
It’s used for frontend rendering so Angular can track each PVE and avoid re-rendering the entire list when it updates.
--
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: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add Python Virtual Environment Support [texera]
Posted by "SarahAsad23 (via GitHub)" <gi...@apache.org>.
SarahAsad23 commented on code in PR #4359:
URL: https://github.com/apache/texera/pull/4359#discussion_r3114328292
##########
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.");
Review Comment:
This can happen when no CU is selected in the UI, but the CU list is still visible. If the user clicks the ‘+’ button without selecting a CU first, the cuId will be null. That’s the scenario this check is handling.
--
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: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add Python Virtual Environment Support [texera]
Posted by "SarahAsad23 (via GitHub)" <gi...@apache.org>.
SarahAsad23 commented on PR #4359:
URL: https://github.com/apache/texera/pull/4359#issuecomment-4306933448
This PR has been closed following an internal discussion to break it into smaller parts. One of the resulting PRs is #4484 .
--
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: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add Python Virtual Environment Support [texera]
Posted by "chenlica (via GitHub)" <gi...@apache.org>.
chenlica commented on PR #4359:
URL: https://github.com/apache/texera/pull/4359#issuecomment-4265336654
@SarahAsad23 Please resolve the conflicts.
--
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: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add Python Virtual Environment Support [texera]
Posted by "kunwp1 (via GitHub)" <gi...@apache.org>.
kunwp1 commented on code in PR #4359:
URL: https://github.com/apache/texera/pull/4359#discussion_r3113252714
##########
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
Review Comment:
It's not a good idea to hardcode all the python packages because it's hard to sync with the latest requirements.
##########
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)
Review Comment:
I think the VenvRoot and all the subdirectories except PVE related directories should be created when a computing unit is created. The lifecycle of those directories is not clear in this PR.
##########
amber/src/main/scala/org/apache/texera/web/resource/pythonvirtualenvironment/PveManager.scala:
##########
Review Comment:
Add some comments of what this manager is and also add comment for each method if the method is too long.
##########
frontend/src/app/workspace/component/power-button/computing-unit-selection.component.ts:
##########
@@ -56,6 +56,27 @@ import {
isComputingUnitShmTooLarge,
getJvmMemorySliderConfig,
} from "../../../common/util/computing-unit.util";
+import { PvePackageResponse, WorkflowPveService } from "../../service/virtual-environment/virtual-environment.service";
+
+type PackageRow = {
+ name: string;
+ operator?: "==" | ">=" | "<=";
+ version?: string;
+ isHighlighted?: boolean;
Review Comment:
Maybe rename it to `delete` or `deleteToggle`?
##########
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 = {
Review Comment:
Consider removing `ensureParentDir` and `ensureDirExists`. I don't think we need to check whether the directory exists or not.
##########
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)
Review Comment:
You don't need to run createDirectories step-by-step. Can simply run a single createDirectories call.
##########
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.");
Review Comment:
I don't see a way to set cuId == null in the frontend. What's a test scenario to test this error?
##########
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) {
Review Comment:
Same here.
##########
amber/src/main/scala/org/apache/texera/web/resource/pythonvirtualenvironment/PveResource.scala:
##########
Review Comment:
Check if there are any dead codes in this file.
##########
amber/src/main/scala/org/apache/texera/web/resource/pythonvirtualenvironment/PveResource.scala:
##########
@@ -0,0 +1,230 @@
+/*
+ * 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 com.fasterxml.jackson.databind.ObjectMapper
+import org.glassfish.jersey.server.ChunkedOutput
+
+import java.util.concurrent.LinkedBlockingQueue
+import javax.ws.rs._
+import javax.ws.rs.core.MediaType
+import scala.concurrent.ExecutionContext.Implicits.global
+import scala.concurrent.Future
+import scala.jdk.CollectionConverters._
+import java.util
+
+case class PackageResponse(system: java.util.List[String], user: java.util.List[String])
+
+@Path("/pve")
+@Consumes(Array(MediaType.APPLICATION_JSON))
+class PveResource {
+
+ // --------------------------------------------------
+ // Create / Install packages (SSE)
+ // --------------------------------------------------
+ @GET
+ @Produces(Array("text/event-stream"))
+ def createPve(
+ @QueryParam("packages") packagesJson: String,
+ @QueryParam("cuid") cuid: Int,
+ @QueryParam("pveName") pveName: String
+ ): ChunkedOutput[String] = {
+
+ if (cuid == 0) {
+ throw new BadRequestException("Missing or invalid cuid")
+ }
+
+ if (pveName == null || pveName.trim.isEmpty) {
+ throw new BadRequestException("Missing environment name")
+ }
+
+ val mapper = new ObjectMapper()
+ val packages =
+ if (packagesJson == null || packagesJson.trim.isEmpty)
+ List.empty[String]
+ else
+ mapper.readValue(packagesJson, classOf[Array[String]]).toList
+
+ val queue = new LinkedBlockingQueue[String]()
+ val chunkedOutput = new ChunkedOutput[String](classOf[String])
+
+ Future {
+ try {
+
+ if (!PveManager.pveExists(cuid, pveName)) {
+ PveManager.createNewPve(cuid, queue, pveName)
+ }
+
+ if (packages.nonEmpty) {
+ PveManager.installPackages(packages, cuid, queue, pveName)
+ }
+
+ } catch {
+ case e: Exception =>
+ queue.put(s"[ERR] ${e.getMessage}")
+ } finally {
+ queue.put("__DONE__")
+ }
+ }
+
+ Future {
+ var done = false
+ while (!done) {
+ val line = queue.take()
+ if (line == "__DONE__") {
+ chunkedOutput.write("data: __DONE__\n\n")
+ done = true
+ } else chunkedOutput.write(s"data: $line\n\n")
+ }
+ chunkedOutput.close()
+ }
+
+ chunkedOutput
+ }
+
+ // --------------------------------------------------
+ // Get installed packages
+ // --------------------------------------------------
+ @GET
+ @Path("/packages")
+ @Produces(Array(MediaType.APPLICATION_JSON))
+ def getInstalledPackages(
+ @QueryParam("cuid") cuid: Int,
+ @QueryParam("pveName") pveName: String
+ ): util.Map[String, util.List[String]] = {
+
+ if (cuid == 0) {
+ throw new BadRequestException("Missing or invalid cuid")
+ }
+
+ if (pveName == null || pveName.trim.isEmpty) {
Review Comment:
Is there a scenario where pveName can be empty when the backend is executing? This might be a deadcode.
##########
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 = {
Review Comment:
Add comment of this method. It's too long to understand.
##########
amber/src/main/scala/org/apache/texera/web/TexeraWebApplication.scala:
##########
@@ -160,7 +161,7 @@ class TexeraWebApplication
environment.jersey.register(classOf[UserQuotaResource])
environment.jersey.register(classOf[AdminSettingsResource])
environment.jersey.register(classOf[AIAssistantResource])
-
Review Comment:
Does the creation of PVE involve Texera Web Application?
##########
bin/k8s/templates/gateway-routes.yaml:
##########
@@ -100,27 +101,47 @@ spec:
parentRefs:
- name: {{ .Release.Name }}-gateway
{{- if and .Values.gatewayConfig .Values.gatewayConfig.hostname }}
- hostnames:
- - {{ .Values.gatewayConfig.hostname }}
+hostnames:
+ - {{ .Values.gatewayConfig.hostname }}
{{- end }}
- rules:
- - matches:
- - path:
- type: PathPrefix
- value: /wsapi
- - path:
- type: RegularExpression
- value: "^/api/executions/\\d+/stats/\\d+$"
- - path:
- type: PathPrefix
- value: /api/executions/result/export
- backendRefs:
- - group: gateway.envoyproxy.io
- kind: Backend
- name: texera-dynamic-backend
+rules:
+ - matches:
+ - path:
+ type: PathPrefix
+ value: /pve
+ timeouts:
+ request: 10m
Review Comment:
Why do we need timeouts and why is it 10 minutes?
##########
frontend/src/app/common/type/workflow-computing-unit.ts:
##########
@@ -50,6 +50,7 @@ export interface DashboardWorkflowComputingUnit {
metrics: WorkflowComputingUnitMetrics;
isOwner: boolean;
accessPrivilege: "READ" | "WRITE" | "NONE";
+ packages: string[];
Review Comment:
Do we need this change? If so, can you add a comment?
##########
bin/k8s/templates/gateway-routes.yaml:
##########
Review Comment:
Please remove all the unnecessary diffs from this file. I see so many of them.
##########
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 {
Review Comment:
Remove this?
##########
amber/src/main/scala/org/apache/texera/web/resource/pythonvirtualenvironment/PveResource.scala:
##########
@@ -0,0 +1,230 @@
+/*
+ * 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 com.fasterxml.jackson.databind.ObjectMapper
+import org.glassfish.jersey.server.ChunkedOutput
+
+import java.util.concurrent.LinkedBlockingQueue
+import javax.ws.rs._
+import javax.ws.rs.core.MediaType
+import scala.concurrent.ExecutionContext.Implicits.global
+import scala.concurrent.Future
+import scala.jdk.CollectionConverters._
+import java.util
+
+case class PackageResponse(system: java.util.List[String], user: java.util.List[String])
+
+@Path("/pve")
+@Consumes(Array(MediaType.APPLICATION_JSON))
+class PveResource {
+
+ // --------------------------------------------------
+ // Create / Install packages (SSE)
+ // --------------------------------------------------
+ @GET
+ @Produces(Array("text/event-stream"))
+ def createPve(
+ @QueryParam("packages") packagesJson: String,
+ @QueryParam("cuid") cuid: Int,
+ @QueryParam("pveName") pveName: String
+ ): ChunkedOutput[String] = {
+
+ if (cuid == 0) {
+ throw new BadRequestException("Missing or invalid cuid")
+ }
+
+ if (pveName == null || pveName.trim.isEmpty) {
+ throw new BadRequestException("Missing environment name")
+ }
+
+ val mapper = new ObjectMapper()
+ val packages =
+ if (packagesJson == null || packagesJson.trim.isEmpty)
+ List.empty[String]
+ else
+ mapper.readValue(packagesJson, classOf[Array[String]]).toList
+
+ val queue = new LinkedBlockingQueue[String]()
+ val chunkedOutput = new ChunkedOutput[String](classOf[String])
+
+ Future {
+ try {
+
+ if (!PveManager.pveExists(cuid, pveName)) {
+ PveManager.createNewPve(cuid, queue, pveName)
+ }
+
+ if (packages.nonEmpty) {
+ PveManager.installPackages(packages, cuid, queue, pveName)
+ }
+
+ } catch {
+ case e: Exception =>
+ queue.put(s"[ERR] ${e.getMessage}")
+ } finally {
+ queue.put("__DONE__")
+ }
+ }
+
+ Future {
+ var done = false
+ while (!done) {
+ val line = queue.take()
+ if (line == "__DONE__") {
+ chunkedOutput.write("data: __DONE__\n\n")
+ done = true
+ } else chunkedOutput.write(s"data: $line\n\n")
+ }
+ chunkedOutput.close()
+ }
+
+ chunkedOutput
+ }
+
+ // --------------------------------------------------
+ // Get installed packages
+ // --------------------------------------------------
+ @GET
+ @Path("/packages")
+ @Produces(Array(MediaType.APPLICATION_JSON))
+ def getInstalledPackages(
+ @QueryParam("cuid") cuid: Int,
+ @QueryParam("pveName") pveName: String
+ ): util.Map[String, util.List[String]] = {
+
+ if (cuid == 0) {
+ throw new BadRequestException("Missing or invalid cuid")
+ }
+
+ if (pveName == null || pveName.trim.isEmpty) {
+ throw new BadRequestException("Missing environment name")
+ }
+
+ try {
+
+ println(s"[PVE] HIT getInstalledPackages cuid=$cuid")
+
+ val (systemPkgsRaw, userPkgsRaw) = PveManager.getSystemAndUserPackages(cuid, pveName)
+
+ println(s"[PVE] raw systemPkgsRaw isNull=${systemPkgsRaw == null} value=$systemPkgsRaw")
+ println(s"[PVE] raw userPkgsRaw isNull=${userPkgsRaw == null} value=$userPkgsRaw")
+
+ val systemPkgs = Option(systemPkgsRaw).getOrElse(Seq.empty[String]).toList.asJava
+ val userPkgs = Option(userPkgsRaw).getOrElse(Seq.empty[String]).toList.asJava
+
+ val resp = Map("system" -> systemPkgs, "user" -> userPkgs).asJava
+
+ println(
+ s"[PVE] returning keys=${resp.keySet()} systemSize=${systemPkgs.size()} userSize=${userPkgs.size()}"
+ )
+
+ resp
+
+ } catch {
+ case e: Exception =>
+ e.printStackTrace()
+ Map(
+ "system" -> List(s"Error: ${e.getMessage}").asJava,
+ "user" -> List.empty[String].asJava
+ ).asJava
+ }
+ }
+
+ // --------------------------------------------------
+ // Fetch PVEs
+ // --------------------------------------------------
+ @GET
+ @Path("/pves")
+ @Produces(Array(MediaType.APPLICATION_JSON))
+ def fetchPVEs(@QueryParam("cuid") cuid: Int): util.List[util.Map[String, Object]] = {
+
+ if (cuid == 0) {
Review Comment:
Is there a scenario where cuid can be 0?
##########
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()) {
Review Comment:
Same here. How to test the error?
##########
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) {
Review Comment:
Comment for this method. How you prettified the output.
##########
frontend/src/app/workspace/component/power-button/computing-unit-selection.component.ts:
##########
@@ -56,6 +56,27 @@ import {
isComputingUnitShmTooLarge,
getJvmMemorySliderConfig,
} from "../../../common/util/computing-unit.util";
+import { PvePackageResponse, WorkflowPveService } from "../../service/virtual-environment/virtual-environment.service";
+
+type PackageRow = {
+ name: string;
+ operator?: "==" | ">=" | "<=";
+ version?: string;
+ isHighlighted?: boolean;
+};
+
+type PveDraft = {
+ id: number;
Review Comment:
Do we need a PVE ID in this PR?
##########
amber/src/main/scala/org/apache/texera/web/resource/pythonvirtualenvironment/PveResource.scala:
##########
@@ -0,0 +1,230 @@
+/*
+ * 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 com.fasterxml.jackson.databind.ObjectMapper
+import org.glassfish.jersey.server.ChunkedOutput
+
+import java.util.concurrent.LinkedBlockingQueue
+import javax.ws.rs._
+import javax.ws.rs.core.MediaType
+import scala.concurrent.ExecutionContext.Implicits.global
+import scala.concurrent.Future
+import scala.jdk.CollectionConverters._
+import java.util
+
+case class PackageResponse(system: java.util.List[String], user: java.util.List[String])
+
+@Path("/pve")
+@Consumes(Array(MediaType.APPLICATION_JSON))
+class PveResource {
+
+ // --------------------------------------------------
+ // Create / Install packages (SSE)
+ // --------------------------------------------------
+ @GET
+ @Produces(Array("text/event-stream"))
+ def createPve(
+ @QueryParam("packages") packagesJson: String,
+ @QueryParam("cuid") cuid: Int,
+ @QueryParam("pveName") pveName: String
+ ): ChunkedOutput[String] = {
+
+ if (cuid == 0) {
+ throw new BadRequestException("Missing or invalid cuid")
+ }
+
+ if (pveName == null || pveName.trim.isEmpty) {
+ throw new BadRequestException("Missing environment name")
+ }
+
+ val mapper = new ObjectMapper()
+ val packages =
+ if (packagesJson == null || packagesJson.trim.isEmpty)
+ List.empty[String]
+ else
+ mapper.readValue(packagesJson, classOf[Array[String]]).toList
+
+ val queue = new LinkedBlockingQueue[String]()
+ val chunkedOutput = new ChunkedOutput[String](classOf[String])
+
+ Future {
+ try {
+
+ if (!PveManager.pveExists(cuid, pveName)) {
+ PveManager.createNewPve(cuid, queue, pveName)
+ }
+
+ if (packages.nonEmpty) {
+ PveManager.installPackages(packages, cuid, queue, pveName)
+ }
+
+ } catch {
+ case e: Exception =>
+ queue.put(s"[ERR] ${e.getMessage}")
+ } finally {
+ queue.put("__DONE__")
+ }
+ }
+
+ Future {
+ var done = false
+ while (!done) {
+ val line = queue.take()
+ if (line == "__DONE__") {
+ chunkedOutput.write("data: __DONE__\n\n")
+ done = true
+ } else chunkedOutput.write(s"data: $line\n\n")
+ }
+ chunkedOutput.close()
+ }
+
+ chunkedOutput
+ }
+
+ // --------------------------------------------------
+ // Get installed packages
+ // --------------------------------------------------
+ @GET
+ @Path("/packages")
+ @Produces(Array(MediaType.APPLICATION_JSON))
+ def getInstalledPackages(
+ @QueryParam("cuid") cuid: Int,
+ @QueryParam("pveName") pveName: String
+ ): util.Map[String, util.List[String]] = {
+
+ if (cuid == 0) {
Review Comment:
Is there a scenario where cuid can be 0?
##########
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(
Review Comment:
Need comment about this method. Why do we need a separate thread?
--
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: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add Python Virtual Environment Support [texera]
Posted by "SarahAsad23 (via GitHub)" <gi...@apache.org>.
SarahAsad23 commented on code in PR #4359:
URL: https://github.com/apache/texera/pull/4359#discussion_r3125686899
##########
amber/src/main/scala/org/apache/texera/web/TexeraWebApplication.scala:
##########
@@ -160,7 +161,7 @@ class TexeraWebApplication
environment.jersey.register(classOf[UserQuotaResource])
environment.jersey.register(classOf[AdminSettingsResource])
environment.jersey.register(classOf[AIAssistantResource])
-
Review Comment:
Yes. The PVE functionality is exposed through backend REST endpoints, so the resource needs to be registered in TexeraWebApplication for those API calls to be available.
--
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: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add Python Virtual Environment Support [texera]
Posted by "SarahAsad23 (via GitHub)" <gi...@apache.org>.
SarahAsad23 commented on code in PR #4359:
URL: https://github.com/apache/texera/pull/4359#discussion_r3106213004
##########
amber/src/main/scala/org/apache/texera/web/resource/pythonvirtualenvironment/WsPveResource.scala:
##########
Review Comment:
They have been merged.
--
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: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add Python Virtual Environment Support [texera]
Posted by "chenlica (via GitHub)" <gi...@apache.org>.
chenlica commented on PR #4359:
URL: https://github.com/apache/texera/pull/4359#issuecomment-4301990218
@SarahAsad23 It will be good to give a reason to close this PR. If there are new related PRs, please add those links here.
--
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: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add Python Virtual Environment Support [texera]
Posted by "SarahAsad23 (via GitHub)" <gi...@apache.org>.
SarahAsad23 commented on code in PR #4359:
URL: https://github.com/apache/texera/pull/4359#discussion_r3114016198
##########
bin/k8s/templates/gateway-routes.yaml:
##########
@@ -100,27 +101,47 @@ spec:
parentRefs:
- name: {{ .Release.Name }}-gateway
{{- if and .Values.gatewayConfig .Values.gatewayConfig.hostname }}
- hostnames:
- - {{ .Values.gatewayConfig.hostname }}
+hostnames:
+ - {{ .Values.gatewayConfig.hostname }}
{{- end }}
- rules:
- - matches:
- - path:
- type: PathPrefix
- value: /wsapi
- - path:
- type: RegularExpression
- value: "^/api/executions/\\d+/stats/\\d+$"
- - path:
- type: PathPrefix
- value: /api/executions/result/export
- backendRefs:
- - group: gateway.envoyproxy.io
- kind: Backend
- name: texera-dynamic-backend
+rules:
+ - matches:
+ - path:
+ type: PathPrefix
+ value: /pve
+ timeouts:
+ request: 10m
Review Comment:
To prevent the request from timing out when waiting for the pip output.
--
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: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: Add Python Virtual Environment Support [texera]
Posted by "kunwp1 (via GitHub)" <gi...@apache.org>.
kunwp1 commented on code in PR #4359:
URL: https://github.com/apache/texera/pull/4359#discussion_r3096023218
##########
access-control-service/src/main/scala/org/apache/texera/service/resource/AccessControlResource.scala:
##########
@@ -43,6 +43,8 @@ object AccessControlResource extends LazyLogging {
private val wsapiWorkflowWebsocket: Regex = """.*/wsapi/workflow-websocket.*""".r
private val apiExecutionsStats: Regex = """.*/api/executions/[0-9]+/stats/[0-9]+.*""".r
private val apiExecutionsResultExport: Regex = """.*/api/executions/result/export.*""".r
+ private val authPveRoot: Regex = """auth/pve/?""".r
Review Comment:
Better to combine these two regex into a single regex.
##########
frontend/src/app/workspace/component/power-button/computing-unit-selection.component.ts:
##########
@@ -61,6 +82,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;
Review Comment:
lowercase the PVE and capitalize M to make the style consistent with others.
##########
frontend/src/app/workspace/component/power-button/computing-unit-selection.component.ts:
##########
@@ -160,14 +189,18 @@ export class ComputingUnitSelectionComponent implements OnInit {
.subscribe(unit => {
const wid = this.workflowActionService.getWorkflowMetadata()?.wid;
- // ── compare with the *previous* cuid, not the one we are just about to store ──
+ // compare with the previous cuid, not the one we are just about to store
if (isDefined(wid) && unit?.computingUnit.cuid !== this.lastSelectedCuid) {
this.updateWorkflowModificationStatus(wid);
}
- // update local caches **after** the comparison
+ // update local caches after the comparison
Review Comment:
Revert this unnecessary change.
##########
frontend/src/app/workspace/component/power-button/computing-unit-selection.component.html:
##########
@@ -134,7 +134,7 @@
<span
*ngIf="unit.status === 'Pending'"
class="unit-status-indicator"
- >(Connecting)</span
+ >(Connecting)</span
Review Comment:
Revert this change.
##########
amber/src/main/scala/org/apache/texera/web/resource/pythonvirtualenvironment/PveManager.scala:
##########
@@ -0,0 +1,661 @@
+package org.apache.texera.web.resource.pythonvirtualenvironment
+
+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
+
+ /** Return (systemPkgs, userPkgs) as plain strings from metadata. */
+ 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 = {
Review Comment:
Remove these unnecessary comments.
##########
frontend/src/app/workspace/component/power-button/computing-unit-selection.component.ts:
##########
@@ -606,7 +639,7 @@ export class ComputingUnitSelectionComponent implements OnInit {
// Get the current memory in GB
const memoryValue = parseResourceNumber(this.selectedMemory);
const memoryUnit = parseResourceUnit(this.selectedMemory);
- let cuMemoryInGb = memoryUnit === "Gi" ? memoryValue : memoryUnit === "Mi" ? Math.floor(memoryValue / 1024) : 1;
+ const cuMemoryInGb = memoryUnit === "Gi" ? memoryValue : memoryUnit === "Mi" ? Math.floor(memoryValue / 1024) : 1;
Review Comment:
Revert this change.
##########
frontend/src/app/workspace/component/power-button/computing-unit-selection.component.ts:
##########
@@ -160,14 +189,18 @@ export class ComputingUnitSelectionComponent implements OnInit {
.subscribe(unit => {
const wid = this.workflowActionService.getWorkflowMetadata()?.wid;
- // ── compare with the *previous* cuid, not the one we are just about to store ──
+ // compare with the previous cuid, not the one we are just about to store
Review Comment:
Revert this unnecessary change.
##########
frontend/src/app/workspace/component/power-button/computing-unit-selection.component.html:
##########
@@ -90,7 +90,7 @@
<span
*ngIf="!selectedComputingUnit"
class="connect-text"
- >Connect</span
+ >Connect</span
Review Comment:
Revert this change.
##########
amber/src/main/scala/org/apache/texera/web/resource/pythonvirtualenvironment/WsPveResource.scala:
##########
Review Comment:
What's the purpose of separating WsPveResource from PveResource? Why not combine them?
##########
frontend/src/app/workspace/component/power-button/computing-unit-selection.component.ts:
##########
@@ -227,7 +260,7 @@ export class ComputingUnitSelectionComponent implements OnInit {
next: (latestWorkflowExecution: WorkflowExecutionsEntry) => {
this.selectComputingUnit(this.workflowId, latestWorkflowExecution.cuId);
},
- error: (err: unknown) => {
+ error: (_err: unknown) => {
Review Comment:
Revert this change.
--
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: notifications-unsubscribe@texera.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org