diff options
Diffstat (limited to 'core/src/main/scala')
9 files changed, 103 insertions, 11 deletions
diff --git a/core/src/main/scala/spark/deploy/master/Master.scala b/core/src/main/scala/spark/deploy/master/Master.scala index 4a4d9908a0..f4a74830c6 100644 --- a/core/src/main/scala/spark/deploy/master/Master.scala +++ b/core/src/main/scala/spark/deploy/master/Master.scala @@ -65,6 +65,15 @@ private[spark] class Master(host: String, port: Int, webUiPort: Int) extends Act val applicationMetricsSystem = MetricsSystem.createMetricsSystem("applications") val masterSource = new MasterSource(this) + // Add default MetricsServlet handler to web ui + masterMetricsSystem.metricsServlet foreach { m => + webUi.handlers = m.getHandlers ++ webUi.handlers + } + + applicationMetricsSystem.metricsServlet foreach { m => + webUi.handlers = m.getHandlers ++ webUi.handlers + } + val masterPublicAddress = { val envVar = System.getenv("SPARK_PUBLIC_DNS") if (envVar != null) envVar else host diff --git a/core/src/main/scala/spark/deploy/master/ui/MasterWebUI.scala b/core/src/main/scala/spark/deploy/master/ui/MasterWebUI.scala index 31bdb7854e..d9503663f4 100644 --- a/core/src/main/scala/spark/deploy/master/ui/MasterWebUI.scala +++ b/core/src/main/scala/spark/deploy/master/ui/MasterWebUI.scala @@ -57,7 +57,7 @@ class MasterWebUI(val master: ActorRef, requestedPort: Int) extends Logging { } } - val handlers = Array[(String, Handler)]( + var handlers = Array[(String, Handler)]( ("/static", createStaticHandler(MasterWebUI.STATIC_RESOURCE_DIR)), ("/app/json", (request: HttpServletRequest) => applicationPage.renderJson(request)), ("/app", (request: HttpServletRequest) => applicationPage.render(request)), diff --git a/core/src/main/scala/spark/deploy/worker/Worker.scala b/core/src/main/scala/spark/deploy/worker/Worker.scala index 0e46fa281e..92f8cbc610 100644 --- a/core/src/main/scala/spark/deploy/worker/Worker.scala +++ b/core/src/main/scala/spark/deploy/worker/Worker.scala @@ -101,6 +101,12 @@ private[spark] class Worker( logInfo("Spark home: " + sparkHome) createWorkDir() webUi = new WorkerWebUI(this, workDir, Some(webUiPort)) + + // Add default MetricsServlet handlers to webUi + metricsSystem.metricsServlet foreach { m => + webUi.handlers = m.getHandlers ++ webUi.handlers + } + webUi.start() connectToMaster() diff --git a/core/src/main/scala/spark/deploy/worker/ui/WorkerWebUI.scala b/core/src/main/scala/spark/deploy/worker/ui/WorkerWebUI.scala index 742e0a5fb6..d345cbecac 100644 --- a/core/src/main/scala/spark/deploy/worker/ui/WorkerWebUI.scala +++ b/core/src/main/scala/spark/deploy/worker/ui/WorkerWebUI.scala @@ -49,7 +49,7 @@ class WorkerWebUI(val worker: Worker, val workDir: File, requestedPort: Option[I val indexPage = new IndexPage(this) - val handlers = Array[(String, Handler)]( + var handlers = Array[(String, Handler)]( ("/static", createStaticHandler(WorkerWebUI.STATIC_RESOURCE_DIR)), ("/log", (request: HttpServletRequest) => log(request)), ("/logPage", (request: HttpServletRequest) => logPage(request)), diff --git a/core/src/main/scala/spark/metrics/MetricsConfig.scala b/core/src/main/scala/spark/metrics/MetricsConfig.scala index 3e32e9c82f..d10dc45395 100644 --- a/core/src/main/scala/spark/metrics/MetricsConfig.scala +++ b/core/src/main/scala/spark/metrics/MetricsConfig.scala @@ -36,7 +36,9 @@ private[spark] class MetricsConfig(val configFile: Option[String]) extends Loggi var propertyCategories: mutable.HashMap[String, Properties] = null private def setDefaultProperties(prop: Properties) { - // empty function, any default property can be set here + prop.setProperty("*.sink.servlet.class", "spark.metrics.sink.MetricsServlet") + prop.setProperty("master.sink.servlet.uri", "/metrics/master") + prop.setProperty("applications.sink.servlet.uri", "/metrics/applications") } def initialize() { diff --git a/core/src/main/scala/spark/metrics/MetricsSystem.scala b/core/src/main/scala/spark/metrics/MetricsSystem.scala index 1dacafa135..ae1f853691 100644 --- a/core/src/main/scala/spark/metrics/MetricsSystem.scala +++ b/core/src/main/scala/spark/metrics/MetricsSystem.scala @@ -25,7 +25,7 @@ import java.util.concurrent.TimeUnit import scala.collection.mutable import spark.Logging -import spark.metrics.sink.Sink +import spark.metrics.sink.{MetricsServlet, Sink} import spark.metrics.source.Source /** @@ -35,7 +35,7 @@ import spark.metrics.source.Source * "instance" specify "who" (the role) use metrics system. In spark there are several roles * like master, worker, executor, client driver, these roles will create metrics system * for monitoring. So instance represents these roles. Currently in Spark, several instances - * have already implemented: master, worker, executor, driver. + * have already implemented: master, worker, executor, driver, applications. * * "source" specify "where" (source) to collect metrics data. In metrics system, there exists * two kinds of source: @@ -51,8 +51,8 @@ import spark.metrics.source.Source * Metrics configuration format is like below: * [instance].[sink|source].[name].[options] = xxxx * - * [instance] can be "master", "worker", "executor", "driver", which means only the specified - * instance has this property. + * [instance] can be "master", "worker", "executor", "driver", "applications" which means only + * the specified instance has this property. * wild card "*" can be used to replace instance name, which means all the instances will have * this property. * @@ -72,6 +72,9 @@ private[spark] class MetricsSystem private (val instance: String) extends Loggin val sources = new mutable.ArrayBuffer[Source] val registry = new MetricRegistry() + // Treat MetricsServlet as a special sink as it should be exposed to add handlers to web ui + var metricsServlet: Option[MetricsServlet] = None + metricsConfig.initialize() registerSources() registerSinks() @@ -126,7 +129,11 @@ private[spark] class MetricsSystem private (val instance: String) extends Loggin val sink = Class.forName(classPath) .getConstructor(classOf[Properties], classOf[MetricRegistry]) .newInstance(kv._2, registry) - sinks += sink.asInstanceOf[Sink] + if (kv._1 =="servlet") { + metricsServlet = Some(sink.asInstanceOf[MetricsServlet]) + } else { + sinks += sink.asInstanceOf[Sink] + } } catch { case e: Exception => logError("Sink class " + classPath + " cannot be instantialized", e) } diff --git a/core/src/main/scala/spark/metrics/sink/MetricsServlet.scala b/core/src/main/scala/spark/metrics/sink/MetricsServlet.scala new file mode 100644 index 0000000000..39ede9b2df --- /dev/null +++ b/core/src/main/scala/spark/metrics/sink/MetricsServlet.scala @@ -0,0 +1,61 @@ +/* + * 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 spark.metrics.sink + +import com.codahale.metrics.MetricRegistry +import com.codahale.metrics.json.MetricsModule + +import com.fasterxml.jackson.databind.ObjectMapper + +import java.util.Properties +import java.util.concurrent.TimeUnit +import javax.servlet.http.HttpServletRequest + +import org.eclipse.jetty.server.Handler + +import spark.ui.JettyUtils + +class MetricsServlet(val property: Properties, val registry: MetricRegistry) extends Sink { + val SERVLET_KEY_URI = "uri" + val SERVLET_KEY_SAMPLE = "sample" + + val SERVLET_DEFAULT_URI = "/metrics" + val SERVLET_DEFAULT_SAMPLE = false + + val servletURI = property.getProperty(SERVLET_KEY_URI, SERVLET_DEFAULT_URI) + + val servletShowSample = Option(property.getProperty(SERVLET_KEY_SAMPLE)) match { + case Some(s) => s.toBoolean + case None => SERVLET_DEFAULT_SAMPLE + } + + val mapper = new ObjectMapper().registerModule( + new MetricsModule(TimeUnit.SECONDS, TimeUnit.MILLISECONDS, servletShowSample)) + + def getHandlers = Array[(String, Handler)]( + (servletURI, JettyUtils.createHandler(request => getMetricsSnapshot(request), "text/json")) + ) + + def getMetricsSnapshot(request: HttpServletRequest): String = { + mapper.writeValueAsString(registry) + } + + override def start() { } + + override def stop() { } +} diff --git a/core/src/main/scala/spark/ui/JettyUtils.scala b/core/src/main/scala/spark/ui/JettyUtils.scala index ca6088ad93..1cc85124d3 100644 --- a/core/src/main/scala/spark/ui/JettyUtils.scala +++ b/core/src/main/scala/spark/ui/JettyUtils.scala @@ -48,7 +48,7 @@ private[spark] object JettyUtils extends Logging { implicit def textResponderToHandler(responder: Responder[String]): Handler = createHandler(responder, "text/plain") - private def createHandler[T <% AnyRef](responder: Responder[T], contentType: String, + def createHandler[T <% AnyRef](responder: Responder[T], contentType: String, extractFn: T => String = (in: Any) => in.toString): Handler = { new AbstractHandler { def handle(target: String, diff --git a/core/src/main/scala/spark/ui/SparkUI.scala b/core/src/main/scala/spark/ui/SparkUI.scala index 7599f82a94..7e8a41c72e 100644 --- a/core/src/main/scala/spark/ui/SparkUI.scala +++ b/core/src/main/scala/spark/ui/SparkUI.scala @@ -21,7 +21,7 @@ import javax.servlet.http.HttpServletRequest import org.eclipse.jetty.server.{Handler, Server} -import spark.{Logging, SparkContext, Utils} +import spark.{Logging, SparkContext, SparkEnv, Utils} import spark.ui.env.EnvironmentUI import spark.ui.exec.ExecutorsUI import spark.ui.storage.BlockManagerUI @@ -43,8 +43,15 @@ private[spark] class SparkUI(sc: SparkContext) extends Logging { val jobs = new JobProgressUI(sc) val env = new EnvironmentUI(sc) val exec = new ExecutorsUI(sc) + + // Add MetricsServlet handlers by default + val metricsServletHandlers = SparkEnv.get.metricsSystem.metricsServlet match { + case Some(s) => s.getHandlers + case None => Array[(String, Handler)]() + } + val allHandlers = storage.getHandlers ++ jobs.getHandlers ++ env.getHandlers ++ - exec.getHandlers ++ handlers + exec.getHandlers ++ metricsServletHandlers ++ handlers /** Bind the HTTP server which backs this web interface */ def bind() { |