Skip to content

[SPARK-6688] [core] Always use resolved URIs in EventLoggingListener. #5340

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
vanzin wants to merge 2 commits into from
Closed

[SPARK-6688] [core] Always use resolved URIs in EventLoggingListener. #5340

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
20d2a34
[SPARK-6688] [core] Always use resolved URIs in EventLoggingListener.
Apr 3, 2015
ccfddd9
Resolve at the source.
Apr 3, 2015
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions core/src/main/scala/org/apache/spark/deploy/master/Master.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@
package org.apache.spark.deploy.master

import java.io.FileNotFoundException
import java.net.URLEncoder
import java.net.{URI, URLEncoder}
Copy link
Contributor

@andrewor14 andrewor14 Apr 3, 2015

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this doesn't need to change. I'll fix this when I merge

import java.text.SimpleDateFormat
import java.util.Date

Expand Down Expand Up @@ -755,7 +755,7 @@ private[master] class Master(
}

val eventLogFilePrefix = EventLoggingListener.getLogPath(
eventLogDir, app.id, app.desc.eventLogCodec)
new URI(eventLogDir), app.id, app.desc.eventLogCodec)
Copy link
Contributor

@andrewor14 andrewor14 Apr 3, 2015

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

does this have to be resolveURI as well?

val fs = Utils.getHadoopFileSystem(eventLogDir, hadoopConf)
val inProgressExists = fs.exists(new Path(eventLogFilePrefix +
EventLoggingListener.IN_PROGRESS))
Expand Down
9 changes: 5 additions & 4 deletions core/src/main/scala/org/apache/spark/scheduler/EventLoggingListener.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,7 +47,7 @@ import org.apache.spark.util.{JsonProtocol, Utils}
*/
private[spark] class EventLoggingListener(
appId: String,
logBaseDir: String,
unresolvedLogBaseDir: String,
sparkConf: SparkConf,
hadoopConf: Configuration)
extends SparkListener with Logging {
Expand All @@ -57,11 +57,12 @@ private[spark] class EventLoggingListener(
def this(appId: String, logBaseDir: String, sparkConf: SparkConf) =
this(appId, logBaseDir, sparkConf, SparkHadoopUtil.get.newConfiguration(sparkConf))

private val logBaseDir = Utils.resolveURI(unresolvedLogBaseDir)
private val shouldCompress = sparkConf.getBoolean("spark.eventLog.compress", false)
private val shouldOverwrite = sparkConf.getBoolean("spark.eventLog.overwrite", false)
private val testing = sparkConf.getBoolean("spark.eventLog.testing", false)
private val outputBufferSize = sparkConf.getInt("spark.eventLog.buffer.kb", 100) * 1024
private val fileSystem = Utils.getHadoopFileSystem(new URI(logBaseDir), hadoopConf)
private val fileSystem = Utils.getHadoopFileSystem(logBaseDir, hadoopConf)
private val compressionCodec =
if (shouldCompress) {
Some(CompressionCodec.createCodec(sparkConf))
Expand Down Expand Up @@ -259,13 +260,13 @@ private[spark] object EventLoggingListener extends Logging {
* @return A path which consists of file-system-safe characters.
*/
def getLogPath(
logBaseDir: String,
logBaseDir: URI,
appId: String,
compressionCodecName: Option[String] = None): String = {
val sanitizedAppId = appId.replaceAll("[ :/]", "-").replaceAll("[.${}'\"]", "_").toLowerCase
// e.g. app_123, app_123.lzf
val logName = sanitizedAppId + compressionCodecName.map { "." + _ }.getOrElse("")
Utils.resolveURI(logBaseDir).toString.stripSuffix("/") + "/" + logName
logBaseDir.toString.stripSuffix("/") + "/" + logName
}

/**
Expand Down
2 changes: 1 addition & 1 deletion core/src/test/scala/org/apache/spark/deploy/history/FsHistoryProviderSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,7 @@ class FsHistoryProviderSuite extends FunSuite with BeforeAndAfter with Matchers
inProgress: Boolean,
codec: Option[String] = None): File = {
val ip = if (inProgress) EventLoggingListener.IN_PROGRESS else ""
val logUri = EventLoggingListener.getLogPath(testDir.getAbsolutePath, appId)
val logUri = EventLoggingListener.getLogPath(testDir.toURI, appId)
val logPath = new URI(logUri).getPath + ip
new File(logPath)
}
Expand Down
23 changes: 17 additions & 6 deletions core/src/test/scala/org/apache/spark/scheduler/EventLoggingListenerSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -95,7 +95,7 @@ class EventLoggingListenerSuite extends FunSuite with LocalSparkContext with Bef
}

test("Log overwriting") {
val logUri = EventLoggingListener.getLogPath(testDir.getAbsolutePath, "test")
val logUri = EventLoggingListener.getLogPath(testDir.toURI, "test")
val logPath = new URI(logUri).getPath
// Create file before writing the event log
new FileOutputStream(new File(logPath)).close()
Expand All @@ -107,16 +107,27 @@ class EventLoggingListenerSuite extends FunSuite with LocalSparkContext with Bef

test("Event log name") {
// without compression
assert(s"file:/base-dir/app1" === EventLoggingListener.getLogPath("/base-dir", "app1"))
assert(s"file:/base-dir/app1" === EventLoggingListener.getLogPath(
Utils.resolveURI("/base-dir"), "app1"))
// with compression
assert(s"file:/base-dir/app1.lzf" ===
EventLoggingListener.getLogPath("/base-dir", "app1", Some("lzf")))
EventLoggingListener.getLogPath(Utils.resolveURI("/base-dir"), "app1", Some("lzf")))
// illegal characters in app ID
assert(s"file:/base-dir/a-fine-mind_dollar_bills__1" ===
EventLoggingListener.getLogPath("/base-dir", "a fine:mind$dollar{bills}.1"))
EventLoggingListener.getLogPath(Utils.resolveURI("/base-dir"),
"a fine:mind$dollar{bills}.1"))
// illegal characters in app ID with compression
assert(s"file:/base-dir/a-fine-mind_dollar_bills__1.lz4" ===
EventLoggingListener.getLogPath("/base-dir", "a fine:mind$dollar{bills}.1", Some("lz4")))
EventLoggingListener.getLogPath(Utils.resolveURI("/base-dir"),
"a fine:mind$dollar{bills}.1", Some("lz4")))
}

test("SPARK-6688: logger should always use resolved URIs") {
val conf = getLoggingConf(testDirPath)
.set("spark.hadoop.fs.defaultFS", "unsupported://example.com")
val eventLogger = new EventLoggingListener("test", testDirPath.toUri().getPath(), conf)
eventLogger.start()
eventLogger.stop()
}

/* ----------------- *
Expand Down Expand Up @@ -178,7 +189,7 @@ class EventLoggingListenerSuite extends FunSuite with LocalSparkContext with Bef
assert(sc.eventLogger.isDefined)
val eventLogger = sc.eventLogger.get
val eventLogPath = eventLogger.logPath
val expectedLogDir = testDir.toURI().toString()
val expectedLogDir = testDir.toURI()
assert(eventLogPath === EventLoggingListener.getLogPath(
expectedLogDir, sc.applicationId, compressionCodec.map(CompressionCodec.getShortName)))

Expand Down