[SPARK-27641][CORE] Fix MetricsSystem to remove unregistered source correctly #24556

chummyhe89 · 2019-05-08T13:06:12Z

What changes were proposed in this pull request?

check whether the source is registered before remove it.
return the registered source from LongAccumulatorSource and DoubleAccumulatorSource's register method,then we can remove it use metricsSystem.removeSource

How was this patch tested?

add new unit tests

…orrectly

szhem · 2019-05-08T15:10:04Z

core/src/main/scala/org/apache/spark/metrics/MetricsSystem.scala

-    val regName = buildRegistryName(source)
-    registry.removeMatching((name: String, _: Metric) => name.startsWith(regName))
+    if (sources.contains(source)) {
+      sources -= source


Is it really necessary to navigate sources twice - the first during contains, the 2nd - while removing it?

I add existing check because original code removes the metrics no matter whether the source was registered or not. i will use index to avoid the twice navigating.thanks for your advice.

szhem · 2019-05-08T15:17:36Z

core/src/main/scala/org/apache/spark/metrics/MetricsSystem.scala

+    if (sources.contains(source)) {
+      sources -= source
+      val regName = buildRegistryName(source)
+      registry.removeMatching((name: String, _: Metric) => name.startsWith(regName))


The original JIRA is more generic than it seems to be.
AccumulatorSource there is just for example. Just consider any other custom Source and not just AccumulatorSource.

There is a bug in the MetricsSystem that allows to add (supports) multiple sources with the same name, but does not allow to remove a single source with its metrics only. Removing a single source leads to removing all the registered metrics from all the sources of the same name.

Removing metrics from this shared registry by exact match seems to be more accurate than just by the corresponding prefix.

Apologies for my misunderstanding.I will correct it soon.

szhem · 2019-05-08T15:23:29Z

core/src/main/scala/org/apache/spark/metrics/source/AccumulatorSource.scala

@@ -64,10 +64,11 @@ class DoubleAccumulatorSource extends AccumulatorSource
 */


Changes in this file seems to be not necessary if removing metrics by exact match from the shared registry as described above.

szhem · 2019-05-12T18:04:02Z

core/src/main/scala/org/apache/spark/metrics/MetricsSystem.scala

+  private[spark] def getRegisteredNames(source: Source): Seq[String] = {
+    val sourceNamePrefix = buildRegistryName(source)
+    val metricNames = source.metricRegistry.getMetrics.keySet()
+    JavaConverters.asScalaSet(metricNames).map(k => s"${sourceNamePrefix}.${k}").toSeq


There is MetricRegistry.name method to create names instead of string interpolation.

It seems that JavaConverters are not used in such a way throughout the codebase

The result of the method is ArrayBuffer, so every lookup is O(N). What is the reason of converting a Set into a Seq?

szhem · 2019-05-12T18:13:27Z

core/src/main/scala/org/apache/spark/metrics/MetricsSystem.scala

+    if (index != -1) {
+      sources.remove(index)
+      val regNames = getRegisteredNames(source)
+      registry.removeMatching((name: String, _: Metric) => regNames.contains(name))


Is it really necessary to iterate over all the metrics in the registry to remove the necessary metrics?
Isn't it more efficient just to remove the metrics from the registry by their names using MetricRegistry.remove(String)?

szhem · 2019-05-12T18:14:13Z

core/src/main/scala/org/apache/spark/metrics/source/AccumulatorSource.scala

@@ -64,10 +64,11 @@ class DoubleAccumulatorSource extends AccumulatorSource
 */
 @Experimental
 object LongAccumulatorSource {
-  def register(sc: SparkContext, accumulators: Map[String, LongAccumulator]): Unit = {
+  def register(sc: SparkContext, accumulators: Map[String, LongAccumulator]): Source = {


What is the reason of changing this public API and exposing the package private Source?

it give a chance to remove the registered source of AccumulatorSource's subclass in spark

AccumulatorSource is package private as well. So to extend it, it will be necessary to place your code within the same package. So there is no need to expose package private Source and make it a part of the public contract.

To implement tests without changing the API you could use something like the following

val source = new LongAccumulatorSource source.register(accumulators) sc.env.metricsSystem.registerSource(source)

... instead of

LongAccumulatorSource.register(mockContext, Map("laF" -> laFirst, "laS" -> laSecond))

So the necessary source to delete during the tests will be already stored in the local variable.

It makes sense. thanks for your advise.

szhem · 2019-05-12T18:14:18Z

core/src/main/scala/org/apache/spark/metrics/source/AccumulatorSource.scala

    val source = new LongAccumulatorSource
    source.register(accumulators)
    sc.env.metricsSystem.registerSource(source)
+    source


What is the reason of changing this public API and exposing the package private Source?

szhem · 2019-05-12T18:14:25Z

core/src/main/scala/org/apache/spark/metrics/source/AccumulatorSource.scala

@@ -81,9 +82,10 @@ object LongAccumulatorSource {
 */
 @Experimental
 object DoubleAccumulatorSource {
-  def register(sc: SparkContext, accumulators: Map[String, DoubleAccumulator]): Unit = {
+  def register(sc: SparkContext, accumulators: Map[String, DoubleAccumulator]): Source = {


What is the reason of changing this public API and exposing the package private Source?

szhem · 2019-05-12T18:14:31Z

core/src/main/scala/org/apache/spark/metrics/source/AccumulatorSource.scala

    val source = new DoubleAccumulatorSource
    source.register(accumulators)
    sc.env.metricsSystem.registerSource(source)
+    source


What is the reason of changing this public API and exposing the package private Source?

szhem · 2019-05-12T18:21:05Z

core/src/test/scala/org/apache/spark/metrics/MetricsSystemSuite.scala

+    assert(metricsSystem.invokePrivate(registry()).getNames.contains("AccumulatorSource.laT"))
+    assert(!metricsSystem.invokePrivate(registry()).getNames.contains("AccumulatorSource.laF"))
+    assert(!metricsSystem.invokePrivate(registry()).getNames.contains("AccumulatorSource.laS"))
+  }


I'd suggest to pay an attention to how the other test cases in this test suite are organised by grouping the related blocks of code with the empty lines.

beliefer · 2019-05-13T11:16:25Z

core/src/main/scala/org/apache/spark/metrics/MetricsSystem.scala

-    val regName = buildRegistryName(source)
-    registry.removeMatching((name: String, _: Metric) => name.startsWith(regName))
+    val index = sources.indexOf(source)
+    if (index != -1) {


If there use sources.contains(source), it looks more graceful.

use index can avoid navigating sources twice.

using index can avoid navigating sources twice.

I know it, now.

beliefer · 2019-05-13T11:20:10Z

core/src/main/scala/org/apache/spark/metrics/MetricsSystem.scala

@@ -166,9 +172,12 @@ private[spark] class MetricsSystem private (
  }

  def removeSource(source: Source) {
-    sources -= source


I think sources -= source is is already looks good.
The method -= of mutable.ArrayBuffer as follows:

def -= (x: A): this.type = { val i = indexOf(x) if (i != -1) remove(i) this }

I think the original code works good.

but the original code will remove the source's metrics even if it is not registered.

I don't understand your meaning.If source not registered,The method -= of mutable.ArrayBuffer do nothing.

def removeSource(source: Source) { sources -= source val regName = buildRegistryName(source) registry.removeMatching((name: String, _: Metric) => name.startsWith(regName)) }

Even if the source is not in sources
the original code will execute the code after sources -= source which will remove the metrics of the source.

OK. I get it.
This change looks good to me.

What metrics would it remove erroneously, if the source were never registered?

szhem · 2019-05-15T08:24:47Z

core/src/main/scala/org/apache/spark/metrics/source/AccumulatorSource.scala

@@ -64,10 +64,11 @@ class DoubleAccumulatorSource extends AccumulatorSource
 */
 @Experimental
 object LongAccumulatorSource {
-  def register(sc: SparkContext, accumulators: Map[String, LongAccumulator]): Unit = {
+  def register(sc: SparkContext, accumulators: Map[String, LongAccumulator]): Source = {


AccumulatorSource is package private as well. So to extend it, it will be necessary to place your code within the same package. So there is no need to expose package private Source and make it a part of the public contract.

To implement tests without changing the API you could use something like the following

val source = new LongAccumulatorSource source.register(accumulators) sc.env.metricsSystem.registerSource(source)

... instead of

LongAccumulatorSource.register(mockContext, Map("laF" -> laFirst, "laS" -> laSecond))

So the necessary source to delete during the tests will be already stored in the local variable.

szhem

LGTM, thanks for the PR!

AmplabJenkins · 2019-09-16T18:13:28Z

Can one of the admins verify this patch?

chummyhe89 · 2019-12-23T10:14:01Z

too old to merge,so close it

[SPARK-27641][CORE] Fix MetricsSystem to remove unregistered source c…

67df51b

…orrectly

szhem suggested changes May 8, 2019

View reviewed changes

remove metrics of the specific source by exact match

38b37b3

szhem suggested changes May 12, 2019

View reviewed changes

beliefer reviewed May 13, 2019

View reviewed changes

use MetricRegistry.name method to generate metric name

06ae21f

szhem suggested changes May 15, 2019

View reviewed changes

revert changes to AccumulatorSource's subclass

38cf244

szhem approved these changes May 16, 2019

View reviewed changes

dongjoon-hyun added the SPARK CORE label Jun 14, 2019

Merge branch 'master' into master

71eee02

chummyhe89 closed this Dec 23, 2019

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-27641][CORE] Fix MetricsSystem to remove unregistered source correctly #24556

[SPARK-27641][CORE] Fix MetricsSystem to remove unregistered source correctly #24556

chummyhe89 commented May 8, 2019

szhem May 8, 2019

chummyhe89 May 9, 2019

szhem May 8, 2019 •

edited

chummyhe89 May 9, 2019

szhem May 8, 2019

szhem May 12, 2019

szhem May 12, 2019

szhem May 12, 2019

chummyhe89 May 14, 2019

szhem May 15, 2019

chummyhe89 May 15, 2019

szhem May 12, 2019

szhem May 12, 2019

szhem May 12, 2019

szhem May 12, 2019

beliefer May 13, 2019

chummyhe89 May 14, 2019

chummyhe89 May 14, 2019

beliefer May 16, 2019

beliefer May 13, 2019

chummyhe89 May 14, 2019

beliefer May 15, 2019

chummyhe89 May 15, 2019

beliefer May 16, 2019

srowen Dec 21, 2019

szhem May 15, 2019

szhem left a comment

AmplabJenkins commented Sep 16, 2019

chummyhe89 commented Dec 23, 2019

		@@ -64,10 +64,11 @@ class DoubleAccumulatorSource extends AccumulatorSource
		*/

[SPARK-27641][CORE] Fix MetricsSystem to remove unregistered source correctly #24556

[SPARK-27641][CORE] Fix MetricsSystem to remove unregistered source correctly #24556

Conversation

chummyhe89 commented May 8, 2019

What changes were proposed in this pull request?

How was this patch tested?

Choose a reason for hiding this comment

Choose a reason for hiding this comment

szhem May 8, 2019 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

szhem left a comment

Choose a reason for hiding this comment

AmplabJenkins commented Sep 16, 2019

chummyhe89 commented Dec 23, 2019

szhem May 8, 2019 •

edited