ForVic commented on code in PR #59696:
URL: https://github.com/apache/spark/pull/59696#discussion_r4355703077
##########
core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala:
##########
@@ -1687,7 +1704,73 @@ private[history] class FsHistoryProvider(conf:
SparkConf, clock: Clock)
KVUtils.open(newStorePath, metadata, conf, live = false)
}
- private def createInMemoryStore(attempt: AttemptInfoWrapper): KVStore = {
+ /**
+ * Restores a history snapshot and checks that it contains the expected
application attempt, so
+ * an incomplete snapshot falls back to replay instead of producing a store
the UI cannot load.
+ */
+ private def restoreSnapshot(
+ store: KVStore,
+ snapshot: Path,
+ appId: String,
+ attempt: AttemptInfoWrapper): Unit = {
+ snapshotStore.restoreSnapshot(store, snapshot)
+ val restored = try {
+ Some(store.read(classOf[org.apache.spark.status.ApplicationInfoWrapper],
appId).info)
+ } catch {
+ case _: NoSuchElementException => None
+ }
+ if (!restored.exists(_.attempts.exists(_.attemptId ==
attempt.info.attemptId))) {
+ throw new IOException(
+ s"History snapshot $snapshot does not contain app $appId /
${attempt.info.attemptId}")
+ }
+ }
+
+ private def createDiskStoreFromSnapshot(
+ dm: HistoryServerDiskManager,
+ appId: String,
+ attempt: AttemptInfoWrapper,
+ metadata: AppStatusStoreMetadata,
+ manifestPath: Path): KVStore = {
+ val lease = dm.leaseExact(snapshotStore.snapshotSize(manifestPath))
+ val startNs = System.nanoTime()
+ val newStorePath = try {
+ Utils.tryWithResource(KVUtils.open(lease.tmpPath, metadata, conf, live =
false)) { store =>
+ restoreSnapshot(store, manifestPath, appId, attempt)
+ }
+ lease.commit(appId, attempt.info.attemptId)
+ } catch {
+ case e: Exception =>
+ lease.rollback()
+ throw e
+ }
+ val durationMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNs)
+ logInfo(log"Restored history snapshot ${MDC(PATH, manifestPath)} for " +
+ log"${MDC(APP_ID, appId)} into disk store in ${MDC(TOTAL_TIME,
durationMs)} ms")
+
+ KVUtils.open(newStorePath, metadata, conf, live = false)
+ }
+
+ private def createInMemoryStore(appId: String, attempt: AttemptInfoWrapper):
KVStore = {
+ snapshotStore.findSnapshot(appId, attempt.info.attemptId).foreach {
snapshot =>
+ val store = new InMemoryStore()
+ store.setMetadata(new
AppStatusStoreMetadata(AppStatusStore.CURRENT_VERSION))
+ val startNs = System.nanoTime()
+ try {
+ restoreSnapshot(store, snapshot, appId, attempt)
+ val durationMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() -
startNs)
+ logInfo(log"Restored history snapshot ${MDC(PATH, snapshot)} for " +
+ log"${MDC(APP_ID, appId)} into in-memory store in ${MDC(TOTAL_TIME,
durationMs)} ms")
+ return store
Review Comment:
Good catch, fixed in 39137513c45f. Snapshots now record the app's retention
limits, and if the History Server is set to keep more than the snapshot
captured, it replays the event log instead. Added your `retainedJobs=1` vs 1000
case as a test.
##########
core/src/main/scala/org/apache/spark/deploy/history/HistorySnapshotStore.scala:
##########
@@ -0,0 +1,511 @@
+/*
+ * 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.deploy.history
+
+import java.io.{BufferedInputStream, BufferedOutputStream, DataInputStream,
DataOutputStream, IOException}
+import java.net.URLEncoder
+import java.nio.charset.StandardCharsets
+import java.util.{Properties, UUID}
+
+import scala.util.control.NonFatal
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.{FileSystem, FSDataOutputStream, Path}
+
+import org.apache.spark.SparkConf
+import org.apache.spark.deploy.SparkHadoopUtil
+import org.apache.spark.internal.Logging
+import org.apache.spark.internal.LogKeys._
+import org.apache.spark.internal.config.History._
+import org.apache.spark.status._
+import org.apache.spark.status.api.v1
+import org.apache.spark.status.protobuf.KVStoreProtobufSerializer
+import org.apache.spark.util.Utils
+import org.apache.spark.util.kvstore.KVStore
+
+/**
+ * Writes and restores protobuf snapshots of an application's status store
under
+ * `spark.history.snapshot.path`, so the History Server can load a completed
application without
+ * replaying its event log. Snapshot files use the same permissions as event
logs.
+ */
+private[spark] class HistorySnapshotStore(conf: SparkConf, hadoopConf:
Configuration)
+ extends Logging {
+
+ import HistorySnapshotStore._
+
+ def this(conf: SparkConf) = this(conf,
SparkHadoopUtil.get.newConfiguration(conf))
+
+ private val configuredRoot: Option[Path] =
+ if (isEnabled(conf)) conf.get(SNAPSHOT_PATH).map(new Path(_)) else None
+
+ // Resolved lazily so a misconfigured snapshot path does not affect
application startup.
+ private lazy val fs: FileSystem =
configuredRoot.get.getFileSystem(hadoopConf)
+
+ private lazy val rootPath: Option[Path] =
configuredRoot.map(fs.makeQualified)
+
+ private val serializer = new KVStoreProtobufSerializer()
+
+ /**
+ * Writes a versioned history snapshot for an application attempt and
returns the manifest path.
+ *
+ * The snapshot is published under the configured snapshot root using a
fresh manifest and data
+ * directory so readers can safely continue using the previous published
snapshot while a new one
+ * is being written.
+ */
+ def writeSnapshot(store: KVStore, appId: String, attemptId: Option[String]):
Option[Path] = {
+ snapshotDir(appId, attemptId).map { attemptDir =>
+ mkdirs(attemptDir.getParent)
+ mkdirs(attemptDir)
+ val snapshotId = newSnapshotId()
+ val dataDir = new Path(attemptDir, s"$SNAPSHOT_DIR_PREFIX$snapshotId")
+ val manifestPath = new Path(attemptDir,
s"$MANIFEST_FILE_PREFIX$snapshotId.properties")
+
+ try {
+ mkdirs(dataDir)
+ val entries = writeEntityFiles(store, dataDir, appId)
+
+ val manifest = new Properties()
+ manifest.setProperty("schemaVersion", SNAPSHOT_SCHEMA_VERSION.toString)
+ manifest.setProperty("snapshotDir", dataDir.getName)
+ manifest.setProperty("classNames",
entries.map(_.className).mkString(","))
+ entries.foreach { entry =>
+ manifest.setProperty(s"${entry.className}.file", entry.fileName)
+ }
+
+ Utils.tryWithResource(create(manifestPath, overwrite = false)) { out =>
+ manifest.store(out, "spark history snapshot")
+ }
+ // Only older manifests are deleted, so readers always see either the
old snapshot or the
+ // newly published one.
+ cleanupStaleSnapshots(attemptDir, manifestPath)
+ manifestPath
+ } catch {
+ case NonFatal(e) =>
+ deletePathQuietly(manifestPath, recursive = false)
+ deletePathQuietly(dataDir, recursive = true)
+ throw e
+ }
+ }
+ }
+
+ /**
+ * Returns the latest published history snapshot manifest for an application
attempt, if any.
+ *
+ * When multiple manifests exist, the latest versioned manifest is treated
as the active
+ * published snapshot.
+ */
+ def findSnapshot(appId: String, attemptId: Option[String]): Option[Path] = {
+ try {
+ snapshotDir(appId, attemptId).flatMap(latestManifest)
+ } catch {
+ case NonFatal(e) =>
+ logWarning(log"Failed to look up history snapshot for ${MDC(APP_ID,
appId)}.", e)
+ None
+ }
+ }
+
+ /**
+ * Loads a published history snapshot into the provided KVStore.
+ *
+ * The manifest controls which entity files are restored and where the
snapshot data directory is
+ * located relative to the manifest path.
+ */
+ def restoreSnapshot(store: KVStore, manifestPath: Path): Unit = {
+ val manifest = loadManifest(manifestPath)
+ val schemaVersion = manifest.getProperty("schemaVersion", "-1").toInt
+ require(schemaVersion == SNAPSHOT_SCHEMA_VERSION,
+ s"Unsupported history snapshot version $schemaVersion at $manifestPath")
+
+ val dataRoot = snapshotDataRoot(manifestPath, manifest)
+ manifestEntries(manifestPath, manifest).foreach { entry =>
+ loadSnapshotClass(manifestPath, entry.className).foreach { klass =>
+ readEntityFile(store, klass, new Path(dataRoot, entry.fileName))
+ }
+ }
+ }
+
+ /**
+ * Returns the total on-storage size of a published history snapshot,
including its manifest.
+ *
+ * This is used by History Server disk-store allocation so the target local
store can reserve
+ * enough space before restoring a snapshot.
+ */
+ def snapshotSize(manifestPath: Path): Long = {
+ val manifest = loadManifest(manifestPath)
+ val root = snapshotDataRoot(manifestPath, manifest)
+ manifestEntries(manifestPath, manifest)
+ .foldLeft(fs.getFileStatus(manifestPath).getLen) { (total, entry) =>
+ total + fs.getFileStatus(new Path(root, entry.fileName)).getLen
+ }
+ }
+
+ /**
+ * Deletes all snapshots of an application attempt. The History Server calls
this when it cleans
+ * up the attempt's event log, so snapshots do not outlive their event logs.
+ */
+ def deleteSnapshots(appId: String, attemptId: Option[String]): Unit = {
+ Utils.tryLogNonFatalError {
+ snapshotDir(appId, attemptId).foreach { attemptDir =>
+ deletePathQuietly(attemptDir, recursive = true)
+ val appDir = attemptDir.getParent
+ if (fs.exists(appDir) && fs.listStatus(appDir).isEmpty) {
+ fs.delete(appDir, false)
+ }
+ }
+ }
+ }
+
+ /**
+ * Returns the snapshot directory of an application attempt. The IDs come
from event logs, and
+ * Hadoop normalizes `.` and `..` components, so the result is checked to
stay within the root.
+ */
+ private def snapshotDir(appId: String, attemptId: Option[String]):
Option[Path] = {
+ rootPath.map { root =>
+ val dir = new Path(new Path(root, encode(appId)),
encode(attemptId.getOrElse("_default_")))
+ if (dir.getParent == null || dir.getParent.getParent != root) {
+ throw new IllegalArgumentException(
+ s"Snapshot path for app $appId / $attemptId escapes the snapshot
directory")
+ }
+ dir
+ }
+ }
+
+ /** Reads and parses a snapshot manifest file. */
+ private def loadManifest(manifestPath: Path): Properties = {
+ val properties = new Properties()
+ Utils.tryWithResource(fs.open(manifestPath)) { in =>
+ properties.load(in)
+ }
+ properties
+ }
+
+ /** Returns the newest published manifest under a snapshot root, ordered by
manifest name. */
+ private def latestManifest(rootPath: Path): Option[Path] = {
+ if (!fs.exists(rootPath)) {
+ None
+ } else {
+
fs.listStatus(rootPath).map(_.getPath).filter(isVersionedManifest).sortBy(_.getName)
+ .lastOption
+ }
+ }
+
+ /** Removes stale manifests and data only when this writer still owns the
latest publication. */
+ private def cleanupStaleSnapshots(rootPath: Path, keepManifestPath: Path):
Unit = {
+ try {
+ // A concurrent writer may have published a newer manifest after this
write completed.
+ // In that case, avoid deleting anything and let the newer publication
own cleanup.
+ if (latestManifest(rootPath).exists(_ != keepManifestPath)) {
+ logWarning(log"Skipping cleanup for history snapshot root ${MDC(PATH,
rootPath)} " +
+ log"because another manifest became active while publishing " +
+ log"${MDC(PATHS, keepManifestPath)}.")
+ } else {
+ val manifests = fs.listStatus(rootPath).map(_.getPath).toSeq
+ staleManifestsForCleanup(manifests,
keepManifestPath).foreach(deleteSnapshot)
+ }
+ } catch {
+ case NonFatal(e) =>
+ logWarning(log"Failed to clean up stale history snapshots under
${MDC(PATH, rootPath)}.",
+ e)
+ }
+ }
+
+ /** Deletes a manifest and its associated snapshot artifacts, tolerating
partial corruption. */
+ private def deleteSnapshot(manifestPath: Path): Unit = {
+ try {
+ deletePathQuietly(snapshotDataRoot(manifestPath,
loadManifest(manifestPath)),
+ recursive = true)
+ } catch {
+ case NonFatal(e) =>
+ logWarning(log"Failed to load history snapshot manifest at " +
+ log"${MDC(PATH, manifestPath)} for deletion.", e)
+ }
+ deletePathQuietly(manifestPath, recursive = false)
+ }
+
+ private def mkdirs(dir: Path): Unit = {
+ if (!FileSystem.mkdirs(fs, dir,
EventLogFileWriter.LOG_FOLDER_PERMISSIONS)) {
+ throw new IOException(s"Failed to create history snapshot directory
$dir")
+ }
+ }
+
+ private def create(path: Path, overwrite: Boolean): FSDataOutputStream = {
+ val out = fs.create(path, overwrite)
+ try {
+ fs.setPermission(path, EventLogFileWriter.LOG_FILE_PERMISSIONS)
+ } catch {
+ case NonFatal(e) =>
+ Utils.closeQuietly(out)
+ throw e
+ }
+ out
+ }
+
+ private def deletePathQuietly(path: Path, recursive: Boolean): Unit = {
+ Utils.tryLogNonFatalError {
+ if (fs.exists(path)) {
+ fs.delete(path, recursive)
+ }
+ }
+ }
+
+ /** Writes one file per entity type into the snapshot data directory. */
+ private def writeEntityFiles(
+ store: KVStore,
+ dataDir: Path,
+ appId: String): Seq[SnapshotEntry] = {
+ snapshotEntityClasses.zipWithIndex.flatMap { case (klass, idx) =>
+ val fileName = f"$idx%03d-${klass.getSimpleName}.pb"
+ val path = new Path(dataDir, fileName)
+ if (writeEntityFile(store, klass, path, appId) > 0L) {
+ Some(SnapshotEntry(klass.getName, fileName))
+ } else {
+ fs.delete(path, false)
+ None
+ }
+ }
+ }
+
+ /** Writes all records for one entity type into a single snapshot file. */
+ private def writeEntityFile(
+ store: KVStore,
+ klass: Class[_],
+ path: Path,
+ appId: String): Long = {
+ singletonEntity(store, klass, appId)
+ .map { value =>
+ writeValuesFile(path, Iterator.single(redact(value)))
+ }
+ .getOrElse {
+
Utils.tryWithResource(store.view(klass.asInstanceOf[Class[AnyRef]]).closeableIterator())
{
+ it =>
+ writeValuesFile(path,
+
Iterator.continually(it).takeWhile(_.hasNext).map(_.next().asInstanceOf[AnyRef]))
+ }
+ }
+ }
+
+ /** Serializes a stream of KVStore values using length-delimited protobuf
records. */
+ private def writeValuesFile(path: Path, values: Iterator[AnyRef]): Long = {
+ var count = 0L
+ Utils.tryWithResource(
+ new DataOutputStream(new BufferedOutputStream(create(path, overwrite =
true)))) { out =>
+ while (values.hasNext) {
+ val bytes = serializer.serialize(values.next())
+ out.writeInt(bytes.length)
+ out.write(bytes)
+ count += 1
+ }
+ }
+ count
+ }
+
+ /**
+ * Redacts the application environment like the event log does. The live
store keeps it
+ * unredacted, but snapshots are written to shared storage and served by the
History Server.
+ */
+ private def redact(value: AnyRef): AnyRef = value match {
+ case env: ApplicationEnvironmentInfoWrapper =>
+ val info = env.info
+ new ApplicationEnvironmentInfoWrapper(new v1.ApplicationEnvironmentInfo(
+ info.runtime,
+ Utils.redact(conf, info.sparkProperties),
+ Utils.redact(conf, info.hadoopProperties),
+ Utils.redact(conf, info.systemProperties),
+ Utils.redact(conf, info.metricsProperties),
+ info.classpathEntries,
+ info.resourceProfiles))
+ case other => other
+ }
+
+ /** Returns singleton snapshot entities whose keys are not discovered
through a KVStore view. */
+ private def singletonEntity(store: KVStore, klass: Class[_], appId: String):
Option[AnyRef] = {
+ def read(key: Any): Option[AnyRef] = {
+ try {
+ Some(store.read(klass.asInstanceOf[Class[AnyRef]], key))
+ } catch {
+ case _: NoSuchElementException => None
+ }
+ }
+
+ if (klass == classOf[org.apache.spark.status.ApplicationInfoWrapper]) {
+ read(appId)
+ } else if (klass == classOf[ApplicationEnvironmentInfoWrapper]) {
+ read(classOf[ApplicationEnvironmentInfoWrapper].getName)
+ } else if (klass == classOf[AppSummary]) {
+ read(classOf[AppSummary].getName)
+ } else {
+ None
+ }
+ }
+
+ /**
+ * Restores one entity file by replaying its length-delimited protobuf
records into the store.
+ * Record sizes are checked against the remaining file length before
allocating, so a corrupt
+ * size fails with an IOException rather than an OutOfMemoryError.
+ */
+ private def readEntityFile(store: KVStore, klass: Class[_], path: Path):
Unit = {
+ var remaining = fs.getFileStatus(path).getLen
+ Utils.tryWithResource(new DataInputStream(new
BufferedInputStream(fs.open(path)))) { in =>
+ while (remaining > 0) {
+ val size = in.readInt()
+ remaining -= Integer.BYTES
+ if (size < 0 || size > remaining) {
+ throw new IOException(s"Invalid record size $size in snapshot file
$path")
+ }
+ val bytes = new Array[Byte](size)
Review Comment:
Agreed, fixed in 39137513c45f. Added `spark.history.snapshot.maxRecordSize`
(default 256m, max 512m, matching `eventLog.maxLineLength`). Records over the
limit now fail with an `IOException` and fall back to replay instead of OOMing.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
