diff --git a/core/amber/computing-unit.dockerfile b/core/amber/computing-unit.dockerfile index 96529b22d4f..a5b21f04d48 100644 --- a/core/amber/computing-unit.dockerfile +++ b/core/amber/computing-unit.dockerfile @@ -13,6 +13,8 @@ RUN apt-get update && apt-get install -y \ netcat \ unzip \ python3-pip \ + libpq-dev \ + python3-dev \ && apt-get clean # Install Python dependencies diff --git a/core/amber/src/main/scala/edu/uci/ics/texera/web/resource/ResultResource.scala b/core/amber/src/main/scala/edu/uci/ics/texera/web/resource/ResultResource.scala index 206546e8d56..ffe83824395 100644 --- a/core/amber/src/main/scala/edu/uci/ics/texera/web/resource/ResultResource.scala +++ b/core/amber/src/main/scala/edu/uci/ics/texera/web/resource/ResultResource.scala @@ -5,7 +5,7 @@ import edu.uci.ics.amber.core.virtualidentity.WorkflowIdentity import edu.uci.ics.texera.web.auth.SessionUser import edu.uci.ics.texera.web.model.websocket.request.ResultExportRequest import edu.uci.ics.texera.web.model.websocket.response.ResultExportResponse -import edu.uci.ics.texera.web.service.{ResultExportService, WorkflowService} +import edu.uci.ics.texera.web.service.ResultExportService import io.dropwizard.auth.Auth import javax.ws.rs._ diff --git a/core/amber/src/main/scala/edu/uci/ics/texera/web/service/ExecutionResultService.scala b/core/amber/src/main/scala/edu/uci/ics/texera/web/service/ExecutionResultService.scala index 7d0f6b86e83..5596f236f41 100644 --- a/core/amber/src/main/scala/edu/uci/ics/texera/web/service/ExecutionResultService.scala +++ b/core/amber/src/main/scala/edu/uci/ics/texera/web/service/ExecutionResultService.scala @@ -4,7 +4,7 @@ import akka.actor.Cancellable import com.fasterxml.jackson.annotation.{JsonTypeInfo, JsonTypeName} import com.fasterxml.jackson.databind.node.ObjectNode import com.typesafe.scalalogging.LazyLogging -import edu.uci.ics.amber.core.storage.DocumentFactory.{ICEBERG, MONGODB} +import edu.uci.ics.amber.core.storage.DocumentFactory.ICEBERG import edu.uci.ics.amber.core.storage.VFSResourceType.MATERIALIZED_RESULT import edu.uci.ics.amber.core.storage.model.VirtualDocument import edu.uci.ics.amber.core.storage.{DocumentFactory, StorageConfig, VFSURIFactory} diff --git a/core/amber/webserver.dockerfile b/core/amber/webserver.dockerfile index 76b3ce9f9c2..e0c0d035ff3 100644 --- a/core/amber/webserver.dockerfile +++ b/core/amber/webserver.dockerfile @@ -26,7 +26,7 @@ COPY core/ . RUN rm -rf amber/user-resources/* RUN apt-get update -RUN apt-get install -y netcat unzip python3-pip +RUN apt-get install -y netcat unzip python3-pip libpq-dev python3-dev RUN pip3 install python-lsp-server python-lsp-server[websockets] RUN pip3 install -r amber/requirements.txt diff --git a/core/gui/proxy.config.json b/core/gui/proxy.config.json index c3fadc17c04..979a0f9d5dd 100755 --- a/core/gui/proxy.config.json +++ b/core/gui/proxy.config.json @@ -4,7 +4,6 @@ "secure": false, "changeOrigin": true }, - "/api": { "target": "http://texera.example.com:30080", "secure": false, diff --git a/core/gui/src/app/app.module.ts b/core/gui/src/app/app.module.ts index f62768cf00b..164d9c926f5 100644 --- a/core/gui/src/app/app.module.ts +++ b/core/gui/src/app/app.module.ts @@ -144,6 +144,7 @@ import { SocialLoginModule, SocialAuthServiceConfig, GoogleSigninButtonModule } import { GoogleLoginProvider } from "@abacritt/angularx-social-login"; import { lastValueFrom } from "rxjs"; import { HubSearchResultComponent } from "./hub/component/hub-search-result/hub-search-result.component"; +import { NzProgressModule } from "ng-zorro-antd/progress"; registerLocaleData(en); @@ -295,6 +296,7 @@ registerLocaleData(en); NzDividerModule, SocialLoginModule, GoogleSigninButtonModule, + NzProgressModule, ], providers: [ provideNzI18n(en_US), diff --git a/core/gui/src/app/dashboard/component/dashboard.component.html b/core/gui/src/app/dashboard/component/dashboard.component.html index 5473ef1f408..4dff29a12a9 100644 --- a/core/gui/src/app/dashboard/component/dashboard.component.html +++ b/core/gui/src/app/dashboard/component/dashboard.component.html @@ -152,7 +152,7 @@ *ngIf="!isLogin && googleLogin" type="standard" size="large" - width="200"> + [width]="200"> diff --git a/core/gui/src/app/workspace/component/power-button/computing-unit-selection.component.html b/core/gui/src/app/workspace/component/power-button/computing-unit-selection.component.html index c15d7b95d40..3be76bcc176 100644 --- a/core/gui/src/app/workspace/component/power-button/computing-unit-selection.component.html +++ b/core/gui/src/app/workspace/component/power-button/computing-unit-selection.component.html @@ -1,52 +1,181 @@ - - -
- - - - +
+
+
+ CPU + +
- - - {{ unit.computingUnit.name }} +
+ Memory + +
+
+ + + + +
    +
  • +
    + + + {{ getComputingUnitName(unit.uri) }} + +
    - -
- - + [nzTooltipTitle]="'Terminate ' + getComputingUnitName(unit.uri)" + (click)="terminateComputingUnit(unit.computingUnit.cuid); $event.stopPropagation()">
+ +
  • +
  • +
    + + Computing Unit +
    +
    • + + + - - -
    + + Create Computing Unit + +
    +
    + Select Memory + + + + + +
    + +
    + Select CPU + + + + + +
    +
    +     + + + +     + + +
    +
    +

    CPU

    +

    + {{getCpuValue() | number:'1.2-2' || '0.00'}} + / {{getCpuLimit()}} {{getCpuLimitUnit()}} +

    +
    +
    +

    Memory

    +

    + {{getMemoryValue() | number:'1.2-2' || '0.00'}} + / {{getMemoryLimit()}} {{getMemoryLimitUnit()}} +

    +
    diff --git a/core/gui/src/app/workspace/component/power-button/computing-unit-selection.component.scss b/core/gui/src/app/workspace/component/power-button/computing-unit-selection.component.scss index ca50bc7dce3..00f186b0349 100644 --- a/core/gui/src/app/workspace/component/power-button/computing-unit-selection.component.scss +++ b/core/gui/src/app/workspace/component/power-button/computing-unit-selection.component.scss @@ -1,33 +1,124 @@ -nz-select { - width: 300px; +.computing-units-selection { + display: grid; + grid-template-columns: 1fr; + justify-content: center; + align-items: center; + width: 100%; +} + +.computing-units-selection.metrics-visible { + grid-template-columns: repeat(2, 1fr); + justify-content: center; + align-items: center; + justify-items: end; + width: 100%; + gap: 10px; } -nz-option { - height: 100%; +.computing-units-dropdown { + min-width: 250px; + max-width: 350px; } -.computing-unit-option { +#computing-unit-option.unit-selected { + background-color: #f1eeee; +} + +.create-computing-unit { display: flex; + gap: 10px; + justify-content: start; align-items: center; - justify-content: space-between; - width: 100%; } -.unit-details { +.computing-unit-name { display: flex; align-items: center; - gap: 8px; - flex-grow: 1; + justify-content: center; +} + +.terminate-icon:hover { + opacity: 0.8; } -.terminate-box { - width: 16px; /* Size of the red box */ +.memory-selection, +.cpu-selection { + width: 200px; +} + +.create-compute-unit-container { + display: grid; + grid-template-columns: repeat(2, 1fr); + gap: 10px; + justify-content: start; + align-items: center; +} + +.select-unit { + display: flex; + flex-direction: column; + gap: 10px; + justify-content: center; + align-items: start; +} + +.metrics-container { + display: grid; + align-items: center; + grid-template-rows: repeat(2, 1fr); + gap: 3px; + height: 32px; + width: 100%; + padding: 0; + border: none; +} + +.metric-item { + display: grid; + grid-template-columns: repeat(2, 1fr); + width: 100%; + height: 16px; + align-items: center; +} + +#cpu-progress-bar, +#memory-progress-bar { height: 16px; - background-color: red; - border-radius: 2px; /* Optional, if you want slightly rounded corners */ - cursor: pointer; /* Pointer cursor to indicate it's clickable */ } -.terminate-box:hover { - opacity: 0.8; /* Slight opacity on hover for visual feedback */ +.metric-label { + font-size: 10px; +} + +.resource-metrics { + width: 250px; + display: grid; + grid-template-columns: repeat(2, 1fr); + grid-template-rows: repeat(1, 1fr); + justify-content: start; + align-items: center; + gap: 5px; +} + +.general-metric { + display: flex; + flex-direction: column; + width: 100%; + background-color: #f9fafb; + border-radius: 3px; + padding: 10px; + gap: 3px; +} + +.metric-unit { + font-size: 8px; +} + +.metric-name { + font-size: 10px; + margin: 0; +} + +.metric-value { + margin: 0; } diff --git a/core/gui/src/app/workspace/component/power-button/computing-unit-selection.component.ts b/core/gui/src/app/workspace/component/power-button/computing-unit-selection.component.ts index 68e9c9f81ab..6229d82b4f7 100644 --- a/core/gui/src/app/workspace/component/power-button/computing-unit-selection.component.ts +++ b/core/gui/src/app/workspace/component/power-button/computing-unit-selection.component.ts @@ -23,6 +23,10 @@ export class ComputingUnitSelectionComponent implements OnInit { computingUnits: DashboardWorkflowComputingUnit[] = []; private readonly REFRESH_INTERVAL_MS = 2000; + addComputeUnitModalVisible = false; + selectedMemory: string = "2Gi"; + selectedCpu: string = "2"; + constructor( private computingUnitService: WorkflowComputingUnitManagingService, private notificationService: NotificationService, @@ -75,8 +79,10 @@ export class ComputingUnitSelectionComponent implements OnInit { */ startComputingUnit(): void { const computeUnitName = `Compute for Workflow ${this.workflowId}`; + const computeCPU = this.selectedCpu; + const computeMemory = this.selectedMemory; this.computingUnitService - .createComputingUnit(computeUnitName) + .createComputingUnit(computeUnitName, computeCPU, computeMemory) .pipe(untilDestroyed(this)) .subscribe({ next: (unit: DashboardWorkflowComputingUnit) => { @@ -104,7 +110,7 @@ export class ComputingUnitSelectionComponent implements OnInit { .pipe(untilDestroyed(this)) .subscribe({ next: (res: Response) => { - this.notificationService.success(`Terminated computing unit with URI: ${uri}`); + this.notificationService.success(`Terminated ${this.getComputingUnitName(uri)}`); this.refreshComputingUnits(); }, error: (err: unknown) => this.notificationService.error("Failed to terminate computing unit"), @@ -126,10 +132,223 @@ export class ComputingUnitSelectionComponent implements OnInit { } } + isComputingUnitRunning(): boolean { + return this.selectedComputingUnit != null && this.selectedComputingUnit.status === "Running"; + } + + computeStatus(): string { + if (!this.selectedComputingUnit) { + return "processing"; + } + switch (this.selectedComputingUnit.status) { + case "Running": + return "success"; + case "Pending" || "Terminating": + return "warning"; + default: + return "error"; + } + } + + cannotSelectUnit(unit: DashboardWorkflowComputingUnit): boolean { + return !unit || unit.status !== "Running"; + } + + isSelectedUnit(unit: DashboardWorkflowComputingUnit): boolean { + return unit.uri === this.selectedComputingUnit?.uri; + } + + showAddComputeUnitModalVisible(): void { + this.addComputeUnitModalVisible = true; + } + + handleAddComputeUnitModalOk(): void { + this.startComputingUnit(); + this.addComputeUnitModalVisible = false; + } + + handleAddComputeUnitModalCancel(): void { + this.addComputeUnitModalVisible = false; + } + + /** + * Gets the computing unit name from the units URI + * @param unitURI (i.e. "computing-unit-85.workflow-computing-unit-svc.workflow-computing-unit-pool.svc.cluster.local") + * @return "Computing unit 85" + */ + getComputingUnitName(unitURI: string): string { + const computingUnit = unitURI.split(".")[0]; + return computingUnit + .split("-") + .map((word, index) => (index < 2 ? word.charAt(0).toUpperCase() + word.slice(1) : word)) + .join(" "); + } + + /** + * Parses computing units resource unit + * @param resource (i.e. "12412512n") + * @return associated unit with resource (i.e. "n", "Mi", "Gi", ...) + */ + parseResourceUnit(resource: string): string { + const match = resource.match(/[a-z].*/i); + return match ? match[0] : ""; + } + + /** + * Parses computing units numerical value + * @param resource (i.e. "12412512n") + * @return associated number with resource (i.e. 12412512) + */ + parseResourceNumber(resource: string): number { + const match = resource.match(/[0-9.]*/); + return match ? Number(match[0]) : 0; + } + + /** + * Convert computing cpu unit resource number to a specific unit + * @param from (i.e. "12412512n") + * @param toUnit (i.e. cores) + * @return i.e. 1.2412512 Cores + */ + cpuResourceConversion(from: string, toUnit: string): string { + // CPU conversion constants (base unit: nanocores) + type CpuUnit = "n" | "u" | "m" | ""; + const cpuUnits: Record = { + n: 1, // nanocores + u: 10 ** 3, // microcores + m: 10 ** 6, // millicores + "": 10 ** 9, // cores + }; + + const fromNumber: number = this.parseResourceNumber(from); + const fromUnit: string = this.parseResourceUnit(from); + + if (!(fromUnit in cpuUnits) || !(toUnit in cpuUnits)) { + return ""; + } + return `${fromNumber * (cpuUnits[fromUnit as CpuUnit] / cpuUnits[toUnit as CpuUnit])} ${toUnit}`; + } + + /** + * Convert computing unit memory resource number to a specific unit + * @param from (i.e. "523Mi") + * @param toUnit (i.e. "Gi") + * @return i.e. 0.524 Gi + */ + memoryResourceConversion(from: string, toUnit: string): string { + // Memory conversion constants (base unit: bytes) + type MemoryUnit = "Ki" | "Mi" | "Gi" | ""; + const memoryUnits = { + "": 1, // bytes + Ki: 1024, // KiB + Mi: 1024 ** 2, // MiB + Gi: 1024 ** 3, // GiB + }; + + const fromNumber: number = this.parseResourceNumber(from); + const fromUnit: string = this.parseResourceUnit(from); + + if (!(fromUnit in memoryUnits) || !(toUnit in memoryUnits)) { + return ""; + } + return `${fromNumber * (memoryUnits[fromUnit as MemoryUnit] / memoryUnits[toUnit as MemoryUnit])} ${toUnit}`; + } + + getCurrentComputingUnitCpuUsage(): string { + return this.selectedComputingUnit?.metrics.cpuUsage || ""; + } + + getCurrentComputingUnitMemoryUsage(): string { + return this.selectedComputingUnit?.metrics.memoryUsage || ""; + } + + getCurrentComputingUnitCpuLimit(): string { + return this.selectedComputingUnit?.resourceLimits.cpuLimit || ""; + } + + getCurrentComputingUnitMemoryLimit(): string { + return this.selectedComputingUnit?.resourceLimits.memoryLimit || ""; + } + /** * Get badge color based on the unit's status. */ getBadgeColor(status: string): string { return status === "Running" ? "green" : "yellow"; } + + getCpuLimit(): number { + return this.parseResourceNumber(this.getCurrentComputingUnitCpuLimit()); + } + + getCpuLimitUnit(): string { + let unit = this.parseResourceUnit(this.getCurrentComputingUnitCpuLimit()); + if (!unit) { + return this.getCpuLimit() == 1 ? "Core" : "Cores"; + } + return this.parseResourceUnit(this.getCurrentComputingUnitCpuLimit()); + } + + getMemoryLimit(): number { + return this.parseResourceNumber(this.getCurrentComputingUnitMemoryLimit()); + } + + getMemoryLimitUnit(): string { + return this.parseResourceUnit(this.getCurrentComputingUnitMemoryLimit()); + } + + getCpuValue(): number { + // convert to appropriate unit based on the limit unit + const cpuLimitUnit: string = this.getCpuLimitUnit(); + const convertedValue: string = this.cpuResourceConversion(this.getCurrentComputingUnitCpuUsage(), cpuLimitUnit); + return this.parseResourceNumber(convertedValue); + } + + getMemoryValue(): number { + // convert to appropriate unit based on the limit + const memoryLimitUnit: string = this.getMemoryLimitUnit(); + const convertedValue: string = this.memoryResourceConversion( + this.getCurrentComputingUnitMemoryUsage(), + memoryLimitUnit + ); + return this.parseResourceNumber(convertedValue); + } + + getCpuPercentage(): number { + // handle divison by zero + const cpuLimit = this.getCpuLimit(); + if (cpuLimit <= 0) { + return 0; + } + return (this.getCpuValue() / cpuLimit) * 100; + } + + getCpuStatus(): "success" | "exception" | "active" | "normal" { + const usage = this.getCpuValue(); + const limit = this.getCpuLimit(); + return usage >= limit ? "exception" : "active"; + } + + getMemoryPercentage(): number { + // handle divison by zero + const memoryLimit = this.getMemoryLimit(); + if (memoryLimit <= 0) { + return 0; + } + return (this.getMemoryValue() / memoryLimit) * 100; + } + + getMemoryStatus(): "success" | "exception" | "active" | "normal" { + const usage = this.getMemoryValue(); + const limit = this.getMemoryLimit(); + return usage >= limit ? "exception" : "active"; + } + + getCpuUnit(): string { + return this.parseResourceUnit(this.getCurrentComputingUnitCpuUsage()); + } + + getMemoryUnit(): string { + return this.parseResourceUnit(this.getCurrentComputingUnitMemoryUsage()); + } } diff --git a/core/gui/src/app/workspace/service/workflow-computing-unit/workflow-computing-unit-managing.service.ts b/core/gui/src/app/workspace/service/workflow-computing-unit/workflow-computing-unit-managing.service.ts index 0188813bbf6..d8e0aefae18 100644 --- a/core/gui/src/app/workspace/service/workflow-computing-unit/workflow-computing-unit-managing.service.ts +++ b/core/gui/src/app/workspace/service/workflow-computing-unit/workflow-computing-unit-managing.service.ts @@ -1,10 +1,11 @@ import { Injectable } from "@angular/core"; -import { HttpClient } from "@angular/common/http"; +import { HttpClient, HttpParams } from "@angular/common/http"; import { Observable } from "rxjs"; import { AppSettings } from "../../../common/app-setting"; -import { DashboardWorkflowComputingUnit } from "../../types/workflow-computing-unit"; +import { DashboardWorkflowComputingUnit, WorkflowComputingUnitMetrics } from "../../types/workflow-computing-unit"; export const COMPUTING_UNIT_BASE_URL = "computing-unit"; +export const COMPUTING_UNIT_METRICS_BASE_URL = "resource-metrics"; export const COMPUTING_UNIT_CREATE_URL = `${COMPUTING_UNIT_BASE_URL}/create`; export const COMPUTING_UNIT_TERMINATE_URL = `${COMPUTING_UNIT_BASE_URL}/terminate`; export const COMPUTING_UNIT_LIST_URL = `${COMPUTING_UNIT_BASE_URL}`; @@ -18,11 +19,18 @@ export class WorkflowComputingUnitManagingService { /** * Create a new workflow computing unit (pod). * @param name The name for the computing unit. + * @param cpuLimit The cpu resource limit for the computing unit. + * @param memoryLimit The memory resource limit for the computing unit. * @param unitType * @returns An Observable of the created WorkflowComputingUnit. */ - public createComputingUnit(name: string, unitType: string = "k8s_pod"): Observable { - const body = { name, unitType }; + public createComputingUnit( + name: string, + cpuLimit: string, + memoryLimit: string, + unitType: string = "k8s_pod" + ): Observable { + const body = { name, cpuLimit, memoryLimit, unitType }; return this.http.post( `${AppSettings.getApiEndpoint()}/${COMPUTING_UNIT_CREATE_URL}`, @@ -50,4 +58,15 @@ export class WorkflowComputingUnitManagingService { `${AppSettings.getApiEndpoint()}/${COMPUTING_UNIT_LIST_URL}` ); } + + /** + * Get a computing units resource metrics + * @returns an Observable of WorkflowComputingUnitMetrics + * @param cuid + */ + public getComputingUnitMetrics(cuid: number): Observable { + return this.http.get( + `${AppSettings.getApiEndpoint()}/${COMPUTING_UNIT_BASE_URL}/${cuid}/metrics` + ); + } } diff --git a/core/gui/src/app/workspace/types/workflow-computing-unit.ts b/core/gui/src/app/workspace/types/workflow-computing-unit.ts index 238e8013606..33565b50a27 100644 --- a/core/gui/src/app/workspace/types/workflow-computing-unit.ts +++ b/core/gui/src/app/workspace/types/workflow-computing-unit.ts @@ -6,8 +6,20 @@ export interface WorkflowComputingUnit { terminateTime: number | undefined; } +export interface WorkflowComputingUnitResourceLimit { + cpuLimit: string; + memoryLimit: string; +} + +export interface WorkflowComputingUnitMetrics { + cpuUsage: string; + memoryUsage: string; +} + export interface DashboardWorkflowComputingUnit { computingUnit: WorkflowComputingUnit; uri: string; status: string; + metrics: WorkflowComputingUnitMetrics; + resourceLimits: WorkflowComputingUnitResourceLimit; } diff --git a/core/gui/src/styles.scss b/core/gui/src/styles.scss index 6c2ac14306b..fd9a924c32a 100644 --- a/core/gui/src/styles.scss +++ b/core/gui/src/styles.scss @@ -76,3 +76,21 @@ hr { .annotation-highlight { background-color: #6a5acd; } + +// For compute-unit-selection.html +#cpu-progress-bar .ant-progress-inner, +#memory-progress-bar .ant-progress-inner { + vertical-align: super !important; +} + +#cpu-progress-bar *, +#memory-progress-bar * { + line-height: 16px !important; +} + +#computing-unit-option .ant-menu-title-content { + display: flex !important; + width: 100% !important; + align-items: center !important; + justify-content: space-between !important; +} diff --git a/core/scripts/server.sh b/core/scripts/server.sh index 61201d64e06..282b8cc5689 100755 --- a/core/scripts/server.sh +++ b/core/scripts/server.sh @@ -1,2 +1,3 @@ +#!/bin/bash cd amber target/texera-0.1-SNAPSHOT/bin/texera-web-application \ No newline at end of file diff --git a/core/scripts/texera-helmchart/files/texera_ddl.sql b/core/scripts/texera-helmchart/files/texera_ddl.sql index 3c188e22c52..71dd433aa6a 100644 --- a/core/scripts/texera-helmchart/files/texera_ddl.sql +++ b/core/scripts/texera-helmchart/files/texera_ddl.sql @@ -1,7 +1,6 @@ CREATE SCHEMA IF NOT EXISTS `texera_db`; USE `texera_db`; -DROP TABLE IF EXISTS `workflow_runtime_statistics`; DROP TABLE IF EXISTS `workflow_user_access`; DROP TABLE IF EXISTS `workflow_of_user`; DROP TABLE IF EXISTS `user_config`; @@ -14,6 +13,8 @@ DROP TABLE IF EXISTS `workflow_executions`; DROP TABLE IF EXISTS `dataset`; DROP TABLE IF EXISTS `dataset_user_access`; DROP TABLE IF EXISTS `dataset_version`; +DROP TABLE IF EXISTS operator_executions; +DROP TABLE IF EXISTS operator_runtime_statistics; SET PERSIST time_zone = '+00:00'; -- this line is mandatory SET PERSIST sql_mode=(SELECT REPLACE(@@sql_mode,'ONLY_FULL_GROUP_BY','')); @@ -143,24 +144,6 @@ CREATE TABLE IF NOT EXISTS public_project FOREIGN KEY (`pid`) REFERENCES `project` (`pid`) ON DELETE CASCADE ) ENGINE = INNODB; -CREATE TABLE IF NOT EXISTS workflow_runtime_statistics -( - `workflow_id` INT UNSIGNED NOT NULL, - `execution_id` INT UNSIGNED NOT NULL, - `operator_id` VARCHAR(100) NOT NULL, - `time` TIMESTAMP(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6), - `input_tuple_cnt` INT UNSIGNED NOT NULL DEFAULT 0, - `output_tuple_cnt` INT UNSIGNED NOT NULL DEFAULT 0, - `status` TINYINT NOT NULL DEFAULT 1, - `data_processing_time` BIGINT UNSIGNED NOT NULL DEFAULT 0, - `control_processing_time` BIGINT UNSIGNED NOT NULL DEFAULT 0, - `idle_time` BIGINT UNSIGNED NOT NULL DEFAULT 0, - `num_workers` INT UNSIGNED NOT NULL DEFAULT 0, - PRIMARY KEY (`workflow_id`, `execution_id`, `operator_id`, `time`), - FOREIGN KEY (`workflow_id`) REFERENCES `workflow` (`wid`) ON DELETE CASCADE, - FOREIGN KEY (`execution_id`) REFERENCES `workflow_executions` (`eid`) ON DELETE CASCADE -) ENGINE = INNODB; - CREATE TABLE IF NOT EXISTS dataset ( `did` INT UNSIGNED AUTO_INCREMENT NOT NULL, @@ -259,4 +242,26 @@ CREATE TABLE IF NOT EXISTS workflow_computing_unit `creation_time` TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, `terminate_time` TIMESTAMP DEFAULT NULL, PRIMARY KEY (`cuid`) -) ENGINE = INNODB; \ No newline at end of file +) ENGINE = INNODB; + +CREATE TABLE IF NOT EXISTS operator_executions ( + operator_execution_id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY, + workflow_execution_id INT UNSIGNED NOT NULL, + operator_id VARCHAR(100) NOT NULL, + UNIQUE (workflow_execution_id, operator_id), + FOREIGN KEY (workflow_execution_id) REFERENCES workflow_executions (eid) ON DELETE CASCADE +); + +CREATE TABLE IF NOT EXISTS operator_runtime_statistics ( + operator_execution_id BIGINT UNSIGNED NOT NULL, + time TIMESTAMP(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6), + input_tuple_cnt BIGINT UNSIGNED NOT NULL DEFAULT 0, + output_tuple_cnt BIGINT UNSIGNED NOT NULL DEFAULT 0, + status TINYINT NOT NULL DEFAULT 1, + data_processing_time BIGINT UNSIGNED NOT NULL DEFAULT 0, + control_processing_time BIGINT UNSIGNED NOT NULL DEFAULT 0, + idle_time BIGINT UNSIGNED NOT NULL DEFAULT 0, + num_workers INT UNSIGNED NOT NULL DEFAULT 0, + PRIMARY KEY (operator_execution_id, time), + FOREIGN KEY (operator_execution_id) REFERENCES operator_executions (operator_execution_id) ON DELETE CASCADE +); \ No newline at end of file diff --git a/core/scripts/texera-helmchart/templates/metrics-server.yaml b/core/scripts/texera-helmchart/templates/metrics-server.yaml new file mode 100644 index 00000000000..31ab7e011b3 --- /dev/null +++ b/core/scripts/texera-helmchart/templates/metrics-server.yaml @@ -0,0 +1,202 @@ +apiVersion: v1 +kind: ServiceAccount +metadata: + labels: + k8s-app: metrics-server + name: metrics-server + namespace: kube-system +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: ClusterRole +metadata: + labels: + k8s-app: metrics-server + rbac.authorization.k8s.io/aggregate-to-admin: "true" + rbac.authorization.k8s.io/aggregate-to-edit: "true" + rbac.authorization.k8s.io/aggregate-to-view: "true" + name: system:aggregated-metrics-reader +rules: +- apiGroups: + - metrics.k8s.io + resources: + - pods + - nodes + verbs: + - get + - list + - watch +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: ClusterRole +metadata: + labels: + k8s-app: metrics-server + name: system:metrics-server +rules: +- apiGroups: + - "" + resources: + - nodes/metrics + verbs: + - get +- apiGroups: + - "" + resources: + - pods + - nodes + verbs: + - get + - list + - watch +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: RoleBinding +metadata: + labels: + k8s-app: metrics-server + name: metrics-server-auth-reader + namespace: kube-system +roleRef: + apiGroup: rbac.authorization.k8s.io + kind: Role + name: extension-apiserver-authentication-reader +subjects: +- kind: ServiceAccount + name: metrics-server + namespace: kube-system +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: ClusterRoleBinding +metadata: + labels: + k8s-app: metrics-server + name: metrics-server:system:auth-delegator +roleRef: + apiGroup: rbac.authorization.k8s.io + kind: ClusterRole + name: system:auth-delegator +subjects: +- kind: ServiceAccount + name: metrics-server + namespace: kube-system +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: ClusterRoleBinding +metadata: + labels: + k8s-app: metrics-server + name: system:metrics-server +roleRef: + apiGroup: rbac.authorization.k8s.io + kind: ClusterRole + name: system:metrics-server +subjects: +- kind: ServiceAccount + name: metrics-server + namespace: kube-system +--- +apiVersion: v1 +kind: Service +metadata: + labels: + k8s-app: metrics-server + name: metrics-server + namespace: kube-system +spec: + ports: + - name: https + port: 443 + protocol: TCP + targetPort: https + selector: + k8s-app: metrics-server +--- +apiVersion: apps/v1 +kind: Deployment +metadata: + labels: + k8s-app: metrics-server + name: metrics-server + namespace: kube-system +spec: + selector: + matchLabels: + k8s-app: metrics-server + strategy: + rollingUpdate: + maxUnavailable: 0 + template: + metadata: + labels: + k8s-app: metrics-server + spec: + containers: + - args: + - --cert-dir=/tmp + - --secure-port=10250 + - --kubelet-preferred-address-types=InternalIP,ExternalIP,Hostname + - --kubelet-use-node-status-port + - --metric-resolution=15s + - --kubelet-insecure-tls + image: registry.k8s.io/metrics-server/metrics-server:v0.7.2 + imagePullPolicy: IfNotPresent + livenessProbe: + failureThreshold: 3 + httpGet: + path: /livez + port: https + scheme: HTTPS + periodSeconds: 10 + name: metrics-server + ports: + - containerPort: 10250 + name: https + protocol: TCP + readinessProbe: + failureThreshold: 3 + httpGet: + path: /readyz + port: https + scheme: HTTPS + initialDelaySeconds: 20 + periodSeconds: 10 + resources: + requests: + cpu: 100m + memory: 200Mi + securityContext: + allowPrivilegeEscalation: false + capabilities: + drop: + - ALL + readOnlyRootFilesystem: true + runAsNonRoot: true + runAsUser: 1000 + seccompProfile: + type: RuntimeDefault + volumeMounts: + - mountPath: /tmp + name: tmp-dir + nodeSelector: + kubernetes.io/os: linux + priorityClassName: system-cluster-critical + serviceAccountName: metrics-server + volumes: + - emptyDir: {} + name: tmp-dir +--- +apiVersion: apiregistration.k8s.io/v1 +kind: APIService +metadata: + labels: + k8s-app: metrics-server + name: v1beta1.metrics.k8s.io +spec: + group: metrics.k8s.io + groupPriorityMinimum: 100 + insecureSkipTLSVerify: true + service: + name: metrics-server + namespace: kube-system + version: v1beta1 + versionPriority: 100 diff --git a/core/scripts/texera-helmchart/templates/workflow-computing-unit-manager-service-account.yaml b/core/scripts/texera-helmchart/templates/workflow-computing-unit-manager-service-account.yaml index 4c19642df40..cd14d575c06 100644 --- a/core/scripts/texera-helmchart/templates/workflow-computing-unit-manager-service-account.yaml +++ b/core/scripts/texera-helmchart/templates/workflow-computing-unit-manager-service-account.yaml @@ -14,6 +14,9 @@ rules: - apiGroups: [""] resources: ["pods"] verbs: ["get", "list", "watch", "create", "delete"] + - apiGroups: ["metrics.k8s.io"] # Added metrics permissions + resources: ["pods"] + verbs: ["list", "get"] # Added metrics permissions --- apiVersion: rbac.authorization.k8s.io/v1 diff --git a/core/scripts/texera-helmchart/templates/workflow-computing-units-deployment.yaml b/core/scripts/texera-helmchart/templates/workflow-computing-units-deployment.yaml index 63d1482cc57..90f89c3cc56 100644 --- a/core/scripts/texera-helmchart/templates/workflow-computing-units-deployment.yaml +++ b/core/scripts/texera-helmchart/templates/workflow-computing-units-deployment.yaml @@ -18,6 +18,7 @@ spec: containers: - name: {{ .Values.workflowComputingUnit.name }} image: {{ .Values.workflowComputingUnit.imageName }} + imagePullPolicy: Never ports: - containerPort: {{ .Values.workflowComputingUnitPool.service.port }} env: diff --git a/core/workflow-computing-unit-managing-service/build.sbt b/core/workflow-computing-unit-managing-service/build.sbt index a4924ac3314..e0bdda78c89 100644 --- a/core/workflow-computing-unit-managing-service/build.sbt +++ b/core/workflow-computing-unit-managing-service/build.sbt @@ -21,7 +21,8 @@ libraryDependencies ++= Seq( "mysql" % "mysql-connector-java" % "8.0.33", "com.softwaremill.sttp.client4" %% "core" % "4.0.0-M6", "com.lihaoyi" %% "upickle" % "3.1.0", - "com.typesafe" % "config" % "1.4.2" + "com.typesafe" % "config" % "1.4.2", + "io.fabric8" % "kubernetes-client" % "6.12.1" ) // Compiler Options diff --git a/core/workflow-computing-unit-managing-service/src/main/scala/edu/uci/ics/texera/service/resource/WorkflowComputingUnitManagingResource.scala b/core/workflow-computing-unit-managing-service/src/main/scala/edu/uci/ics/texera/service/resource/WorkflowComputingUnitManagingResource.scala index 79b7b018215..3f492987254 100644 --- a/core/workflow-computing-unit-managing-service/src/main/scala/edu/uci/ics/texera/service/resource/WorkflowComputingUnitManagingResource.scala +++ b/core/workflow-computing-unit-managing-service/src/main/scala/edu/uci/ics/texera/service/resource/WorkflowComputingUnitManagingResource.scala @@ -8,12 +8,15 @@ import edu.uci.ics.texera.dao.jooq.generated.tables.WorkflowComputingUnit.WORKFL import edu.uci.ics.texera.dao.jooq.generated.tables.pojos.WorkflowComputingUnit import edu.uci.ics.texera.service.resource.WorkflowComputingUnitManagingResource.{ DashboardWorkflowComputingUnit, + TerminationResponse, WorkflowComputingUnitCreationParams, + WorkflowComputingUnitMetrics, + WorkflowComputingUnitResourceLimit, WorkflowComputingUnitTerminationParams, - TerminationResponse, context } import edu.uci.ics.texera.service.util.KubernetesClientService +import edu.uci.ics.texera.service.util.KubernetesMetricService.{getPodLimits, getPodMetrics} import jakarta.ws.rs._ import jakarta.ws.rs.core.MediaType import org.jooq.DSLContext @@ -27,14 +30,31 @@ object WorkflowComputingUnitManagingResource { .getInstance(StorageConfig.jdbcUrl, StorageConfig.jdbcUsername, StorageConfig.jdbcPassword) .createDSLContext() - case class WorkflowComputingUnitCreationParams(name: String, unitType: String) + case class WorkflowComputingUnitCreationParams( + name: String, + unitType: String, + cpuLimit: String, + memoryLimit: String + ) case class WorkflowComputingUnitTerminationParams(uri: String, name: String) + case class WorkflowComputingUnitResourceLimit( + cpuLimit: String, + memoryLimit: String + ) + + case class WorkflowComputingUnitMetrics( + cpuUsage: String, + memoryUsage: String + ) + case class DashboardWorkflowComputingUnit( computingUnit: WorkflowComputingUnit, uri: String, - status: String + status: String, + metrics: WorkflowComputingUnitMetrics, + resourceLimits: WorkflowComputingUnitResourceLimit ) case class TerminationResponse(message: String, uri: String) @@ -75,13 +95,15 @@ class WorkflowComputingUnitManagingResource { val insertedUnit = wcDao.fetchOneByCuid(UInteger.valueOf(cuid)) // Create the pod with the generated CUID - val pod = KubernetesClientService.createPod(cuid) + val pod = KubernetesClientService.createPod(cuid, param.cpuLimit, param.memoryLimit) // Return the dashboard response DashboardWorkflowComputingUnit( insertedUnit, KubernetesClientService.generatePodURI(cuid).toString, - pod.getStatus.getPhase + pod.getStatus.getPhase, + getComputingUnitMetric(cuid.toString), + WorkflowComputingUnitResourceLimit(param.cpuLimit, param.memoryLimit) ) } } @@ -112,7 +134,10 @@ class WorkflowComputingUnitManagingResource { DashboardWorkflowComputingUnit( computingUnit = unit, uri = KubernetesClientService.generatePodURI(cuid).toString, - status = if (pod != null && pod.getStatus != null) pod.getStatus.getPhase else "Unknown" + status = + if (pod != null && pod.getStatus != null) pod.getStatus.getPhase else "Unknown", + getComputingUnitMetric(cuid.toString), + getComputingUnitLimits(cuid.toString) ) }) @@ -147,4 +172,36 @@ class WorkflowComputingUnitManagingResource { TerminationResponse(s"Successfully terminated compute unit with URI $podURI", podURI) } + + /** + * Retrieves the CPU and memory metrics for a computing unit identified by its `cuid`. + * + * @param cuid The computing unit ID. + * @return A `WorkflowComputingUnitMetrics` object with CPU and memory usage data. + */ + @GET + @Produces(Array(MediaType.APPLICATION_JSON)) + @Path("/{cuid}/metrics") + def getComputingUnitMetric(@PathParam("cuid") cuid: String): WorkflowComputingUnitMetrics = { + val metrics: Map[String, String] = getPodMetrics(cuid.toInt) + + WorkflowComputingUnitMetrics( + metrics.getOrElse("cpu", ""), + metrics.getOrElse("memory", "") + ) + } + + @GET + @Produces(Array(MediaType.APPLICATION_JSON)) + @Path("/{cuid}/limits") + def getComputingUnitLimits( + @PathParam("cuid") cuid: String + ): WorkflowComputingUnitResourceLimit = { + val podLimits: Map[String, String] = getPodLimits(cuid.toInt) + + WorkflowComputingUnitResourceLimit( + podLimits.getOrElse("cpu", ""), + podLimits.getOrElse("memory", "") + ) + } } diff --git a/core/workflow-computing-unit-managing-service/src/main/scala/edu/uci/ics/texera/service/util/KubernetesClientService.scala b/core/workflow-computing-unit-managing-service/src/main/scala/edu/uci/ics/texera/service/util/KubernetesClientService.scala index 487661369f6..f2b6d07fc9e 100644 --- a/core/workflow-computing-unit-managing-service/src/main/scala/edu/uci/ics/texera/service/util/KubernetesClientService.scala +++ b/core/workflow-computing-unit-managing-service/src/main/scala/edu/uci/ics/texera/service/util/KubernetesClientService.scala @@ -7,6 +7,7 @@ import config.WorkflowComputingUnitManagingServiceConf.{ computeUnitServiceName } import edu.uci.ics.amber.core.storage.StorageConfig +import io.kubernetes.client.custom.Quantity import io.kubernetes.client.openapi.apis.CoreV1Api import io.kubernetes.client.openapi.models._ import io.kubernetes.client.openapi.{ApiClient, Configuration} @@ -120,12 +121,15 @@ object KubernetesClientService { * @param cuid The computing unit ID. * @return The newly created V1Pod object. */ - def createPod(cuid: Int): V1Pod = { + def createPod(cuid: Int, cpuLimit: String, memoryLimit: String): V1Pod = { val podName = generatePodName(cuid) if (getPodFromLabel(poolNamespace, s"name=$podName") != null) { throw new Exception(s"Pod with cuid $cuid already exists") } + val cpu: Quantity = new Quantity(cpuLimit) + val memory: Quantity = new Quantity(memoryLimit) + val pod: V1Pod = new V1Pod() .apiVersion("v1") .kind("Pod") @@ -160,6 +164,17 @@ object KubernetesClientService { new V1EnvVar().name("JDBC_PASSWORD").value(StorageConfig.jdbcPassword) ) ) + .resources( + new V1ResourceRequirements() + .limits( + util.Map.of( + "cpu", + cpu, + "memory", + memory + ) + ) // may want to add requests as well to make efficient use of the CPU resources + ) ) ) .hostname(podName) diff --git a/core/workflow-computing-unit-managing-service/src/main/scala/edu/uci/ics/texera/service/util/KubernetesMetricService.scala b/core/workflow-computing-unit-managing-service/src/main/scala/edu/uci/ics/texera/service/util/KubernetesMetricService.scala new file mode 100644 index 00000000000..ec49ed4ad45 --- /dev/null +++ b/core/workflow-computing-unit-managing-service/src/main/scala/edu/uci/ics/texera/service/util/KubernetesMetricService.scala @@ -0,0 +1,75 @@ +package edu.uci.ics.texera.service.util + +import config.WorkflowComputingUnitManagingServiceConf +import io.fabric8.kubernetes.api.model.metrics.v1beta1.PodMetricsList +import io.fabric8.kubernetes.api.model.PodList +import io.fabric8.kubernetes.client.{KubernetesClient, KubernetesClientBuilder} + +import scala.jdk.CollectionConverters._ + +object KubernetesMetricService { + + // Initialize the Kubernetes client + val client: KubernetesClient = new KubernetesClientBuilder().build() + + private val namespace = WorkflowComputingUnitManagingServiceConf.computeUnitPoolNamespace + + /** + * Retrieves the pod metric for a given ID in the specified namespace. + * + * @param cuid The computing unit id of the pod + * @return The Pod metrics for a given name in a specified namespace. + */ + def getPodMetrics(cuid: Int): Map[String, String] = { + val podMetricsList: PodMetricsList = client.top().pods().metrics(namespace) + val targetPodName = KubernetesClientService.generatePodName(cuid) + + podMetricsList.getItems.asScala + .collectFirst { + case podMetrics if podMetrics.getMetadata.getName == targetPodName => + podMetrics.getContainers.asScala + .collectFirst { + case container => + container.getUsage.asScala.collect { + case (metric, value) => + println(s"Metric - $metric: ${value}") + // CPU is in nanocores and Memory is in Kibibyte + metric -> value.toString + }.toMap + } + .getOrElse(Map.empty[String, String]) + } + .getOrElse { + println(s"No metrics found for pod: $targetPodName in namespace: $namespace") + Map.empty[String, String] + } + } + + /** + * Retrieves the pod limits for a given ID in the specified namespace. + * + * @param cuid The computing unit id of the pod + * @return The Pod limits for a given name in a specified namespace. + */ + def getPodLimits(cuid: Int): Map[String, String] = { + val podList: PodList = client.pods().inNamespace(namespace).list() + val targetPodName = KubernetesClientService.generatePodName(cuid) + + val pod = podList.getItems.asScala.find(pod => { + pod.getMetadata.getName.equals(targetPodName) + }) + + val limits: Map[String, String] = pod + .flatMap { pod => + pod.getSpec.getContainers.asScala.headOption.map { container => + container.getResources.getLimits.asScala.map { + case (key, value) => + key -> value.toString + }.toMap + } + } + .getOrElse(Map.empty[String, String]) + println(limits.toString()) + limits + } +} diff --git a/core/workflow-core/src/main/scala/edu/uci/ics/amber/core/storage/StorageConfig.scala b/core/workflow-core/src/main/scala/edu/uci/ics/amber/core/storage/StorageConfig.scala index b3d6335b246..f95264611a2 100644 --- a/core/workflow-core/src/main/scala/edu/uci/ics/amber/core/storage/StorageConfig.scala +++ b/core/workflow-core/src/main/scala/edu/uci/ics/amber/core/storage/StorageConfig.scala @@ -49,4 +49,4 @@ object StorageConfig { // File storage configurations val fileStorageDirectoryPath: Path = corePath.resolve("amber").resolve("user-resources").resolve("workflow-results") -} \ No newline at end of file +} diff --git a/core/workflow-core/src/main/scala/edu/uci/ics/amber/core/storage/result/iceberg/IcebergDocument.scala b/core/workflow-core/src/main/scala/edu/uci/ics/amber/core/storage/result/iceberg/IcebergDocument.scala index 862c953b2eb..88530aa4c8c 100644 --- a/core/workflow-core/src/main/scala/edu/uci/ics/amber/core/storage/result/iceberg/IcebergDocument.scala +++ b/core/workflow-core/src/main/scala/edu/uci/ics/amber/core/storage/result/iceberg/IcebergDocument.scala @@ -12,7 +12,6 @@ import org.apache.iceberg.types.{Conversions, Types} import java.net.URI import java.util.concurrent.locks.{ReentrantLock, ReentrantReadWriteLock} -import scala.collection.mutable import scala.jdk.CollectionConverters._ import java.nio.ByteBuffer import java.time.{Instant, LocalDate, ZoneOffset} diff --git a/core/workflow-core/src/test/scala/edu/uci/ics/amber/storage/result/iceberg/IcebergTableStatsSpec.scala b/core/workflow-core/src/test/scala/edu/uci/ics/amber/storage/result/iceberg/IcebergTableStatsSpec.scala index 2aca3408c0e..08d618309fe 100644 --- a/core/workflow-core/src/test/scala/edu/uci/ics/amber/storage/result/iceberg/IcebergTableStatsSpec.scala +++ b/core/workflow-core/src/test/scala/edu/uci/ics/amber/storage/result/iceberg/IcebergTableStatsSpec.scala @@ -18,10 +18,9 @@ import org.scalatest.{BeforeAndAfterAll, Suite} import java.net.URI import java.sql.Timestamp -import java.time.{Instant, LocalDate, ZoneId, ZoneOffset} +import java.time.{LocalDate, ZoneId} import java.time.format.DateTimeFormatter import java.util.UUID -import scala.jdk.CollectionConverters._ class IcebergTableStatsSpec extends AnyFlatSpec with BeforeAndAfterAll with Suite { diff --git a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/visualization/networkGraph/NetworkGraphOpDesc.scala b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/visualization/networkGraph/NetworkGraphOpDesc.scala index 4f57d167d53..5e9682bec19 100644 --- a/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/visualization/networkGraph/NetworkGraphOpDesc.scala +++ b/core/workflow-operator/src/main/scala/edu/uci/ics/amber/operator/visualization/networkGraph/NetworkGraphOpDesc.scala @@ -1,7 +1,7 @@ package edu.uci.ics.amber.operator.visualization.networkGraph import com.fasterxml.jackson.annotation.{JsonProperty, JsonPropertyDescription} -import com.kjetland.jackson.jsonSchema.annotations.{JsonSchemaInject, JsonSchemaTitle} +import com.kjetland.jackson.jsonSchema.annotations.JsonSchemaTitle import edu.uci.ics.amber.core.tuple.{AttributeType, Schema} import edu.uci.ics.amber.core.workflow.OutputPort.OutputMode import edu.uci.ics.amber.core.workflow.{InputPort, OutputPort, PortIdentity}