[SPARK-8743] [Streaming]: Deregister Codahale metrics for streaming when StreamingContext is closed #7250

nssalian · 2015-07-07T03:20:36Z

The issue link: https://issues.apache.org/jira/browse/SPARK-8743
Deregister Codahale metrics for streaming when StreamingContext is closed

Design:
Adding the method calls in the appropriate start() and stop () methods for the StreamingContext

Actions in the PullRequest:

Added the registerSource method call to the start method for the Streaming Context.
Added the removeSource method to the stop method.
Added comments for both 1 and 2 and comment to show initialization of the StreamingSource

Requesting Review.

…for the Streaming Context. Added the removeSource method to the stop method. Added comments for both

jerryshao · 2015-07-07T05:51:08Z

streaming/src/main/scala/org/apache/spark/streaming/StreamingContext.scala

@@ -577,6 +575,12 @@ class StreamingContext private[streaming] (
   * @throws IllegalStateException if the StreamingContext is already stopped.
   */
  def start(): Unit = synchronized {
+  /**
+   * Registering Streaming Metrics at the start of the StreamingContext


I think it is better to use // rather than /** */ for internal code commenting.

jerryshao · 2015-07-07T05:56:06Z

This change LGTM, actually will get rid of some annoying logs :).

nssalian · 2015-07-07T07:28:17Z

@jerryshao thank you for the comment. I made the changes.
Please let me know if you think I could add anything additional.

srowen · 2015-07-07T08:20:23Z

Jenkins, test this please

SparkQA · 2015-07-07T08:27:58Z

Test build #36660 has finished for PR 7250 at commit a665965.

This patch fails Scala style tests.
This patch merges cleanly.
This patch adds no public classes.

srowen · 2015-07-07T08:29:57Z

streaming/src/main/scala/org/apache/spark/streaming/StreamingContext.scala

@@ -577,6 +575,10 @@ class StreamingContext private[streaming] (
   * @throws IllegalStateException if the StreamingContext is already stopped.
   */
  def start(): Unit = synchronized {
+  //Registering Streaming Metrics at the start of the StreamingContext


My guess is this fails style checks. Indent it, and add a space

nssalian · 2015-07-07T17:55:12Z

@srowen, I made the changes and ran a dev test on my repo. The scala errors weren't present.

srowen · 2015-07-07T17:59:29Z

@nssalian that is indeed the failure: "Insert a space after the start of the comment" Did you run ./dev/scalastyle?

nssalian · 2015-07-07T18:06:25Z

nsalian-MBP:spark nsalian$ ./dev/scalastyle

Scalastyle checks passed.

nssalian · 2015-07-07T18:10:03Z

Should have phrased it better, the tests ran fine.

tdas · 2015-07-07T19:02:36Z

streaming/src/main/scala/org/apache/spark/streaming/StreamingContext.scala

@@ -688,6 +690,8 @@ class StreamingContext private[streaming] (
    } finally {
      // The state should always be Stopped after calling `stop()`, even if we haven't started yet
      state = STOPPED
+      // De-registering Streaming Metrics of the StreamingContext
+      env.metricsSystem.removeSource(streamingSource)


I think should be done try, after calling scheduler.stop(). In the finally this will fail, if the SparkContext has been stopped and the env is invalid.

I think it would be better to add try as TD said, otherwise you may meet exception when SparkContext is stopped.

Changed it on my local. Added to the latest commit.

tdas · 2015-07-07T19:04:17Z

You need to add unit tests for this. Please add them to the StreamingContextSuite. You can probably add it to existing unit tests that test start and stop.

SparkQA · 2015-07-07T19:28:25Z

Test build #1004 has finished for PR 7250 at commit 7621adf.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

nssalian · 2015-07-07T21:18:02Z

Will update shortly with the changes.

Thank you.

…e of the sources ArrayBuffer

nssalian · 2015-07-07T23:22:33Z

To illustrate,
Test case:

Testing start of the streamingContext and checking state.
Storing the size of the sources ArrayBuffer which will have a new source added
Sleep for 100 ms.
Stopping context and checking state
Also checking whether the size of the ArrayBuffer was decreased as the source was removed.
I changed the scope of the sources ArrayBuffer to do this.

Would like some feedback on this approach.

Thank you.

nssalian · 2015-07-08T18:01:29Z

@srowen, I wanted to check whether the size was decremented at all.

Couldn't think of a way to assert that the source has been removed since the sources ArrayBuffer is still private.

srowen · 2015-07-08T18:09:09Z

But this line doesn't check the state afterwards. You're reading the same value twice; x == x - 1 can never be true. (There's an extra space at the end of that line too.) I think there's still an outstanding comment from TD here too.

…compare with the original size after removal

nssalian · 2015-07-08T18:58:20Z

Based on @srowen's comment, I made the change and added updatedSourcesSize to check the ArrayBuffer size after the source is remove to assert that the size was indeed decremented.

jerryshao · 2015-07-09T00:52:26Z

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

@@ -73,7 +73,7 @@ private[spark] class MetricsSystem private (
  private[this] val metricsConfig = new MetricsConfig(conf)

  private val sinks = new mutable.ArrayBuffer[Sink]
-  private val sources = new mutable.ArrayBuffer[Source]
+  val sources = new mutable.ArrayBuffer[Source]


I think it would be better not to loose the restriction to expose this to the public user just only for test convenience. you could refer here to see how to test with private field.

Thanks @jerryshao will add a test similarly.

@nssalian This has not been addressed. This should not be public and see the link to see how to keep it private and yet access it in tests.

jerryshao · 2015-07-09T01:43:34Z

streaming/src/main/scala/org/apache/spark/streaming/StreamingContext.scala

@@ -674,6 +676,8 @@ class StreamingContext private[streaming] (
          logWarning("StreamingContext has already been stopped")
        case ACTIVE =>
          scheduler.stop(stopGracefully)
+          // De-registering Streaming Metrics of the StreamingContext
+          env.metricsSystem.removeSource(streamingSource)


I think here remove the source in ACTIVE state may introduce some corner case problem. For example, since metrics source is added when StreamingContext's state is INITIALIZED, if we met exception at this point, we will never change the state into ACTIVE, so metrics source cannot be removed, since you only assume the state is ACTIVE to remove the source. What do you think?

The idea was to register at the call of the start().
So, based on your comment, that would mean registering the sources after the state is set to INITIALIZEDand before.

def start(): Unit = synchronized { // Registering Streaming Metrics at the start of the StreamingContext assert(env != null) assert(env.metricsSystem != null) env.metricsSystem.registerSource(streamingSource)

Makes sense to have it after INITIALIZED and before the synchronized block of ACTIVE and STOPPED.
@tdas, could add more light.

What I mean is to move this line out of ACTIVE condition, In any case you have to deregister the metrics source, not only in ACTIVE state.

@jerryshao In this PR the streaming metric source is added only after starting the context. So it is fine to deregister only when the state is ACTIVE. Also that maintain the characteristic that stop is idempotent.

I think it is OK for normal case, but what if an exception is met after metrics is successfully registered, but before changing the state into ACTIVE, according to the code, it will try the exception and change state into STOPPED, so at that situation, if we call stop(), we will never de-register the metrics source according to the current implementation.

Yeah good point. I suggested above that the registering be moved after the ACTIVE status change.

tdas · 2015-07-10T21:44:34Z

streaming/src/test/scala/org/apache/spark/streaming/StreamingContextSuite.scala

+
+    ssc.start()
+    assert(ssc.getState() === StreamingContextState.INITIALIZED)
+    assert(StreamingContextSuite.sources.get(StreamingContextSuite.streamingSource)!= "null")


I dont get this statement here. You are calling StreamingContext.sources.get() on an object of type java.lang.reflect.Field (that is, StreamingContextSuite.streamingSource) rather than the actual StreamingContext object ssc above. And then you are comparing that it is not the string "null"? I am not sure what is being tested here.

Checking to see if the source is not returning a null and is actually present in the sources ArrayBuffer.

A much nicer way to do this is to do it they way it is done in https://github.com/apache/spark/blob/master/core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala#L782

using the PrivateTester stuff. Please use that, we want to stay consistent to the rest of the code base, and the code is easier to read and understand.

Makes sense. I'll write that up. Was having problems when I initially wrote as it was in ExecutionManagerSuite. I'll figure out something. Hence went with reflection

Thank you.

… check for Sources containing or not containing StreamingSource

nssalian · 2015-07-11T23:01:08Z

@tdas, I added the new test as per PrivateMethodTester.

srowen · 2015-07-12T08:54:56Z

streaming/src/test/scala/org/apache/spark/streaming/StreamingContextSuite.scala


 import org.apache.spark.storage.StorageLevel
 import org.apache.spark.streaming.dstream.DStream
 import org.apache.spark.streaming.receiver.Receiver
 import org.apache.spark.util.Utils
-import org.apache.spark.{Logging, SparkConf, SparkContext, SparkException, SparkFunSuite}
+
+import org.apache.spark._


I wouldn't make this a _ import. Leave it as it was. ALso I think the collection imports become unnecessary; see below

…fic imports

…trics

nssalian · 2015-07-12T17:32:02Z

Hitting this error on the test for StreamingContextSuite.
Is the streamingSource not being found in the right ArrayBuffer? I tried different variations of the registrations to try alleviate this. Didn't help. @tdas any suggestions?

Ran this:
build/sbt -Pyarn -Phadoop-2.4 -Dhadoop.version=2.5.0 -Pscala-2.10
project streaming
~test-only *StreamingContextSuite

Test failure message:

ArrayBuffer(org.apache.spark.scheduler.DAGSchedulerSource@1473d83a, org.apache.spark.storage.BlockManagerSource@7560392b) did not contain org.apache.spark.streaming.StreamingSource@1c71ddd7 (StreamingContextSuite.scala:322)

srowen · 2015-07-12T21:21:59Z

streaming/src/test/scala/org/apache/spark/streaming/StreamingContextSuite.scala

@@ -21,17 +21,22 @@ import java.io.{File, NotSerializableException}
 import java.util.concurrent.atomic.AtomicInteger

 import org.apache.commons.io.FileUtils
+import org.apache.spark.metrics.MetricsSystem
+import org.apache.spark.metrics.source.Source


These need to be with other spark imports, and the spark imports shouldn't have a blank line in the middle.

… ACTIVE and this check fails

nssalian · 2015-07-12T22:50:07Z

The test failed with the assert(ssc.getState() === StreamingContextState.INITIALIZED)
as after the start() method, the state goes to ACTIVE and fails to match with INITIALIZED.
Moving it to after the initialization of ssc.

@srowen, I've added the changes as you suggested.
Thanks.

srowen · 2015-07-12T23:14:03Z

That can't be the failure you're referring to above, right? looks clearly like the assertion about which sources are still in the registered set. Anyway, the assertion looks better now. You need to rebase this PR now, and two of the imports are still in the wrong place.

…for the Streaming Context. Added the removeSource method to the stop method. Added comments for both Added // instead of /** for commenting in code Added indentation and Space at the comment on line 578; Registering.. Added test case for de-register metrics and made a change to the scope of the sources ArrayBuffer Added additional variable to check the updated Sources size value to compare with the original size after removal Added the removeSource method in try Removed the assert for the env field, added the registerSource line in the INITIALIZED block and kept the removeSource() in the ACTIVE block Added test to check registering and de-registering of streamingSource Removed unused imports Moved the registerSource() call before line 601 Changed scope of sources and corrected comments for helper Removed extra line in Helper Methods section Added helper method for private methods and changed the test logic to check for Sources containing or not containing StreamingSource Changed import statements to remove unnecessary imports and add specific imports Removed types for fields in test for registering and deregistering metrics Changed imports statements, negated test statement and removed postfix Removed added comment to Assert for INITIALIZED state Removing the INITIALIZED check since after start() the state moves to ACTIVE and this check fails Move the INITIALIZED state check to when the ssc is initialized

…PARK-8743

nssalian · 2015-07-12T23:46:41Z

Right. The negation took care of the failure above and I fixed the assertion error as well.
Removed the spacing for all of them.

Made a unified commit. 299a57d
Ignore the last one, I''l revert that.

nssalian · 2015-07-12T23:46:52Z

Right. The negation took care of the failure above and I fixed the assertion error as well.
Removed the spacing for all of them.

Made a unified commit. 299a57d
Ignore the last one, I''l revert that.

nssalian · 2015-07-13T00:36:24Z

@tdas @srowen, shall I create a fresh PR to avoid any confusion?
I can reference this one.

nssalian · 2015-07-13T01:42:13Z

Closing this PR: for #7362

srowen · 2015-07-13T07:50:05Z

@nssalian you can make a new PR, yes. I usually like to stick to one PR to preserve the discussion in one place, but it's not a huge deal.

You don't need to squash your commit history like you might in PRs for other projects, since the merge process will take care of that. Still, you can squash some of your commits in your PR branch and force-push if you like for clarity, though you don't need to.

After a rebase to resolve merge conflicts you can simply force-push your PR branch, note.

…en StreamingContext is closed The issue link: https://issues.apache.org/jira/browse/SPARK-8743 Deregister Codahale metrics for streaming when StreamingContext is closed Design: Adding the method calls in the appropriate start() and stop () methods for the StreamingContext Actions in the PullRequest: 1) Added the registerSource method call to the start method for the Streaming Context. 2) Added the removeSource method to the stop method. 3) Added comments for both 1 and 2 and comment to show initialization of the StreamingSource 4) Added a test case to check for both registration and de-registration of metrics Previous closed PR for reference: #7250 Author: Neelesh Srinivas Salian <[email protected]> Closes #7362 from nssalian/branch-SPARK-8743 and squashes the following commits: 7d998a3 [Neelesh Srinivas Salian] Removed the Thread.sleep() call 8b26397 [Neelesh Srinivas Salian] Moved the scalatest.{} import 0e8007a [Neelesh Srinivas Salian] moved import org.apache.spark{} to correct place daedaa5 [Neelesh Srinivas Salian] Corrected Ordering of imports 8873180 [Neelesh Srinivas Salian] Removed redundancy in imports 59227a4 [Neelesh Srinivas Salian] Changed the ordering of the imports to classify scala and spark imports d8cb577 [Neelesh Srinivas Salian] Added registerSource to start() and removeSource to stop(). Wrote a test to check the registration and de-registration

…en StreamingContext is closed The issue link: https://issues.apache.org/jira/browse/SPARK-8743 Deregister Codahale metrics for streaming when StreamingContext is closed Design: Adding the method calls in the appropriate start() and stop () methods for the StreamingContext Actions in the PullRequest: 1) Added the registerSource method call to the start method for the Streaming Context. 2) Added the removeSource method to the stop method. 3) Added comments for both 1 and 2 and comment to show initialization of the StreamingSource 4) Added a test case to check for both registration and de-registration of metrics Previous closed PR for reference: #7250 Author: Neelesh Srinivas Salian <[email protected]> Closes #7362 from nssalian/branch-SPARK-8743 and squashes the following commits: 7d998a3 [Neelesh Srinivas Salian] Removed the Thread.sleep() call 8b26397 [Neelesh Srinivas Salian] Moved the scalatest.{} import 0e8007a [Neelesh Srinivas Salian] moved import org.apache.spark{} to correct place daedaa5 [Neelesh Srinivas Salian] Corrected Ordering of imports 8873180 [Neelesh Srinivas Salian] Removed redundancy in imports 59227a4 [Neelesh Srinivas Salian] Changed the ordering of the imports to classify scala and spark imports d8cb577 [Neelesh Srinivas Salian] Added registerSource to start() and removeSource to stop(). Wrote a test to check the registration and de-registration (cherry picked from commit b7bcbe2) Signed-off-by: Tathagata Das <[email protected]>

SPARK-8743: Added the registerSource method call to the start method …

92fa04b

…for the Streaming Context. Added the removeSource method to the stop method. Added comments for both

jerryshao reviewed Jul 7, 2015
View reviewed changes

Added // instead of /** for commenting in code

a665965

srowen reviewed Jul 7, 2015
View reviewed changes

Added indentation and Space at the comment on line 578; Registering..

7621adf

tdas reviewed Jul 7, 2015
View reviewed changes

Added test case for de-register metrics and made a change to the scop…

18bcc7e

…e of the sources ArrayBuffer

Added additional variable to check the updated Sources size value to …

e4f00d7

…compare with the original size after removal

jerryshao reviewed Jul 9, 2015
View reviewed changes

Added the removeSource method in try

f5e47e0

jerryshao reviewed Jul 9, 2015
View reviewed changes

tdas mentioned this pull request Jul 9, 2015

[SPARK-8895] MetricsSystem.removeSource should be called in StreamingContext.stop #7317

Closed

tdas reviewed Jul 10, 2015
View reviewed changes

Added helper method for private methods and changed the test logic to…

74598ce

… check for Sources containing or not containing StreamingSource

srowen reviewed Jul 12, 2015
View reviewed changes

Neelesh Srinivas Salian added 2 commits July 12, 2015 07:38

Changed import statements to remove unnecessary imports and add speci…

e37a2f3

…fic imports

Removed types for fields in test for registering and deregistering me…

f54afcf

…trics

srowen reviewed Jul 12, 2015
View reviewed changes

Neelesh Srinivas Salian added 3 commits July 12, 2015 15:38

Changed imports statements, negated test statement and removed postfix

ea0dc1a

Removed added comment to Assert for INITIALIZED state

a0f1950

Removing the INITIALIZED check since after start() the state moves to…

2a81287

… ACTIVE and this check fails

Move the INITIALIZED state check to when the ssc is initialized

5d3af31

Neelesh Srinivas Salian added 2 commits July 12, 2015 16:33

Merge branch 'SPARK-8743' of https://github.com/nssalian/spark into S…

b86e80b

…PARK-8743

nssalian closed this Jul 12, 2015

nssalian reopened this Jul 12, 2015

nssalian mentioned this pull request Jul 13, 2015

[SPARK-8743] [Streaming]: Deregister Codahale metrics for streaming when StreamingContext is closed #7362

Closed

nssalian closed this Jul 13, 2015

[SPARK-8743] [Streaming]: Deregister Codahale metrics for streaming when StreamingContext is closed #7250

[SPARK-8743] [Streaming]: Deregister Codahale metrics for streaming when StreamingContext is closed #7250

Conversation

nssalian commented Jul 7, 2015

Choose a reason for hiding this comment

jerryshao commented Jul 7, 2015

nssalian commented Jul 7, 2015

srowen commented Jul 7, 2015

SparkQA commented Jul 7, 2015

Choose a reason for hiding this comment

nssalian commented Jul 7, 2015

srowen commented Jul 7, 2015

nssalian commented Jul 7, 2015

nssalian commented Jul 7, 2015

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tdas commented Jul 7, 2015

SparkQA commented Jul 7, 2015

nssalian commented Jul 7, 2015

nssalian commented Jul 7, 2015

nssalian commented Jul 8, 2015

srowen commented Jul 8, 2015

nssalian commented Jul 8, 2015

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

nssalian commented Jul 11, 2015

Choose a reason for hiding this comment

nssalian commented Jul 12, 2015

Choose a reason for hiding this comment

nssalian commented Jul 12, 2015

srowen commented Jul 12, 2015

nssalian commented Jul 12, 2015

nssalian commented Jul 12, 2015

nssalian commented Jul 13, 2015

nssalian commented Jul 13, 2015

srowen commented Jul 13, 2015