kunwp1 commented on code in PR #4484:
URL: https://github.com/apache/texera/pull/4484#discussion_r3133500385
##########
frontend/src/app/workspace/component/power-button/computing-unit-selection.component.html:
##########
@@ -427,3 +435,220 @@
</div>
</div>
</ng-template>
+
+<!-- Modal for adding packages -->
+<nz-modal
+ nzWrapClassName="pve-modal"
+ nzClassName="pve-modal"
+ [nzVisible]="pveModalVisible"
+ nzTitle="Python Environments"
+ (nzOnCancel)="closePveModal()"
+ [nzFooter]="customFooter">
+ <ng-template #customFooter>
+ <div class="footer-all">
+ <button
+ nz-button
+ nzType="default"
+ (click)="closePveModal()">
+ 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)">
Review Comment:
I think it's better to show the system packages regardless of whether the
user creates a venv or not. The UI is not very intuitive when the user don't
see anything here in the first place.
##########
frontend/src/app/workspace/component/power-button/computing-unit-selection.component.scss:
##########
Review Comment:
Remove the styles related to adding/removing user packages
##########
frontend/src/app/workspace/component/power-button/computing-unit-selection.component.ts:
##########
@@ -637,4 +669,255 @@ export class ComputingUnitSelectionComponent implements
OnInit {
this.computingUnitStatusService.refreshComputingUnitList();
}
}
+
+ private makeEmptyPve(expanded: boolean): PveDraft {
+ return {
+ id: this.nextPveId++,
+ name: "",
+ userPackages: [],
+ newPackages: [],
+ 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();
+ }
+
+ closePveModal(): void {
+ this.pves.forEach(pve => {
+ pve.source?.close();
+ pve.source = undefined;
+ pve.isInstalling = false;
+ });
+
+ this.pveModalVisible = false;
+ }
+
+ 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 => ({
+ id: this.nextPveId++,
+ 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(),
+ deleteToggle: false,
+ };
+ }),
+ newPackages: [],
+ deletingPackages: [],
+ }));
+
+ if (resp.length > 0) {
+ this.workflowPveService.setPveName(resp[0].pveName);
+
+ this.workflowPveService
+ .getInstalledPackages()
+ .pipe(untilDestroyed(this))
+ .subscribe({
+ next: installedResp => {
+ this.systemPackages = installedResp.system.map(pkgStr => {
+ const [name, version] = pkgStr.split("==");
+ return {
+ name: name.trim(),
+ version: (version ?? "").trim(),
+ };
+ });
+ },
+ error: (err: unknown) => {
+ console.error("Failed to fetch system packages:", err);
+ this.systemPackages = [];
+ },
+ });
+ } else {
+ this.systemPackages = [];
Review Comment:
Try to think of a way to fill-in this part automatically without creating
user virtual environment.
##########
frontend/src/app/workspace/component/power-button/computing-unit-selection.component.ts:
##########
@@ -637,4 +669,255 @@ export class ComputingUnitSelectionComponent implements
OnInit {
this.computingUnitStatusService.refreshComputingUnitList();
}
}
+
+ private makeEmptyPve(expanded: boolean): PveDraft {
+ return {
+ id: this.nextPveId++,
+ name: "",
+ userPackages: [],
+ newPackages: [],
+ 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();
+ }
+
+ closePveModal(): void {
+ this.pves.forEach(pve => {
+ pve.source?.close();
+ pve.source = undefined;
+ pve.isInstalling = false;
+ });
+
+ this.pveModalVisible = false;
+ }
+
+ 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 => ({
+ id: this.nextPveId++,
+ 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(),
+ deleteToggle: false,
+ };
+ }),
+ newPackages: [],
+ deletingPackages: [],
+ }));
+
+ if (resp.length > 0) {
+ this.workflowPveService.setPveName(resp[0].pveName);
+
+ this.workflowPveService
+ .getInstalledPackages()
+ .pipe(untilDestroyed(this))
+ .subscribe({
+ next: installedResp => {
+ this.systemPackages = installedResp.system.map(pkgStr => {
+ const [name, version] = pkgStr.split("==");
+ return {
+ name: name.trim(),
+ version: (version ?? "").trim(),
+ };
+ });
+ },
+ error: (err: unknown) => {
+ console.error("Failed to fetch system packages:", err);
+ this.systemPackages = [];
+ },
+ });
+ } else {
+ this.systemPackages = [];
+ }
+ },
+ error: (err: unknown) => {
+ console.error("Failed to fetch PVEs:", err);
+ this.pves = [];
+ this.systemPackages = [];
+ },
+ });
+ }
+
+ scrollToBottomOfPipModal(index: number) {
+ setTimeout(() => {
+ const pre = document.getElementById(`pip-log-${index}`) as HTMLElement |
null;
+ if (pre) {
+ pre.scrollTop = pre.scrollHeight;
+ }
+ }, 50);
+ }
+
+ // Converts raw pip output for UI rendering by escaping unsafe characters and
+ // applying styling to exit codes, errors, warnings, and common success
messages.
+ 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.");
Review Comment:
Same here. At this point when `createVirtualEnvironment` is called, I don't
see a case where cuId is null. You already covered this edge case from the UI.
You can remove notification service error.
##########
amber/src/main/scala/org/apache/texera/web/resource/pythonvirtualenvironment/PveManager.scala:
##########
Review Comment:
Remove all the user package related logics.
##########
amber/src/main/scala/org/apache/texera/web/resource/pythonvirtualenvironment/PveManager.scala:
##########
@@ -0,0 +1,333 @@
+/*
+ * 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 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._
+
+/**
+ * PveManager is responsible for managing Python Virtual Environments (PVEs)
+ * for each Computing Unit
+ *
+ * It supports:
+ * - Creating and initializing isolated Python environments
+ * - Installing and uninstalling Python packages
+ * - Tracking system vs user-installed packages via metadata files
+ * - Streaming pip output logs back to the caller
+ *
+ * Each PVE is stored under:
+ * /tmp/texera-pve/venvs/{cuid}/{pveName}/
+ *
+ * The structure includes:
+ * - pve/ -> actual virtual environment
+ * - metadata/ -> package tracking (system + user)
+ */
+
+object PveManager {
+
+ private val VenvRoot: Path = Paths.get("/tmp/texera-pve/venvs")
+
+ private def cuidDir(cuid: Int, pvename: String): Path = {
+ val dir = VenvRoot.resolve(cuid.toString).resolve(pvename)
+ Files.createDirectories(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 writeMetadata(path: Path, lines: Seq[String]): Unit = {
+ if (path.getParent != null) {
+ Files.createDirectories(path.getParent)
+ }
+ 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 pipEnv: Map[String, String] =
+ Map(
+ "PYTHONUNBUFFERED" -> "1",
+ "PIP_PROGRESS_BAR" -> "off",
+ "PIP_DISABLE_PIP_VERSION_CHECK" -> "1",
+ "PIP_NO_INPUT" -> "1"
+ )
+
+ def getSystemAndUserPackages(cuid: Int, pveName: String): (Seq[String],
Seq[String]) = {
+ val sys = readMetadataList(systemPackagesPath(cuid, pveName))
+ val usr = readMetadataList(userPackagesPath(cuid, pveName))
+ (sys, usr)
+ }
+
+ /**
+ * Creates a new PVE for a CU.
+ *
+ * Behavior:
+ * - If a base PVE exists (PVE_BASE), it clones it for faster setup
+ * - Otherwise, creates a fresh venv and installs dependencies
+ *
+ * Steps:
+ * 1. Remove existing environment (if present)
+ * 2. Create or copy base environment
+ * 3. Install system + operator dependencies
+ * 4. Record installed packages as system metadata
+ *
+ * Logs progress to the provided queue.
+ */
+ 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
+ val python = pythonBinPath(cuid, pveName).toAbsolutePath.toString
+ val envVars = pipEnv
+
+ val pveBase = sys.env.getOrElse("PVE_BASE", "/opt/pve-base")
+ val basePython = Paths.get(pveBase).resolve("bin").resolve("python")
+ val hasBasePve = Files.exists(basePython)
+
+ val requirementsPath =
+ if (!hasBasePve) Paths.get("amber", "requirements.txt")
+ else Paths.get("/tmp", "requirements.txt")
+
+ val operatorRequirementsPath =
+ if (!hasBasePve) Paths.get("amber", "operator-requirements.txt")
+ else Paths.get("/tmp", "operator-requirements.txt")
+
+ 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")
+ }
+
+ Files.createDirectories(venvDirPath.getParent)
+
+ 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
+ }
+
+ if (!Files.exists(requirementsPath)) {
+ queue.put(s"[PVE][ERR] requirements.txt not found at
${requirementsPath.toAbsolutePath}")
+ return
+ }
+
+ if (!Files.exists(operatorRequirementsPath)) {
+ queue.put(
+ s"[PVE][ERR] operator-requirements.txt not found at
${operatorRequirementsPath.toAbsolutePath}"
+ )
+ return
+ }
+
+ Files.createDirectories(metadataDir(cuid, pveName))
+
+ queue.put(s"[PVE] Installing local base requirements from
${requirementsPath.toAbsolutePath}")
+
+ val installReqCode = Process(
+ Seq(python, "-m", "pip", "install", "-r", requirementsPath.toString),
+ None,
+ envVars.toSeq: _*
+ ).!(
+ ProcessLogger(
+ out => queue.put(s"[pip] $out"),
+ err => queue.put(s"[pip][ERR] $err")
+ )
+ )
+ queue.put(s"[PVE] requirements install finished with exit code
$installReqCode")
+
+ if (installReqCode != 0) {
+ queue.put(s"[PVE][ERR] Failed to install requirements.txt
(exit=$installReqCode)")
+ return
+ }
+
+ queue.put(
+ s"[PVE] Installing local operator requirements from
${operatorRequirementsPath.toAbsolutePath}"
+ )
+
+ val installOperatorReqCode = Process(
+ Seq(python, "-m", "pip", "install", "-r",
operatorRequirementsPath.toString),
+ None,
+ envVars.toSeq: _*
+ ).!(
+ ProcessLogger(
+ out => queue.put(s"[pip] $out"),
+ err => queue.put(s"[pip][ERR] $err")
+ )
+ )
+ queue.put(
+ s"[PVE] operator requirements install finished with exit code
$installOperatorReqCode"
+ )
+
+ if (installOperatorReqCode != 0) {
+ queue.put(
+ s"[PVE][ERR] Failed to install operator-requirements.txt
(exit=$installOperatorReqCode)"
+ )
+ return
+ }
+
+ queue.put("[PVE] Running pip freeze")
+ 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")
+ }
+
+ Files.createDirectories(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
+ )
+ ).!(
+ ProcessLogger(
+ out => queue.put(s"[pve] $out"),
+ err => queue.put(s"[pve][ERR] $err")
+ )
+ )
+ queue.put(s"[pve] rewrite finished with exit code $fixCode")
+
+ Files.createDirectories(metadataDir(cuid, pveName))
+
+ queue.put("[PVE] Base environment copied; skipping system requirements
install.")
+
+ queue.put("[PVE] Running pip freeze")
+ val freezeOutput = Process(Seq(python, "-m", "pip", "freeze"), None,
envVars.toSeq: _*).!!
+ val systemFreezeLines =
freezeOutput.split("\n").map(_.trim).filter(_.nonEmpty).toSeq
+
+ writeMetadata(systemPackagesPath(cuid, pveName), systemFreezeLines)
+ writeMetadata(userPackagesPath(cuid, pveName), Seq.empty)
+
+ queue.put(s"[PVE] Created new environment for cuid=$cuid")
+ }
+
+ def pveExists(cuid: Int, pveName: String): Boolean =
+ Files.exists(pythonBinPath(cuid, pveName)) &&
Files.exists(pipBinPath(cuid, pveName))
+
+ def getEnvironments(cuid: Int): List[String] = {
+
+ val cuPath = Paths.get("/tmp/texera-pve/venvs").resolve(cuid.toString)
Review Comment:
Why not use `cuidDir` function?
##########
amber/src/main/scala/org/apache/texera/web/resource/pythonvirtualenvironment/PveManager.scala:
##########
@@ -0,0 +1,333 @@
+/*
+ * 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 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._
+
+/**
+ * PveManager is responsible for managing Python Virtual Environments (PVEs)
+ * for each Computing Unit
+ *
+ * It supports:
+ * - Creating and initializing isolated Python environments
+ * - Installing and uninstalling Python packages
+ * - Tracking system vs user-installed packages via metadata files
+ * - Streaming pip output logs back to the caller
+ *
+ * Each PVE is stored under:
+ * /tmp/texera-pve/venvs/{cuid}/{pveName}/
+ *
+ * The structure includes:
+ * - pve/ -> actual virtual environment
+ * - metadata/ -> package tracking (system + user)
+ */
+
+object PveManager {
+
+ private val VenvRoot: Path = Paths.get("/tmp/texera-pve/venvs")
+
+ private def cuidDir(cuid: Int, pvename: String): Path = {
+ val dir = VenvRoot.resolve(cuid.toString).resolve(pvename)
+ Files.createDirectories(dir)
Review Comment:
All the methods below are all calling `Files.createDirectories(dir)` in the
`cuidDir()`. It doesn't look right. Consider calling this somewhere else.
##########
frontend/src/app/workspace/component/power-button/computing-unit-selection.component.ts:
##########
@@ -637,4 +669,255 @@ export class ComputingUnitSelectionComponent implements
OnInit {
this.computingUnitStatusService.refreshComputingUnitList();
}
}
+
+ private makeEmptyPve(expanded: boolean): PveDraft {
+ return {
+ id: this.nextPveId++,
+ name: "",
+ userPackages: [],
+ newPackages: [],
+ deletingPackages: [],
+ pipOutput: "",
+ prettyPipOutput: "",
+ expanded,
+ isInstalling: false,
+ };
+ }
+
+ addEnvironment(): void {
+ this.pves.push(this.makeEmptyPve(true));
+ }
+
+ trackByPveId(_index: number, pve: PveDraft): number {
Review Comment:
Why not track by `_index`? We can remove `id`
##########
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;
+ deleteToggle?: boolean;
+};
+
+type PveDraft = {
+ id: number;
Review Comment:
I think we can safely remove `id`. I don't see any places that are using
`id` (neither displaying in the UI nor using it on the backend). The backend is
also using the pve name instead of `id`.
##########
frontend/src/app/workspace/component/power-button/computing-unit-selection.component.ts:
##########
@@ -637,4 +669,255 @@ export class ComputingUnitSelectionComponent implements
OnInit {
this.computingUnitStatusService.refreshComputingUnitList();
}
}
+
+ private makeEmptyPve(expanded: boolean): PveDraft {
Review Comment:
You can remove this method if you address the comment above.
##########
frontend/src/app/workspace/component/power-button/computing-unit-selection.component.ts:
##########
Review Comment:
Can you remove user package related entries from this PR?
##########
frontend/src/app/workspace/component/power-button/computing-unit-selection.component.ts:
##########
@@ -637,4 +669,255 @@ export class ComputingUnitSelectionComponent implements
OnInit {
this.computingUnitStatusService.refreshComputingUnitList();
}
}
+
+ private makeEmptyPve(expanded: boolean): PveDraft {
+ return {
+ id: this.nextPveId++,
+ name: "",
+ userPackages: [],
+ newPackages: [],
+ 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();
+ }
+
+ closePveModal(): void {
+ this.pves.forEach(pve => {
+ pve.source?.close();
+ pve.source = undefined;
+ pve.isInstalling = false;
+ });
+
+ this.pveModalVisible = false;
+ }
+
+ 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 => ({
+ id: this.nextPveId++,
+ 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(),
+ deleteToggle: false,
+ };
+ }),
+ newPackages: [],
+ deletingPackages: [],
+ }));
+
+ if (resp.length > 0) {
+ this.workflowPveService.setPveName(resp[0].pveName);
+
+ this.workflowPveService
+ .getInstalledPackages()
+ .pipe(untilDestroyed(this))
+ .subscribe({
+ next: installedResp => {
+ this.systemPackages = installedResp.system.map(pkgStr => {
+ const [name, version] = pkgStr.split("==");
+ return {
+ name: name.trim(),
+ version: (version ?? "").trim(),
+ };
+ });
+ },
+ error: (err: unknown) => {
+ console.error("Failed to fetch system packages:", err);
+ this.systemPackages = [];
+ },
+ });
+ } else {
+ this.systemPackages = [];
+ }
+ },
+ error: (err: unknown) => {
+ console.error("Failed to fetch PVEs:", err);
+ this.pves = [];
+ this.systemPackages = [];
+ },
+ });
+ }
+
+ scrollToBottomOfPipModal(index: number) {
+ setTimeout(() => {
+ const pre = document.getElementById(`pip-log-${index}`) as HTMLElement |
null;
+ if (pre) {
+ pre.scrollTop = pre.scrollHeight;
+ }
+ }, 50);
+ }
+
+ // Converts raw pip output for UI rendering by escaping unsafe characters and
+ // applying styling to exit codes, errors, warnings, and common success
messages.
+ 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.");
Review Comment:
Same here. `OK` button is disabled when env name is not defined.
##########
frontend/src/app/workspace/component/power-button/computing-unit-selection.component.ts:
##########
@@ -64,6 +85,12 @@ import {
styleUrls: ["./computing-unit-selection.component.scss"],
})
export class ComputingUnitSelectionComponent implements OnInit {
+ // variables for creating a virtual environment
+ nextPveId = 1;
+ pves: PveDraft[] = [this.makeEmptyPve(true)];
Review Comment:
Make it `[]`? Because `showPVModalVisible()` will fill out this list.
##########
frontend/src/app/workspace/component/power-button/computing-unit-selection.component.html:
##########
@@ -427,3 +435,220 @@
</div>
</div>
</ng-template>
+
+<!-- Modal for adding packages -->
+<nz-modal
+ nzWrapClassName="pve-modal"
+ nzClassName="pve-modal"
+ [nzVisible]="pveModalVisible"
+ nzTitle="Python Environments"
+ (nzOnCancel)="closePveModal()"
+ [nzFooter]="customFooter">
+ <ng-template #customFooter>
+ <div class="footer-all">
+ <button
+ nz-button
+ nzType="default"
+ (click)="closePveModal()">
+ 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.deleteToggle }">
+ <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" />
+ </div>
+
+ <div class="field operator operator-select">
+ <label style="visibility: hidden">Op</label>
+ <nz-select nzPlaceHolder="Select">
+ <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)" />
+ </div>
+ </div>
+
+ <button
+ nz-button
+ nzType="default"
+ nzShape="circle"
+ nzDanger
+ [ngClass]="{ 'highlighted-btn': pkg.deleteToggle }">
+ <i
+ nz-icon
+ nzType="delete"></i>
+ </button>
+ </div>
+
+ <div class="add-btn">
+ <button
+ nz-button
+ nzType="primary"
+ nzShape="circle">
+ <i
+ nz-icon
+ nzType="plus"></i>
+ </button>
+ </div>
+
Review Comment:
Remove these parts as they are not related to this PR. Add it in the next PR.
##########
frontend/src/app/workspace/component/power-button/computing-unit-selection.component.ts:
##########
@@ -637,4 +669,255 @@ export class ComputingUnitSelectionComponent implements
OnInit {
this.computingUnitStatusService.refreshComputingUnitList();
}
}
+
+ private makeEmptyPve(expanded: boolean): PveDraft {
+ return {
+ id: this.nextPveId++,
+ name: "",
+ userPackages: [],
+ newPackages: [],
+ 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();
+ }
+
+ closePveModal(): void {
+ this.pves.forEach(pve => {
+ pve.source?.close();
+ pve.source = undefined;
+ pve.isInstalling = false;
+ });
+
+ this.pveModalVisible = false;
+ }
+
+ 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 => ({
+ id: this.nextPveId++,
+ 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(),
+ deleteToggle: false,
+ };
+ }),
+ newPackages: [],
+ deletingPackages: [],
+ }));
+
+ if (resp.length > 0) {
+ this.workflowPveService.setPveName(resp[0].pveName);
+
+ this.workflowPveService
+ .getInstalledPackages()
+ .pipe(untilDestroyed(this))
+ .subscribe({
+ next: installedResp => {
+ this.systemPackages = installedResp.system.map(pkgStr => {
+ const [name, version] = pkgStr.split("==");
+ return {
+ name: name.trim(),
+ version: (version ?? "").trim(),
+ };
+ });
+ },
+ error: (err: unknown) => {
+ console.error("Failed to fetch system packages:", err);
+ this.systemPackages = [];
+ },
+ });
+ } else {
+ this.systemPackages = [];
+ }
+ },
+ error: (err: unknown) => {
+ console.error("Failed to fetch PVEs:", err);
+ this.pves = [];
+ this.systemPackages = [];
+ },
+ });
+ }
+
+ scrollToBottomOfPipModal(index: number) {
+ setTimeout(() => {
+ const pre = document.getElementById(`pip-log-${index}`) as HTMLElement |
null;
+ if (pre) {
+ pre.scrollTop = pre.scrollHeight;
+ }
+ }, 50);
+ }
+
+ // Converts raw pip output for UI rendering by escaping unsafe characters and
+ // applying styling to exit codes, errors, warnings, and common success
messages.
+ 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;
+ }
+
+ const trimmedName = env.name.trim().toLowerCase();
+ env.name = trimmedName;
+
+ if (!/^[a-zA-Z0-9]+$/.test(trimmedName)) {
Review Comment:
Move this check above line 829 or remove `A-Z`
##########
frontend/src/app/workspace/component/power-button/computing-unit-selection.component.ts:
##########
@@ -637,4 +669,255 @@ export class ComputingUnitSelectionComponent implements
OnInit {
this.computingUnitStatusService.refreshComputingUnitList();
}
}
+
+ private makeEmptyPve(expanded: boolean): PveDraft {
+ return {
+ id: this.nextPveId++,
+ name: "",
+ userPackages: [],
+ newPackages: [],
+ 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();
+ }
+
+ closePveModal(): void {
+ this.pves.forEach(pve => {
+ pve.source?.close();
+ pve.source = undefined;
+ pve.isInstalling = false;
+ });
+
+ this.pveModalVisible = false;
+ }
+
+ 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 still couldn't understand the reason you added this notification service
error. `getPVEs` is called from `showPVEmodalVisible()` and I don't see a case
where unit is null in the html. This seems like a dead code.
```
<i
nz-icon
nzType="plus"
nz-tooltip
*ngIf="unit.isOwner"
[nzTooltipTitle]="'Python Environment'"
(click)="selectedComputingUnit = unit; showPVEmodalVisible();
$event.stopPropagation()">
</i>
```
##########
frontend/src/app/workspace/component/power-button/computing-unit-selection.component.ts:
##########
@@ -637,4 +669,255 @@ export class ComputingUnitSelectionComponent implements
OnInit {
this.computingUnitStatusService.refreshComputingUnitList();
}
}
+
+ private makeEmptyPve(expanded: boolean): PveDraft {
+ return {
+ id: this.nextPveId++,
+ name: "",
+ userPackages: [],
+ newPackages: [],
+ 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();
+ }
+
+ closePveModal(): void {
+ this.pves.forEach(pve => {
+ pve.source?.close();
+ pve.source = undefined;
+ pve.isInstalling = false;
+ });
+
+ this.pveModalVisible = false;
+ }
+
+ 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 => ({
+ id: this.nextPveId++,
+ 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(),
+ deleteToggle: false,
+ };
+ }),
+ newPackages: [],
+ deletingPackages: [],
+ }));
+
+ if (resp.length > 0) {
+ this.workflowPveService.setPveName(resp[0].pveName);
+
+ this.workflowPveService
+ .getInstalledPackages()
+ .pipe(untilDestroyed(this))
+ .subscribe({
+ next: installedResp => {
+ this.systemPackages = installedResp.system.map(pkgStr => {
+ const [name, version] = pkgStr.split("==");
+ return {
+ name: name.trim(),
+ version: (version ?? "").trim(),
+ };
+ });
+ },
+ error: (err: unknown) => {
+ console.error("Failed to fetch system packages:", err);
+ this.systemPackages = [];
+ },
+ });
+ } else {
+ this.systemPackages = [];
+ }
+ },
+ error: (err: unknown) => {
+ console.error("Failed to fetch PVEs:", err);
+ this.pves = [];
+ this.systemPackages = [];
+ },
+ });
+ }
+
+ scrollToBottomOfPipModal(index: number) {
+ setTimeout(() => {
+ const pre = document.getElementById(`pip-log-${index}`) as HTMLElement |
null;
+ if (pre) {
+ pre.scrollTop = pre.scrollHeight;
+ }
+ }, 50);
+ }
+
+ // Converts raw pip output for UI rendering by escaping unsafe characters and
+ // applying styling to exit codes, errors, warnings, and common success
messages.
+ 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;
+ }
+
+ const trimmedName = env.name.trim().toLowerCase();
Review Comment:
Why do we need to lowercase the environment name? Is this a restriction of
Python? If not, please remove `toLowerCase()`. If so, please leave a comment
that it's a restriction of Python.
##########
frontend/src/app/workspace/service/virtual-environment/virtual-environment.service.ts:
##########
@@ -0,0 +1,101 @@
+/*
+ * 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 { Injectable } from "@angular/core";
+import { BehaviorSubject, Observable } from "rxjs";
+import { HttpClient, HttpParams } from "@angular/common/http";
+import { AuthService } from "../../../common/service/user/auth.service";
+
+export interface PackageResponse {
+ system: string[];
+ user: string[];
+}
+
+export interface PvePackageResponse {
+ pveName: string;
+ userPackages: string[];
+}
+
+@Injectable({ providedIn: "root" })
+export class WorkflowPveService {
+ private cuidSubject = new BehaviorSubject<number | null>(null);
Review Comment:
I don't see the need of having `BehaviorSubject` on this service. The
cleaner way is to make this a stateless service API and pass the cuid and pve
name directly into each service call.
##########
frontend/src/app/workspace/service/virtual-environment/virtual-environment.service.ts:
##########
@@ -0,0 +1,101 @@
+/*
+ * 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 { Injectable } from "@angular/core";
+import { BehaviorSubject, Observable } from "rxjs";
+import { HttpClient, HttpParams } from "@angular/common/http";
+import { AuthService } from "../../../common/service/user/auth.service";
+
+export interface PackageResponse {
+ system: string[];
+ user: string[];
+}
+
+export interface PvePackageResponse {
+ pveName: string;
+ userPackages: string[];
+}
+
+@Injectable({ providedIn: "root" })
+export class WorkflowPveService {
+ private cuidSubject = new BehaviorSubject<number | null>(null);
+
+ constructor(private http: HttpClient) {}
+
+ private pveNameSubject = new BehaviorSubject<string | null>(null);
Review Comment:
Same here.
##########
frontend/src/app/workspace/service/virtual-environment/virtual-environment.service.ts:
##########
@@ -0,0 +1,101 @@
+/*
+ * 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 { Injectable } from "@angular/core";
+import { BehaviorSubject, Observable } from "rxjs";
+import { HttpClient, HttpParams } from "@angular/common/http";
+import { AuthService } from "../../../common/service/user/auth.service";
+
+export interface PackageResponse {
+ system: string[];
+ user: string[];
+}
+
+export interface PvePackageResponse {
+ pveName: string;
+ userPackages: string[];
+}
+
+@Injectable({ providedIn: "root" })
+export class WorkflowPveService {
+ private cuidSubject = new BehaviorSubject<number | null>(null);
+
+ constructor(private http: HttpClient) {}
+
+ private pveNameSubject = new BehaviorSubject<string | null>(null);
+
+ setCuid(cuid: number): void {
+ this.cuidSubject.next(cuid);
+ }
+
+ setPveName(pveName: string): void {
+ this.pveNameSubject.next(pveName);
+ }
+
+ private requireCuid(): number {
+ const cuid = this.cuidSubject.value;
+ if (cuid === null) {
+ throw new Error("cuid is not set");
+ }
+ return cuid;
+ }
+
+ private requirePveName(): string {
+ const pveName = this.pveNameSubject.value;
+ if (pveName === null) {
+ throw new Error("Environment Name is not set");
+ }
+
+ return pveName;
+ }
Review Comment:
Remove all these and just simply pass cuid and pve name to each API.
##########
frontend/src/app/workspace/component/power-button/computing-unit-selection.component.ts:
##########
@@ -637,4 +669,255 @@ export class ComputingUnitSelectionComponent implements
OnInit {
this.computingUnitStatusService.refreshComputingUnitList();
}
}
+
+ private makeEmptyPve(expanded: boolean): PveDraft {
+ return {
+ id: this.nextPveId++,
+ name: "",
+ userPackages: [],
+ newPackages: [],
+ 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();
+ }
+
+ closePveModal(): void {
+ this.pves.forEach(pve => {
+ pve.source?.close();
+ pve.source = undefined;
+ pve.isInstalling = false;
+ });
+
+ this.pveModalVisible = false;
+ }
+
+ 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 => ({
+ id: this.nextPveId++,
+ 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(),
+ deleteToggle: false,
+ };
+ }),
+ newPackages: [],
+ deletingPackages: [],
+ }));
+
+ if (resp.length > 0) {
+ this.workflowPveService.setPveName(resp[0].pveName);
+
+ this.workflowPveService
+ .getInstalledPackages()
+ .pipe(untilDestroyed(this))
+ .subscribe({
+ next: installedResp => {
+ this.systemPackages = installedResp.system.map(pkgStr => {
+ const [name, version] = pkgStr.split("==");
+ return {
+ name: name.trim(),
+ version: (version ?? "").trim(),
+ };
+ });
+ },
+ error: (err: unknown) => {
+ console.error("Failed to fetch system packages:", err);
+ this.systemPackages = [];
+ },
+ });
+ } else {
+ this.systemPackages = [];
+ }
+ },
+ error: (err: unknown) => {
+ console.error("Failed to fetch PVEs:", err);
+ this.pves = [];
+ this.systemPackages = [];
+ },
+ });
+ }
+
+ scrollToBottomOfPipModal(index: number) {
+ setTimeout(() => {
+ const pre = document.getElementById(`pip-log-${index}`) as HTMLElement |
null;
+ if (pre) {
+ pre.scrollTop = pre.scrollHeight;
+ }
+ }, 50);
+ }
+
+ // Converts raw pip output for UI rendering by escaping unsafe characters and
+ // applying styling to exit codes, errors, warnings, and common success
messages.
+ 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;
+ }
+
+ const trimmedName = env.name.trim().toLowerCase();
+ env.name = trimmedName;
+
+ if (!/^[a-zA-Z0-9]+$/.test(trimmedName)) {
+ this.notificationService.error("Environment name must contain only
letters and numbers.");
+ return;
+ }
+
+ const duplicateExists = this.pves.some(
+ (pve, i) => i !== index && (pve.name ?? "").trim().toLowerCase() ===
trimmedName
+ );
+
+ if (duplicateExists) {
+ this.notificationService.error("An environment with this name already
exists.");
+ return;
+ }
+
+ env.isInstalling = true;
+
+ this.workflowPveService.setCuid(cuId);
+ this.workflowPveService.setPveName(env.name);
+
+ const packageArray: string[] = [];
Review Comment:
Remove user packages from this PR
##########
frontend/src/app/workspace/component/power-button/computing-unit-selection.component.ts:
##########
@@ -637,4 +669,255 @@ export class ComputingUnitSelectionComponent implements
OnInit {
this.computingUnitStatusService.refreshComputingUnitList();
}
}
+
+ private makeEmptyPve(expanded: boolean): PveDraft {
+ return {
+ id: this.nextPveId++,
+ name: "",
+ userPackages: [],
+ newPackages: [],
+ 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();
+ }
+
+ closePveModal(): void {
+ this.pves.forEach(pve => {
+ pve.source?.close();
+ pve.source = undefined;
+ pve.isInstalling = false;
+ });
+
+ this.pveModalVisible = false;
+ }
+
+ 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 => ({
+ id: this.nextPveId++,
+ 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(),
+ deleteToggle: false,
+ };
+ }),
+ newPackages: [],
+ deletingPackages: [],
+ }));
+
+ if (resp.length > 0) {
+ this.workflowPveService.setPveName(resp[0].pveName);
+
+ this.workflowPveService
+ .getInstalledPackages()
+ .pipe(untilDestroyed(this))
+ .subscribe({
+ next: installedResp => {
+ this.systemPackages = installedResp.system.map(pkgStr => {
+ const [name, version] = pkgStr.split("==");
+ return {
+ name: name.trim(),
+ version: (version ?? "").trim(),
+ };
+ });
+ },
+ error: (err: unknown) => {
+ console.error("Failed to fetch system packages:", err);
+ this.systemPackages = [];
+ },
+ });
+ } else {
+ this.systemPackages = [];
+ }
+ },
+ error: (err: unknown) => {
+ console.error("Failed to fetch PVEs:", err);
+ this.pves = [];
+ this.systemPackages = [];
+ },
+ });
+ }
+
+ scrollToBottomOfPipModal(index: number) {
+ setTimeout(() => {
+ const pre = document.getElementById(`pip-log-${index}`) as HTMLElement |
null;
+ if (pre) {
+ pre.scrollTop = pre.scrollHeight;
+ }
+ }, 50);
+ }
+
+ // Converts raw pip output for UI rendering by escaping unsafe characters and
+ // applying styling to exit codes, errors, warnings, and common success
messages.
+ 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;
+ }
+
+ const trimmedName = env.name.trim().toLowerCase();
+ env.name = trimmedName;
+
+ if (!/^[a-zA-Z0-9]+$/.test(trimmedName)) {
+ this.notificationService.error("Environment name must contain only
letters and numbers.");
+ return;
+ }
+
+ const duplicateExists = this.pves.some(
+ (pve, i) => i !== index && (pve.name ?? "").trim().toLowerCase() ===
trimmedName
+ );
+
+ if (duplicateExists) {
+ this.notificationService.error("An environment with this name already
exists.");
+ return;
+ }
+
+ env.isInstalling = true;
+
+ this.workflowPveService.setCuid(cuId);
+ this.workflowPveService.setPveName(env.name);
+
+ 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);
+ }
+ }
+
+ const token = localStorage.getItem("access_token") ?? "";
Review Comment:
Use `AuthService.getAccessToken()`
##########
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])
-
+ environment.jersey.register(classOf[PveResource])
Review Comment:
I still didn't get the reason why PveResource is registered on the texera
webserver, not on the computing unit master. I don't see any architecture
diagram or description that explains why texera-webserver is involved in
creating virtual environment.
##########
amber/src/main/scala/org/apache/texera/web/resource/pythonvirtualenvironment/PveResource.scala:
##########
@@ -0,0 +1,134 @@
+/*
+ * 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.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] = {
+ val queue = new LinkedBlockingQueue[String]()
+ val chunkedOutput = new ChunkedOutput[String](classOf[String])
+
+ Future {
+ try {
+
+ if (!PveManager.pveExists(cuid, pveName)) {
Review Comment:
Why do we need this check? I believe the frontend already checks this. Which
scenario are you trying to solve?
##########
frontend/src/app/workspace/component/power-button/computing-unit-selection.component.ts:
##########
@@ -637,4 +669,255 @@ export class ComputingUnitSelectionComponent implements
OnInit {
this.computingUnitStatusService.refreshComputingUnitList();
}
}
+
+ private makeEmptyPve(expanded: boolean): PveDraft {
+ return {
+ id: this.nextPveId++,
+ name: "",
+ userPackages: [],
+ newPackages: [],
+ 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();
+ }
+
+ closePveModal(): void {
+ this.pves.forEach(pve => {
+ pve.source?.close();
+ pve.source = undefined;
+ pve.isInstalling = false;
+ });
+
+ this.pveModalVisible = false;
+ }
+
+ 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 => ({
+ id: this.nextPveId++,
+ 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(),
+ deleteToggle: false,
+ };
+ }),
+ newPackages: [],
+ deletingPackages: [],
+ }));
+
+ if (resp.length > 0) {
+ this.workflowPveService.setPveName(resp[0].pveName);
+
+ this.workflowPveService
+ .getInstalledPackages()
+ .pipe(untilDestroyed(this))
+ .subscribe({
+ next: installedResp => {
+ this.systemPackages = installedResp.system.map(pkgStr => {
+ const [name, version] = pkgStr.split("==");
+ return {
+ name: name.trim(),
+ version: (version ?? "").trim(),
+ };
+ });
+ },
+ error: (err: unknown) => {
+ console.error("Failed to fetch system packages:", err);
+ this.systemPackages = [];
+ },
+ });
+ } else {
+ this.systemPackages = [];
+ }
+ },
+ error: (err: unknown) => {
+ console.error("Failed to fetch PVEs:", err);
+ this.pves = [];
+ this.systemPackages = [];
+ },
+ });
+ }
+
+ scrollToBottomOfPipModal(index: number) {
+ setTimeout(() => {
+ const pre = document.getElementById(`pip-log-${index}`) as HTMLElement |
null;
+ if (pre) {
+ pre.scrollTop = pre.scrollHeight;
+ }
+ }, 50);
+ }
+
+ // Converts raw pip output for UI rendering by escaping unsafe characters and
+ // applying styling to exit codes, errors, warnings, and common success
messages.
+ 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;
+ }
+
+ const trimmedName = env.name.trim().toLowerCase();
+ env.name = trimmedName;
+
+ if (!/^[a-zA-Z0-9]+$/.test(trimmedName)) {
+ this.notificationService.error("Environment name must contain only
letters and numbers.");
+ return;
+ }
+
+ const duplicateExists = this.pves.some(
+ (pve, i) => i !== index && (pve.name ?? "").trim().toLowerCase() ===
trimmedName
+ );
+
+ if (duplicateExists) {
+ this.notificationService.error("An environment with this name already
exists.");
+ return;
+ }
+
+ env.isInstalling = true;
+
+ this.workflowPveService.setCuid(cuId);
+ this.workflowPveService.setPveName(env.name);
+
+ 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);
+ }
+ }
+
+ const token = localStorage.getItem("access_token") ?? "";
+ const tokenParam = token ? `&access-token=${encodeURIComponent(token)}` :
"";
+ const query = encodeURIComponent(JSON.stringify(packageArray));
+
+ env.source?.close();
+ env.source = undefined;
+
+ const url = `/pve/?packages=${query}` + `&cuid=${cuId}` +
`&pveName=${encodeURIComponent(env.name)}` + tokenParam;
+
+ const source = new EventSource(url);
+ env.source = source;
+
+ env.pipOutput += "Starting ...";
+ this.updatePrettyPipOutput(index);
Review Comment:
Is it possible to make the pip output live stream? It would be useful to the
user because user doesn't know the progress behind the scene until the user
sees the final result.
##########
amber/src/main/scala/org/apache/texera/web/resource/pythonvirtualenvironment/PveManager.scala:
##########
@@ -0,0 +1,333 @@
+/*
+ * 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 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._
+
+/**
+ * PveManager is responsible for managing Python Virtual Environments (PVEs)
+ * for each Computing Unit
+ *
+ * It supports:
+ * - Creating and initializing isolated Python environments
+ * - Installing and uninstalling Python packages
+ * - Tracking system vs user-installed packages via metadata files
+ * - Streaming pip output logs back to the caller
+ *
+ * Each PVE is stored under:
+ * /tmp/texera-pve/venvs/{cuid}/{pveName}/
+ *
+ * The structure includes:
+ * - pve/ -> actual virtual environment
+ * - metadata/ -> package tracking (system + user)
+ */
+
+object PveManager {
+
+ private val VenvRoot: Path = Paths.get("/tmp/texera-pve/venvs")
+
+ private def cuidDir(cuid: Int, pvename: String): Path = {
+ val dir = VenvRoot.resolve(cuid.toString).resolve(pvename)
+ Files.createDirectories(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 writeMetadata(path: Path, lines: Seq[String]): Unit = {
+ if (path.getParent != null) {
+ Files.createDirectories(path.getParent)
+ }
+ 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 pipEnv: Map[String, String] =
+ Map(
+ "PYTHONUNBUFFERED" -> "1",
+ "PIP_PROGRESS_BAR" -> "off",
+ "PIP_DISABLE_PIP_VERSION_CHECK" -> "1",
+ "PIP_NO_INPUT" -> "1"
+ )
+
+ def getSystemAndUserPackages(cuid: Int, pveName: String): (Seq[String],
Seq[String]) = {
+ val sys = readMetadataList(systemPackagesPath(cuid, pveName))
+ val usr = readMetadataList(userPackagesPath(cuid, pveName))
+ (sys, usr)
+ }
+
+ /**
+ * Creates a new PVE for a CU.
+ *
+ * Behavior:
+ * - If a base PVE exists (PVE_BASE), it clones it for faster setup
+ * - Otherwise, creates a fresh venv and installs dependencies
+ *
+ * Steps:
+ * 1. Remove existing environment (if present)
Review Comment:
Why does this function remove existing environment?? It should only create a
new PVE.
##########
amber/src/main/scala/org/apache/texera/web/resource/pythonvirtualenvironment/PveManager.scala:
##########
@@ -0,0 +1,333 @@
+/*
+ * 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 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._
+
+/**
+ * PveManager is responsible for managing Python Virtual Environments (PVEs)
+ * for each Computing Unit
+ *
+ * It supports:
+ * - Creating and initializing isolated Python environments
+ * - Installing and uninstalling Python packages
+ * - Tracking system vs user-installed packages via metadata files
+ * - Streaming pip output logs back to the caller
+ *
+ * Each PVE is stored under:
+ * /tmp/texera-pve/venvs/{cuid}/{pveName}/
+ *
+ * The structure includes:
+ * - pve/ -> actual virtual environment
+ * - metadata/ -> package tracking (system + user)
+ */
+
+object PveManager {
+
+ private val VenvRoot: Path = Paths.get("/tmp/texera-pve/venvs")
+
+ private def cuidDir(cuid: Int, pvename: String): Path = {
+ val dir = VenvRoot.resolve(cuid.toString).resolve(pvename)
+ Files.createDirectories(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 writeMetadata(path: Path, lines: Seq[String]): Unit = {
+ if (path.getParent != null) {
+ Files.createDirectories(path.getParent)
+ }
+ 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 pipEnv: Map[String, String] =
+ Map(
+ "PYTHONUNBUFFERED" -> "1",
+ "PIP_PROGRESS_BAR" -> "off",
+ "PIP_DISABLE_PIP_VERSION_CHECK" -> "1",
+ "PIP_NO_INPUT" -> "1"
+ )
+
+ def getSystemAndUserPackages(cuid: Int, pveName: String): (Seq[String],
Seq[String]) = {
+ val sys = readMetadataList(systemPackagesPath(cuid, pveName))
+ val usr = readMetadataList(userPackagesPath(cuid, pveName))
+ (sys, usr)
+ }
+
+ /**
+ * Creates a new PVE for a CU.
+ *
+ * Behavior:
+ * - If a base PVE exists (PVE_BASE), it clones it for faster setup
+ * - Otherwise, creates a fresh venv and installs dependencies
+ *
+ * Steps:
+ * 1. Remove existing environment (if present)
+ * 2. Create or copy base environment
+ * 3. Install system + operator dependencies
+ * 4. Record installed packages as system metadata
+ *
+ * Logs progress to the provided queue.
+ */
+ 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
+ val python = pythonBinPath(cuid, pveName).toAbsolutePath.toString
+ val envVars = pipEnv
+
+ val pveBase = sys.env.getOrElse("PVE_BASE", "/opt/pve-base")
+ val basePython = Paths.get(pveBase).resolve("bin").resolve("python")
+ val hasBasePve = Files.exists(basePython)
+
+ val requirementsPath =
+ if (!hasBasePve) Paths.get("amber", "requirements.txt")
+ else Paths.get("/tmp", "requirements.txt")
+
+ val operatorRequirementsPath =
+ if (!hasBasePve) Paths.get("amber", "operator-requirements.txt")
+ else Paths.get("/tmp", "operator-requirements.txt")
+
+ 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")
+ }
+
+ Files.createDirectories(venvDirPath.getParent)
+
+ 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
+ }
+
+ if (!Files.exists(requirementsPath)) {
+ queue.put(s"[PVE][ERR] requirements.txt not found at
${requirementsPath.toAbsolutePath}")
+ return
+ }
+
+ if (!Files.exists(operatorRequirementsPath)) {
+ queue.put(
+ s"[PVE][ERR] operator-requirements.txt not found at
${operatorRequirementsPath.toAbsolutePath}"
+ )
+ return
+ }
+
+ Files.createDirectories(metadataDir(cuid, pveName))
+
+ queue.put(s"[PVE] Installing local base requirements from
${requirementsPath.toAbsolutePath}")
+
+ val installReqCode = Process(
+ Seq(python, "-m", "pip", "install", "-r", requirementsPath.toString),
+ None,
+ envVars.toSeq: _*
+ ).!(
+ ProcessLogger(
+ out => queue.put(s"[pip] $out"),
+ err => queue.put(s"[pip][ERR] $err")
+ )
+ )
+ queue.put(s"[PVE] requirements install finished with exit code
$installReqCode")
+
+ if (installReqCode != 0) {
+ queue.put(s"[PVE][ERR] Failed to install requirements.txt
(exit=$installReqCode)")
+ return
+ }
+
+ queue.put(
+ s"[PVE] Installing local operator requirements from
${operatorRequirementsPath.toAbsolutePath}"
+ )
+
+ val installOperatorReqCode = Process(
+ Seq(python, "-m", "pip", "install", "-r",
operatorRequirementsPath.toString),
+ None,
+ envVars.toSeq: _*
+ ).!(
+ ProcessLogger(
+ out => queue.put(s"[pip] $out"),
+ err => queue.put(s"[pip][ERR] $err")
+ )
+ )
+ queue.put(
+ s"[PVE] operator requirements install finished with exit code
$installOperatorReqCode"
+ )
+
+ if (installOperatorReqCode != 0) {
+ queue.put(
+ s"[PVE][ERR] Failed to install operator-requirements.txt
(exit=$installOperatorReqCode)"
+ )
+ return
+ }
+
+ queue.put("[PVE] Running pip freeze")
+ 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")
+ }
+
+ Files.createDirectories(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
+ )
+ ).!(
+ ProcessLogger(
+ out => queue.put(s"[pve] $out"),
+ err => queue.put(s"[pve][ERR] $err")
+ )
+ )
+ queue.put(s"[pve] rewrite finished with exit code $fixCode")
+
+ Files.createDirectories(metadataDir(cuid, pveName))
+
+ queue.put("[PVE] Base environment copied; skipping system requirements
install.")
+
+ queue.put("[PVE] Running pip freeze")
+ val freezeOutput = Process(Seq(python, "-m", "pip", "freeze"), None,
envVars.toSeq: _*).!!
+ val systemFreezeLines =
freezeOutput.split("\n").map(_.trim).filter(_.nonEmpty).toSeq
+
+ writeMetadata(systemPackagesPath(cuid, pveName), systemFreezeLines)
+ writeMetadata(userPackagesPath(cuid, pveName), Seq.empty)
+
+ queue.put(s"[PVE] Created new environment for cuid=$cuid")
+ }
+
+ def pveExists(cuid: Int, pveName: String): Boolean =
+ Files.exists(pythonBinPath(cuid, pveName)) &&
Files.exists(pipBinPath(cuid, pveName))
+
+ def getEnvironments(cuid: Int): List[String] = {
Review Comment:
Make this method private.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
