You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by pw...@apache.org on 2014/01/06 07:37:44 UTC
[1/4] git commit: Quite akka when remote lifecycle logging is
disabled.
Updated Branches:
refs/heads/master 5b0986a1d -> a2e7e0497
Quite akka when remote lifecycle logging is disabled.
I noticed when connecting to a standalone cluster Spark gives a bunch
of Akka ERROR logs that make it seem like something is failing.
This patch does two things:
1. Akka dead letter logging is turned on/off according to the existing
lifecycle spark property.
2. We explicitly silence akka's EndpointWriter log in log4j. This is necessary
because for some reason that log doesn't pick up on the lifecycle
logging settings. After a few hours of debugging this was the only solution
I found that worked.
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/aaaa6731
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/aaaa6731
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/aaaa6731
Branch: refs/heads/master
Commit: aaaa6731845495743aff4cc9bd64a54b9aa36c27
Parents: d43ad3e
Author: Patrick Wendell <pw...@gmail.com>
Authored: Sun Jan 5 13:57:42 2014 -0800
Committer: Patrick Wendell <pw...@gmail.com>
Committed: Sun Jan 5 15:15:59 2014 -0800
----------------------------------------------------------------------
.../main/scala/org/apache/spark/util/AkkaUtils.scala | 14 ++++++++++++--
1 file changed, 12 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/aaaa6731/core/src/main/scala/org/apache/spark/util/AkkaUtils.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/util/AkkaUtils.scala b/core/src/main/scala/org/apache/spark/util/AkkaUtils.scala
index 362cea5..5729334 100644
--- a/core/src/main/scala/org/apache/spark/util/AkkaUtils.scala
+++ b/core/src/main/scala/org/apache/spark/util/AkkaUtils.scala
@@ -21,6 +21,8 @@ import scala.concurrent.duration.{Duration, FiniteDuration}
import akka.actor.{ActorSystem, ExtendedActorSystem, IndestructibleActorSystem}
import com.typesafe.config.ConfigFactory
+import org.apache.log4j.{Level, Logger}
+
import org.apache.spark.SparkConf
/**
@@ -47,8 +49,13 @@ private[spark] object AkkaUtils {
val akkaTimeout = conf.get("spark.akka.timeout", "100").toInt
val akkaFrameSize = conf.get("spark.akka.frameSize", "10").toInt
- val lifecycleEvents =
- if (conf.get("spark.akka.logLifecycleEvents", "false").toBoolean) "on" else "off"
+ val akkaLogLifecycleEvents = conf.get("spark.akka.logLifecycleEvents", "false").toBoolean
+ val lifecycleEvents = if (akkaLogLifecycleEvents) "on" else "off"
+ if (!akkaLogLifecycleEvents) {
+ Option(Logger.getLogger("akka.remote.EndpointWriter")).map(l => l.setLevel(Level.FATAL))
+ }
+
+ val logAkkaConfig = if (conf.get("spark.akka.logAkkaConfig", "false").toBoolean) "on" else "off"
val akkaHeartBeatPauses = conf.get("spark.akka.heartbeat.pauses", "600").toInt
val akkaFailureDetector =
@@ -73,7 +80,10 @@ private[spark] object AkkaUtils {
|akka.remote.netty.tcp.maximum-frame-size = ${akkaFrameSize}MiB
|akka.remote.netty.tcp.execution-pool-size = $akkaThreads
|akka.actor.default-dispatcher.throughput = $akkaBatchSize
+ |akka.log-config-on-start = $logAkkaConfig
|akka.remote.log-remote-lifecycle-events = $lifecycleEvents
+ |akka.log-dead-letters = $lifecycleEvents
+ |akka.log-dead-letters-during-shutdown = $lifecycleEvents
""".stripMargin)
val actorSystem = if (indestructible) {
[3/4] git commit: Responding to Aaron's review
Posted by pw...@apache.org.
Responding to Aaron's review
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/675d7eb4
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/675d7eb4
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/675d7eb4
Branch: refs/heads/master
Commit: 675d7eb4f064129d275a45df4c5c43f558638422
Parents: 94fdcda
Author: Patrick Wendell <pw...@gmail.com>
Authored: Sun Jan 5 21:23:14 2014 -0800
Committer: Patrick Wendell <pw...@gmail.com>
Committed: Sun Jan 5 21:23:14 2014 -0800
----------------------------------------------------------------------
core/src/main/scala/org/apache/spark/util/AkkaUtils.scala | 2 ++
1 file changed, 2 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/675d7eb4/core/src/main/scala/org/apache/spark/util/AkkaUtils.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/util/AkkaUtils.scala b/core/src/main/scala/org/apache/spark/util/AkkaUtils.scala
index 5729334..7df7e3d 100644
--- a/core/src/main/scala/org/apache/spark/util/AkkaUtils.scala
+++ b/core/src/main/scala/org/apache/spark/util/AkkaUtils.scala
@@ -52,6 +52,8 @@ private[spark] object AkkaUtils {
val akkaLogLifecycleEvents = conf.get("spark.akka.logLifecycleEvents", "false").toBoolean
val lifecycleEvents = if (akkaLogLifecycleEvents) "on" else "off"
if (!akkaLogLifecycleEvents) {
+ // As a workaround for Akka issue #3787, we coerce the "EndpointWriter" log to be silent.
+ // See: https://www.assembla.com/spaces/akka/tickets/3787#/
Option(Logger.getLogger("akka.remote.EndpointWriter")).map(l => l.setLevel(Level.FATAL))
}
[2/4] git commit: Provide logging when attempts to connect to the
master fail.
Posted by pw...@apache.org.
Provide logging when attempts to connect to the master fail.
Without these it's a bit less clear what's going on for the user.
One thing I realize when doing this is that akka itself actually retries
the initial association. So the retry we currently have is redundant with
akka's.
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/94fdcda8
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/94fdcda8
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/94fdcda8
Branch: refs/heads/master
Commit: 94fdcda89638498f127abf3bb5231064182b4945
Parents: aaaa673
Author: Patrick Wendell <pw...@gmail.com>
Authored: Sun Jan 5 15:10:05 2014 -0800
Committer: Patrick Wendell <pw...@gmail.com>
Committed: Sun Jan 5 15:16:01 2014 -0800
----------------------------------------------------------------------
.../scala/org/apache/spark/deploy/client/Client.scala | 12 +++++++++++-
1 file changed, 11 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/94fdcda8/core/src/main/scala/org/apache/spark/deploy/client/Client.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/client/Client.scala b/core/src/main/scala/org/apache/spark/deploy/client/Client.scala
index 9bbd635..481026e 100644
--- a/core/src/main/scala/org/apache/spark/deploy/client/Client.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/client/Client.scala
@@ -24,7 +24,8 @@ import scala.concurrent.duration._
import akka.actor._
import akka.pattern.ask
-import akka.remote.{DisassociatedEvent, RemotingLifecycleEvent}
+import akka.remote.{AssociationErrorEvent, DisassociatedEvent, RemotingLifecycleEvent}
+
import org.apache.spark.{Logging, SparkConf, SparkException}
import org.apache.spark.deploy.{ApplicationDescription, ExecutorState}
import org.apache.spark.deploy.DeployMessages._
@@ -110,6 +111,12 @@ private[spark] class Client(
}
}
+ private def isPossibleMaster(remoteUrl: Address) = {
+ masterUrls.map(s => Master.toAkkaUrl(s))
+ .map(u => AddressFromURIString(u).hostPort)
+ .contains(remoteUrl.hostPort)
+ }
+
override def receive = {
case RegisteredApplication(appId_, masterUrl) =>
appId = appId_
@@ -145,6 +152,9 @@ private[spark] class Client(
logWarning(s"Connection to $address failed; waiting for master to reconnect...")
markDisconnected()
+ case AssociationErrorEvent(cause, _, address, _) if isPossibleMaster(address) =>
+ logWarning(s"Could not connect to $address: $cause")
+
case StopClient =>
markDead()
sender ! true
[4/4] git commit: Merge pull request #333 from
pwendell/logging-silence
Posted by pw...@apache.org.
Merge pull request #333 from pwendell/logging-silence
Quiet ERROR-level Akka Logs
This fixes an issue I've seen where akka logs a bunch of things at ERROR level when connecting to a standalone cluster, even in the normal case. I noticed that even when lifecycle logging was disabled, the netty code inside of akka still logged away via akka's EndpointWriter class. There are also some other log streams that I think are new in akka 2.2.1 that I've disabled.
Finally, I added some better logging to the standalone client. This makes it more clear when a connection failure occurs what is going on. Previously it never explicitly said if a connection attempt had failed.
The commit messages here have some more detail.
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/a2e7e049
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/a2e7e049
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/a2e7e049
Branch: refs/heads/master
Commit: a2e7e0497484554f86bd71e93705eb0422b1512b
Parents: 5b0986a 675d7eb
Author: Patrick Wendell <pw...@gmail.com>
Authored: Sun Jan 5 22:37:36 2014 -0800
Committer: Patrick Wendell <pw...@gmail.com>
Committed: Sun Jan 5 22:37:36 2014 -0800
----------------------------------------------------------------------
.../org/apache/spark/deploy/client/Client.scala | 12 +++++++++++-
.../scala/org/apache/spark/util/AkkaUtils.scala | 16 ++++++++++++++--
2 files changed, 25 insertions(+), 3 deletions(-)
----------------------------------------------------------------------