sunchao commented on code in PR #59696:
URL: https://github.com/apache/spark/pull/59696#discussion_r4353737328
##########
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:
[P2] Check retention settings before preferring the driver snapshot
The snapshot contains the driver's already-pruned status store, but this
return bypasses replay without considering the History Server's retention
settings (the disk path does the same). For example, with driver
`spark.ui.retainedJobs=1` and History Server `spark.ui.retainedJobs=1000`,
three completed jobs produce one restored job instead of the three that replay
returns. A focused write/restore-versus-replay control reproduced exactly that
difference. Increasing retention and restarting the server therefore no longer
retrieves the additional history promised in `docs/monitoring.md`, although it
remains in the event log. Please record/check the capture limits and fall back
to replay when the snapshot cannot satisfy the server's requested retention.
##########
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:
[P2] Bound record allocation independently of the entity-file length
Checking against `remaining` only protects small files. If an entity file
has at least `Int.MaxValue` bytes remaining, a corrupt record length of
`Int.MaxValue` passes this guard and reaches `new Array[Byte](size)`, exceeding
the JVM array-size limit. The resulting `OutOfMemoryError` escapes both
provider `Exception` handlers, so the snapshot does not fall back to the valid
event log. Smaller oversized lengths can likewise exceed the available heap.
Existing event-log replay already has a configurable per-line limit with a hard
cap; snapshot loading needs an independent supported per-record bound before
allocating, with an `IOException` for oversize records. This finding is from
the allocation and exception paths; I did not run a memory-exhaustion test.
##########
core/src/main/scala/org/apache/spark/deploy/history/HistorySnapshotStore.scala:
##########
@@ -0,0 +1,483 @@
+/*
+ * 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, 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.
+ */
+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 =>
+ fs.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 {
+ fs.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(fs.create(manifestPath, 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)
+ }
+ }
+ }
+ }
+
+ private def snapshotDir(appId: String, attemptId: Option[String]):
Option[Path] = {
+ rootPath.map { root =>
+ new Path(new Path(root, encode(appId)),
encode(attemptId.getOrElse("_default_")))
Review Comment:
Verified fixed in e7eb0e10. The current normalized-path check rejects the
dot-component cases; an isolated current-source probe preserved both the
unrelated marker and the other application snapshot. The new current-head CI
regression also passes. Resolving.
##########
core/src/main/scala/org/apache/spark/deploy/history/HistorySnapshotStore.scala:
##########
@@ -0,0 +1,483 @@
+/*
+ * 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, 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.
+ */
+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 =>
+ fs.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 {
+ fs.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(fs.create(manifestPath, 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)
+ }
+ }
+ }
+ }
+
+ private def snapshotDir(appId: String, attemptId: Option[String]):
Option[Path] = {
+ rootPath.map { root =>
+ new Path(new Path(root, encode(appId)),
encode(attemptId.getOrElse("_default_")))
+ }
+ }
+
+ /** 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 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(fs.create(path, true)))) {
Review Comment:
Verified fixed in e7eb0e10. Current-source filesystem checks produced 0770
for the owned directories and 0660 for the manifest/entity files, and the new
permission regression passed in current-head CI. Resolving.
##########
core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala:
##########
@@ -1687,7 +1704,52 @@ private[history] class FsHistoryProvider(conf:
SparkConf, clock: Clock)
KVUtils.open(newStorePath, metadata, conf, live = false)
}
- private def createInMemoryStore(attempt: AttemptInfoWrapper): KVStore = {
+ 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 =>
+ snapshotStore.restoreSnapshot(store, manifestPath)
+ }
+ lease.commit(appId, attempt.info.attemptId)
Review Comment:
Verified fixed in e7eb0e10. Both restore paths validate the expected
app/attempt inside the fallback boundary, before disk commit or memory-store
return. Current-head CI passes the empty and partially truncated entity cases
with memory and disk stores on both backends. Resolving.
--
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]
