ForVic commented on code in PR #59696: URL: https://github.com/apache/spark/pull/59696#discussion_r4345546984
########## 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: Good catch, thanks. The snapshot dir is now validated to be exactly `<root>/<app>/<attempt>` (same approach as `HistoryServerDiskManager.appStorePath`), so `.`/`..` IDs are rejected before any read or delete. Added a test with both cases. Fixed in e7eb0e1093fa. ########## 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: Agreed, this shouldn't be deferred. Snapshot dirs/files now use `LOG_FOLDER_PERMISSIONS`/`LOG_FILE_PERMISSIONS`, same as the event log writer. Added a test. Fixed in e7eb0e1093fa. ########## 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: Thanks. Both restore paths now verify the expected app/attempt is present before committing or returning the store, so this falls back to replay. Extended the corrupt-snapshot test to cover an empty entity file. Fixed in e7eb0e1093fa. ########## core/src/main/scala/org/apache/spark/status/AppStatusListener.scala: ########## @@ -103,12 +108,18 @@ private[spark] class AppStatusListener( } kvstore.onFlush { - if (!live) { + if (!live || historySnapshotStore.isDefined) { val now = System.nanoTime() flush(update(_, now)) } } + historySnapshotStore.foreach { snapshotStore => + kvstore.onClose { + writeHistorySnapshot(snapshotStore) Review Comment: Fixed. `SQLAppStatusListener` now flushes live executions on close when snapshots are enabled; flush triggers run before the close trigger that writes the snapshot. Added a test. Fixed in e7eb0e1093fa. ########## 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) + } 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 { + snapshotStore.restoreSnapshot(store, snapshot) + 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 find. When snapshots are enabled, `SharedState` now installs `StreamingQueryStatusListener` regardless of `spark.ui.enabled` / `spark.sql.streaming.ui.enabled`, matching what replay reconstructs. Added a test. Fixed in e7eb0e1093fa. -- 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]
