Copilot commented on code in PR #3793: URL: https://github.com/apache/celeborn/pull/3793#discussion_r3803364532
########## client-spark/spark-3-ui/src/main/scala/org/apache/spark/shuffle/celeborn/CelebornStatusStore.scala: ########## @@ -0,0 +1,67 @@ +/* + * 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.spark.shuffle.celeborn + +import com.fasterxml.jackson.annotation.JsonIgnore +import org.apache.spark.util.kvstore.{KVIndex, KVStore} + +private[celeborn] case class AggregatedTaskInfoUIData( + shuffleWriteBytes: Long, + shuffleWriteTimeMs: Long, + shuffleReadBytes: Long, + shuffleFetchWaitTimeMs: Long, + taskDurationMs: Long) { + + @JsonIgnore + @KVIndex + def id: String = classOf[AggregatedTaskInfoUIData].getName +} + +private[celeborn] class CelebornPropertiesUIData( + val info: Seq[(String, String)]) { + + @JsonIgnore + @KVIndex + def id: String = classOf[CelebornPropertiesUIData].getName +} Review Comment: `CelebornPropertiesUIData` is persisted to Spark's `KVStore` (Jackson-backed). As a non-case class with a constructor parameter, it may not deserialize reliably across Spark components (especially in History Server replay). Make this a `case class` (preferred in Spark UI store models) or add an explicit Jackson creator/props. Also consider avoiding `Seq[(String, String)]` tuples in persisted data (use a small `case class` like `{ name: String, value: String }` or store as `Map[String, String]`) to reduce serialization edge cases. ########## client-spark/spark-3-ui/src/main/scala/org/apache/spark/shuffle/celeborn/CelebornHistoryServerPlugin.scala: ########## @@ -0,0 +1,45 @@ +/* + * 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.spark.shuffle.celeborn + +import org.apache.spark.SparkConf +import org.apache.spark.scheduler.SparkListener +import org.apache.spark.shuffle.celeborn.ui.CelebornUITab +import org.apache.spark.status.{AppHistoryServerPlugin, ElementTrackingStore} +import org.apache.spark.ui.SparkUI +import org.apache.spark.util.kvstore.KVStore + +/** Registered via SPI at META-INF/services/org.apache.spark.status.AppHistoryServerPlugin. */ +class CelebornHistoryServerPlugin extends AppHistoryServerPlugin { + + override def createListeners( + conf: SparkConf, + store: ElementTrackingStore): Seq[SparkListener] = { + Seq(new CelebornListener(store, conf)) + } + + override def setupUI(ui: SparkUI): Unit = { + val kvstore: KVStore = ui.store.store + val statusStore = new CelebornStatusStore(kvstore) + if (statusStore.hasData()) { + new CelebornUITab(statusStore, ui) + } Review Comment: `setupUI` only attaches the tab when `hasData()` reports non-zero shuffle bytes. This can hide the tab in History Server for apps that have Celeborn configs captured (or want to see '0' summary) but no shuffle bytes recorded, which is inconsistent with the live UI path (always attaches the tab when the UI exists). Consider attaching unconditionally in SHS as well, or broadening the condition to include presence of stored Celeborn properties / stored records. ########## client-spark/spark-3-ui/src/main/scala/org/apache/spark/shuffle/celeborn/ui/CelebornShufflePage.scala: ########## @@ -0,0 +1,130 @@ +/* + * 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.spark.shuffle.celeborn.ui + +import scala.xml.Node + +import org.apache.spark.internal.Logging +import org.apache.spark.shuffle.celeborn.ui.SparkServletBridge.HttpServletRequest +import org.apache.spark.ui.{UIUtils, WebUIPage} +import org.apache.spark.util.Utils + +private[celeborn] class CelebornShufflePage(parent: CelebornUITab) + extends WebUIPage("") with Logging { + + private val store = parent.store + + override def render(request: HttpServletRequest): Seq[Node] = { + try { + renderBody(request) + } catch { + case e: Throwable => + logError("Failed to render Celeborn Shuffle page", e) + val errorContent = + <div class="row-fluid"> + <div class="span12"> + <h4> + <strong>Celeborn Shuffle</strong> + </h4> + <div class="alert alert-error"> + <pre>Failed to render the Celeborn page: {e.getMessage}</pre> + </div> + </div> + </div> + UIUtils.headerSparkPage(request, "Celeborn Shuffle", errorContent, parent) + } + } + + private def renderBody(request: HttpServletRequest): Seq[Node] = { + val taskInfo = store.aggregatedTaskInfo() + val properties = store.celebornProperties() + + val writeBytes = taskInfo.shuffleWriteBytes + val readBytes = taskInfo.shuffleReadBytes + val writeMs = taskInfo.shuffleWriteTimeMs + val readMs = taskInfo.shuffleFetchWaitTimeMs + val durationMs = taskInfo.taskDurationMs + + def mbps(bytes: Long, ms: Long): String = + if (ms <= 0) "N/A" else f"${bytes.toDouble / 1000.0 / 1000.0 / (ms.toDouble / 1000.0)}%.2f" + def pct(part: Long, total: Long): String = + if (total <= 0) "N/A" else f"${part.toDouble * 100.0 / total.toDouble}%.1f%%" + + val summary = + <div> + <ul class="list-unstyled"> + <li> + <strong>Shuffle Write: </strong> + { + s"${Utils.bytesToString(writeBytes)} | Time: ${UIUtils.formatDuration( + writeMs)} | Speed: ${mbps(writeBytes, writeMs)} MB/s" + } + </li> + <li> + <strong>Shuffle Read: </strong> + { + s"${Utils.bytesToString(readBytes)} | Time: ${UIUtils.formatDuration( + readMs)} | Speed: ${mbps(readBytes, readMs)} MB/s" + } + </li> + <li> + <strong>Shuffle Duration (write+read) / Task Duration: </strong> + { + s"${pct(writeMs + readMs, durationMs)} (Write ${pct( + writeMs, + durationMs)}, Read ${pct(readMs, durationMs)})" + } + </li> + </ul> + </div> + + val propertiesTable = UIUtils.listingTable( + propertyHeader, + propertyRow, + properties.info, + fixedWidth = true, + headerClasses = headerClasses) + + val content = + <span> + {summary} + <span class="collapse-aggregated-celebornProperties collapse-table" + onClick="collapseTable('collapse-aggregated-celebornProperties', + 'aggregated-celebornProperties')"> + <h4> + <span class="collapse-table-arrow arrow-open"></span> + <a>Celeborn Properties</a> + </h4> + </span> + <div class="aggregated-celebornProperties collapsible-table"> Review Comment: The collapse target identifiers/classes include mixed casing (`celebornProperties`). Spark UI’s built-in collapse-table patterns typically use lowercase/hyphenated identifiers; sticking to that convention reduces the chance of mismatches in CSS/JS selectors and avoids subtle HTML/CSS case-sensitivity surprises. Consider renaming these to lowercase, hyphen-separated tokens. ########## client-spark/spark-3-ui/src/main/scala/org/apache/spark/shuffle/celeborn/CelebornListener.scala: ########## @@ -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. + */ + +package org.apache.spark.shuffle.celeborn + +import java.util.concurrent.atomic.AtomicLong + +import org.apache.spark.SparkConf +import org.apache.spark.internal.Logging +import org.apache.spark.scheduler._ +import org.apache.spark.util.kvstore.KVStore + +/** + * Collects Celeborn shuffle metrics into the Spark KVStore for live UI and + * HistoryServer replay. + */ +private[celeborn] class CelebornListener( + val kvstore: KVStore, + val conf: SparkConf) + extends SparkListener with Logging { + + private val totalWriteBytes = new AtomicLong(0L) + private val totalWriteTimeMs = new AtomicLong(0L) + private val totalReadBytes = new AtomicLong(0L) + private val totalFetchWaitTimeMs = new AtomicLong(0L) + private val totalTaskDurationMs = new AtomicLong(0L) + + private val lastUpdateTimestamp = new AtomicLong(-1L) + private val updateIntervalMillis = 5000L Review Comment: `updateIntervalMillis` is a magic constant. If this listener ends up being used in large apps, the best flush cadence may differ (write amplification vs UI freshness). Consider making this configurable via `SparkConf` (e.g., `spark.celeborn.ui.flushIntervalMs`) with a sensible default. ########## client-spark/spark-3-ui/src/main/scala/org/apache/spark/shuffle/celeborn/CelebornListener.scala: ########## @@ -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. + */ + +package org.apache.spark.shuffle.celeborn + +import java.util.concurrent.atomic.AtomicLong + +import org.apache.spark.SparkConf +import org.apache.spark.internal.Logging +import org.apache.spark.scheduler._ +import org.apache.spark.util.kvstore.KVStore + +/** + * Collects Celeborn shuffle metrics into the Spark KVStore for live UI and + * HistoryServer replay. + */ +private[celeborn] class CelebornListener( + val kvstore: KVStore, + val conf: SparkConf) + extends SparkListener with Logging { + + private val totalWriteBytes = new AtomicLong(0L) + private val totalWriteTimeMs = new AtomicLong(0L) + private val totalReadBytes = new AtomicLong(0L) + private val totalFetchWaitTimeMs = new AtomicLong(0L) + private val totalTaskDurationMs = new AtomicLong(0L) + + private val lastUpdateTimestamp = new AtomicLong(-1L) + private val updateIntervalMillis = 5000L + + def register(sc: org.apache.spark.SparkContext): Unit = { + sc.addSparkListener(this) + logInfo("CelebornListener registered successfully") + } + + override def onTaskEnd(taskEnd: SparkListenerTaskEnd): Unit = { + Option(taskEnd.taskMetrics).foreach { metrics => + totalWriteBytes.addAndGet(metrics.shuffleWriteMetrics.bytesWritten) + // writeTime is in nanoseconds; normalize to ms. + totalWriteTimeMs.addAndGet(metrics.shuffleWriteMetrics.writeTime / 1000000L) + totalReadBytes.addAndGet(metrics.shuffleReadMetrics.totalBytesRead) + totalFetchWaitTimeMs.addAndGet(metrics.shuffleReadMetrics.fetchWaitTime) + totalTaskDurationMs.addAndGet(taskEnd.taskInfo.duration) + } + mayUpdate() + } + + override def onEnvironmentUpdate(environmentUpdate: SparkListenerEnvironmentUpdate): Unit = { + val celebornProps = environmentUpdate.environmentDetails + .getOrElse("Spark Properties", Seq.empty) + .filter { case (k, _) => k.startsWith("spark.celeborn.") } + .sortBy(_._1) + if (celebornProps.nonEmpty) { + kvstore.write(new CelebornPropertiesUIData(celebornProps.toList)) + } + } + + override def onApplicationEnd(applicationEnd: SparkListenerApplicationEnd): Unit = { + mayUpdate(force = true) + logInfo("CelebornListener: application ended, final flush completed") + } + + private def mayUpdate(force: Boolean = false): Unit = { + val now = System.currentTimeMillis() + val last = lastUpdateTimestamp.get() + if (!force && (last != -1L && (now - last) < updateIntervalMillis)) { + return + } + if (lastUpdateTimestamp.compareAndSet(last, now) || force) { + flushAggregations() + } Review Comment: When `force = true`, the code flushes even if `compareAndSet` fails, but it does not update `lastUpdateTimestamp` in that path. This can lead to extra flushes immediately after a forced flush if subsequent events arrive. Prefer updating the timestamp on forced flushes as well (and typically structure the CAS as a loop so timestamp changes are consistent). ########## pom.xml: ########## @@ -1490,6 +1490,7 @@ <module>client-spark/spark-3</module> <module>client-spark/spark-3-columnar-common</module> <module>client-spark/spark-3-columnar-shuffle</module> + <module>client-spark/spark-3-ui</module> Review Comment: The module/artifact name `spark-3-ui` suggests Spark 3-only, but the PR description and build profiles indicate it also targets Spark 4 (via the servlet bridge). Consider renaming to something version-neutral (e.g., `spark-ui` / `spark-ui-plugin`) to avoid confusion for users and packagers (especially since Spark 4 shaded artifacts also depend on `celeborn-client-spark-3-ui_*`). ########## client-spark/spark-3-ui/pom.xml: ########## @@ -0,0 +1,91 @@ +<?xml version="1.0" encoding="UTF-8"?> +<!-- + ~ 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. + --> +<project xmlns="http://maven.apache.org/POM/4.0.0"; xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd";> + <modelVersion>4.0.0</modelVersion> + <parent> + <groupId>org.apache.celeborn</groupId> + <artifactId>celeborn-parent_${scala.binary.version}</artifactId> + <version>${project.version}</version> + <relativePath>../../pom.xml</relativePath> + </parent> + + <artifactId>celeborn-client-spark-3-ui_${scala.binary.version}</artifactId> + <packaging>jar</packaging> + <name>Celeborn Spark UI Plugin</name> + + <dependencies> + <dependency> + <groupId>org.apache.celeborn</groupId> + <artifactId>celeborn-client-spark-3_${scala.binary.version}</artifactId> + <version>${project.version}</version> + </dependency> + <dependency> + <groupId>org.apache.spark</groupId> + <artifactId>spark-core_${scala.binary.version}</artifactId> + <scope>provided</scope> + <exclusions> + <exclusion> + <groupId>org.xerial.snappy</groupId> + <artifactId>snappy-java</artifactId> + </exclusion> + </exclusions> + </dependency> + <dependency> + <groupId>javax.servlet</groupId> + <artifactId>javax.servlet-api</artifactId> + <scope>provided</scope> + </dependency> + <dependency> + <groupId>jakarta.servlet</groupId> + <artifactId>jakarta.servlet-api</artifactId> + <scope>provided</scope> + </dependency> Review Comment: Both `javax.servlet-api` and `jakarta.servlet-api` are declared for the same artifact. Even with `provided` scope, this increases the chance of accidental cross-version imports and complicates dependency resolution in downstream builds. Since the source root is already selected via `${servlet.source.dir}`, consider moving the servlet dependency selection into Spark-version-specific profiles (include only `javax` for Spark 3 builds and only `jakarta` for Spark 4 builds). -- 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]
