HomeMesosphereNo notifications. 4 unresolved issues.
  • Search
Differential D942

Fix flaky metrics tests, and fix issues with stream monitoring
ClosedAll Users
Actions

Authored by timcharper on Jul 27 2017, 5:26 AM.

Details

Reviewers
zen-dog
unterstein
jenkins
Commits
rMARATHONb58c7b91964a: Fix flaky metrics tests, and fix issues with stream monitoring
JIRA Issues
JIRA MARATHON-7574 MetricsTimerTest fails 2% of the time.
Summary
  • Calling Kamon.start() in the before block causes erratic behavior in Kamon. I don't know why this is. But it is. We should only call Kamon.start() only for tests that require it.
  • Stream monitoring was causing exceptions to be swallowed.
  • Stream monitoring started with the first element to flow through the stream; this meant that streams that have only one element will have a duration of close to 0.
Test Plan

modify test locally to loop it 1000 times to assert flakiness is gone

Diff Detail

Repository
rMARATHON marathon
Lint
Automatic diff as part of commit; lint not applicable.
Unit
Automatic diff as part of commit; unit tests not applicable.
timcharper created this revision.Jul 27 2017, 5:26 AM
jenkins requested changes to this revision.Jul 27 2017, 5:32 AM
This revision now requires changes to proceed.Jul 27 2017, 5:32 AM
jenkins accepted this revision.Jul 27 2017, 5:48 AM
This revision is now accepted and ready to land.Jul 27 2017, 5:48 AM
jenkins added a comment.Jul 27 2017, 5:48 AM
✔ Build of 3860 completed jenkins-public-marathon-phabricator-598.

You can create a DC/OS with your patched Marathon by creating a new pull
request with the following changes in buildinfo.json:

"url": "https://downloads.mesosphere.io/marathon/snapshots/marathon-1.5.0-SNAPSHOT-670-g762a11f.tgz",
"sha1": "38be33908835009efb9e0c35757b838b4e5e3be1"

＼\ ٩( ᐛ )و /／

zen-dog requested changes to this revision.Jul 27 2017, 11:51 AM

I wonder why java.time.Clock and not mesosphere.marathon.core.base.Clock. Otherwise that looks great and it also doesn't fail in my local loop. Thx!

src/main/scala/mesosphere/marathon/metrics/HistogramTimer.scala
30

clock.instant instead of Option[Long] is neat. But I wonder why java.time.Clock and not our own mesosphere.marathon.core.base.Clock which we use everywhere else?

45

I wonder what kind of errors we swallowed here previously...

This revision now requires changes to proceed.Jul 27 2017, 11:51 AM
zen-dog accepted this revision.Jul 27 2017, 3:26 PM
zen-dog added inline comments.
src/main/scala/mesosphere/marathon/metrics/HistogramTimer.scala
30

...because our Clock has ms precision :(

This revision is now accepted and ready to land.Jul 27 2017, 3:26 PM
unterstein accepted this revision.Jul 27 2017, 3:28 PM
Closed by commit rMARATHONb58c7b91964a: Fix flaky metrics tests, and fix issues with stream monitoring (authored by timcharper). · Explain WhyJul 27 2017, 4:01 PM
This revision was automatically updated to reflect the committed changes.
jdef added a subscriber: jdef.Jul 28 2017, 4:11 AM
jdef added inline comments.
src/main/scala/mesosphere/marathon/metrics/HistogramTimer.scala
22

curious: why was final removed?

Revision Contents

Diff 3861

View Options

project/Dependencies.scala

Show First 20 Lines • Show All 156 Lines • ▼ Show 20 Line(s)
157157 val java8Compat = "org.scala-lang.modules" %% "scala-java8-compat" % V.Java8Compat
158158 val scalaLogging = "com.typesafe.scala-logging" %% "scala-logging" % V.ScalaLogging
159159 val scalaxml = "org.scala-lang.modules" %% "scala-xml" % "1.0.5"
160160 val raven = "com.getsentry.raven" % "raven-logback" % V.Raven
161161 val commonsCompress = "org.apache.commons" % "commons-compress" % V.ApacheCommonsCompress
162162 val akkaSse = "de.heikoseeberger" %% "akka-sse" % V.AkkaSSE
163163
164164 object Kamon {
165 val Version = "0.6.5"
165 val Version = "0.6.7"
166166
167167 val core = "io.kamon" %% "kamon-core" % Version % "compile"
168 val akka = "io.kamon" %% "kamon-akka" % "0.6.3" % "compile"
169 val autoweave = "io.kamon" %% "kamon-autoweave" % Version % "compile"
168 val akka = "io.kamon" %% "kamon-akka" % Version % "compile"
169 val autoweave = "io.kamon" %% "kamon-autoweave" % "0.6.5" % "compile"
170170 val scala = "io.kamon" %% "kamon-scala" % Version % "compile"
171171 val systemMetrics = "io.kamon" %% "kamon-system-metrics" % Version % "compile"
172 val akkaHttp = "io.kamon" %% "kamon-akka-http-experimental" % "0.6.3" % "compile"
173172
174173 object Backends {
175174 val statsd = "io.kamon" %% "kamon-statsd" % Version % "compile"
176175 val datadog = "io.kamon" %% "kamon-datadog" % Version % "compile"
177176 val jmx = "io.kamon" %% "kamon-jmx" % Version % "compile"
178177 }
179178
180179 // there are some issues with the Akka modules that are really unclear
Show All 16 Lines
View Options

src/main/scala/mesosphere/marathon/metrics/HistogramTimer.scala

11package mesosphere.marathon
22package metrics
33
44import akka.stream.scaladsl.{ Flow, Source }
55import akka.stream.stage._
66import akka.stream.{ Attributes, FlowShape, Inlet, Outlet }
7import java.time.{ Clock, Duration, Instant }
78import kamon.Kamon
89import kamon.metric.instrument
910import kamon.metric.instrument.Time
1011import mesosphere.util.CallerThreadExecutionContext
1112
1213import scala.Exception
1314import scala.concurrent.Future
1415import scala.concurrent.duration.FiniteDuration
1516import scala.util.control.NonFatal
1617
1718/**
18 * Akka Graph Stage that measures the time from the start of the stream until the end of it.
19 * Akka Graph Stage that measures the time from the start of the stream until the end of it, where start is defined as
20 * the instant the stream is materialized
1921 */
20private final class TimedStage[T](histogram: instrument.Histogram) extends GraphStage[FlowShape[T, T]] {
22private[metrics] class TimedStage[T](histogram: instrument.Histogram, clock: Clock) extends GraphStage[FlowShape[T, T]] {
jdefUnsubmitted
Not Done

curious: why was final removed?

2123 val in = Inlet[T]("timer.in")
2224 val out = Outlet[T]("timer.out")
2325
2426 @scala.throws[Exception](classOf[Exception])
2527 override def createLogic(inheritedAttributes: Attributes): GraphStageLogic = {
2628
2729 val logic = new GraphStageLogic(shape) {
28 var start: Option[Long] = Option.empty[Long]
30 private val start: Instant = clock.instant
zen-dogUnsubmitted
Not Done

clock.instant instead of Option[Long] is neat. But I wonder why java.time.Clock and not our own mesosphere.marathon.core.base.Clock which we use everywhere else?

zen-dogUnsubmitted
Not Done

...because our Clock has ms precision :(

2931
3032 setHandler(in, new InHandler {
3133 @scala.throws[Exception](classOf[Exception])
3234 override def onPush(): Unit = push(out, grab(in))
3335
3436 @scala.throws[Exception](classOf[Exception])
3537 override def onUpstreamFinish(): Unit = {
36 start.foreach { startTime =>
37 histogram.record(System.nanoTime() - startTime)
38 histogram.record(Duration.between(start, clock.instant).toNanos())
39 super.onUpstreamFinish()
3840 }
39 completeStage()
40 }
4141
4242 @scala.throws[Exception](classOf[Exception])
4343 override def onUpstreamFailure(ex: Throwable): Unit = {
44 start.foreach { startTime =>
45 histogram.record(System.nanoTime() - startTime)
44 histogram.record(Duration.between(start, clock.instant).toNanos())
45 super.onUpstreamFailure(ex)
zen-dogUnsubmitted
Not Done

I wonder what kind of errors we swallowed here previously...

4646 }
47 completeStage()
48 }
4947 })
5048
5149 setHandler(out, new OutHandler {
5250 @scala.throws[Exception](classOf[Exception])
53 override def onPull(): Unit = {
54 if (start.isEmpty) start = Some(System.nanoTime())
55 pull(in)
56 }
51 override def onPull(): Unit = pull(in)
5752 })
5853 }
5954 logic
6055 }
6156
6257 override def shape: FlowShape[T, T] = FlowShape.of(in, out)
6358}
6459
Show All 9 Lines
7469 case NonFatal(e) =>
7570 histogram.record(System.nanoTime() - start)
7671 throw e
7772 }
7873 future.onComplete(_ => histogram.record(System.nanoTime() - start))(CallerThreadExecutionContext.callerThreadExecutionContext)
7974 future
8075 }
8176
82 def forSource[T, M](f: => Source[T, M]): Source[T, M] = {
77 def forSource[T, M](f: => Source[T, M])(implicit clock: Clock = Clock.systemUTC): Source[T, M] = {
8378 val start = System.nanoTime()
8479 val src = f
85 val flow = Flow.fromGraph(new TimedStage[T](histogram))
80 val flow = Flow.fromGraph(new TimedStage[T](histogram, clock))
8681 val transformed = src.via(flow)
8782 transformed
8883 }
8984
9085 def blocking[T](f: => T): T = {
9186 val start = System.nanoTime()
9287 try {
9388 f
Show All 20 Lines
View Options

src/main/scala/mesosphere/marathon/metrics/Metrics.scala

11package mesosphere.marathon
22package metrics
33
44import akka.Done
55import akka.actor.{ Actor, ActorRef, ActorRefFactory, Props }
66import akka.stream.scaladsl.Source
7import java.time.Clock
78import kamon.Kamon
89import kamon.metric.SubscriptionsDispatcher.TickMetricSnapshot
910import kamon.metric.instrument.Histogram.DynamicRange
1011import kamon.metric.instrument.{ CollectionContext, Time, UnitOfMeasurement }
1112import kamon.metric.{ Entity, SubscriptionFilter, instrument }
1213import kamon.util.{ MapMerge, MilliTimestamp }
1314
1415import scala.concurrent.Future
Show All 26 Lines
4142 def decrement(times: Long): MinMaxCounter
4243 def refreshValues(): MinMaxCounter
4344 def record(value: Long): MinMaxCounter
4445 def record(value: Long, count: Long): MinMaxCounter
4546}
4647
4748trait Timer {
4849 def apply[T](f: => Future[T]): Future[T]
49 def forSource[T, M](f: => Source[T, M]): Source[T, M]
50 def forSource[T, M](f: => Source[T, M])(implicit clock: Clock = Clock.systemUTC): Source[T, M]
5051 def blocking[T](f: => T): T
5152 def update(value: Long): Timer
5253 def update(duration: FiniteDuration): Timer
5354}
5455
5556object AcceptAllFilter extends SubscriptionFilter {
5657 override def accept(entity: Entity): Boolean = true
5758}
▲ Show 20 Lines • Show All 131 Lines • Show Last 20 Lines
View Options

src/test/scala/mesosphere/UnitTest.scala

11package mesosphere
22
33import java.util.concurrent.{ LinkedBlockingDeque, TimeUnit }
44
55import akka.actor.{ ActorSystem, Scheduler }
66import akka.stream.{ ActorMaterializer, Materializer }
77import akka.testkit.{ TestActor, TestActorRef, TestKitBase }
88import akka.util.Timeout
99import com.typesafe.config.{ Config, ConfigFactory }
1010import com.typesafe.scalalogging.StrictLogging
1111import com.wix.accord.{ Failure, Result, Success }
12import kamon.Kamon
1312import mesosphere.marathon.Normalization
1413import mesosphere.marathon.ValidationFailedException
1514import mesosphere.marathon.api.v2.Validation
1615import mesosphere.marathon.test.Mockito
1716import org.scalatest.matchers.{ BeMatcher, MatchResult, Matcher }
1817import org.scalatest._
1918import org.scalatest.concurrent.{ JavaFutures, ScalaFutures, TimeLimitedTests }
2019import org.scalatest.time.{ Minute, Minutes, Seconds, Span }
▲ Show 20 Lines • Show All 74 Lines • ▼ Show 20 Line(s)
9594 with AppendedClues
9695 with StrictLogging
9796 with Mockito
9897 with BeforeAndAfterAll
9998 with TimeLimitedTests {
10099
101100 override val timeLimit = Span(1, Minute)
102101
103 override def beforeAll(): Unit = {
104 Kamon.start()
105 super.beforeAll()
106 }
107
108102 override implicit lazy val patienceConfig: PatienceConfig = PatienceConfig(timeout = Span(5, Seconds))
109103}
110104
111105abstract class UnitTest extends WordSpec with UnitTestLike
112106
113107trait AkkaUnitTestLike extends UnitTestLike with TestKitBase {
114108 protected lazy val akkaConfig: Config = ConfigFactory.parseString(
115109 s"""
116110 |akka.test.default-timeout=${patienceConfig.timeout.millisPart}
117111 """.stripMargin).withFallback(ConfigFactory.load())
118112 implicit lazy val system: ActorSystem = {
119 Kamon.start()
120113 ActorSystem(suiteName, akkaConfig)
121114 }
122115 implicit lazy val scheduler: Scheduler = system.scheduler
123116 implicit lazy val mat: Materializer = ActorMaterializer()
124117 implicit lazy val ctx: ExecutionContextExecutor = system.dispatcher
125118 implicit val askTimeout: Timeout = Timeout(patienceConfig.timeout.toMillis, TimeUnit.MILLISECONDS)
126119
127120 def newTestActor() =
▲ Show 20 Lines • Show All 47 Lines • Show Last 20 Lines
View Options

src/test/scala/mesosphere/marathon/metrics/MetricsTimerTest.scala

11package mesosphere.marathon
22package metrics
33
4import akka.stream.scaladsl.{ Sink, Source }
4import akka.stream.scaladsl.{ Keep, Sink, Source }
55import kamon.metric.instrument.CollectionContext
66import mesosphere.AkkaUnitTest
7import mesosphere.marathon.test.SettableClock
8import org.scalatest.Inside
9import org.scalatest.concurrent.Eventually
10import org.scalatest.exceptions.TestFailedException
711
812import scala.Exception
913import scala.concurrent.Promise
10import scala.util.Try
14import scala.concurrent.duration._
15import scala.util.{ Try, Failure }
1116
12class MetricsTimerTest extends AkkaUnitTest {
17class MetricsTimerTest extends AkkaUnitTest with Eventually with Inside {
18
1319 "Metrics Timers" should {
1420 "time crashing call" in {
1521 When("doing the call (but the future is delayed)")
1622 val timer = HistogramTimer("timer")
1723
1824 val failure: RuntimeException = new scala.RuntimeException("failed")
1925 val attempt = Try(timer(throw failure))
2026
▲ Show 20 Lines • Show All 45 Lines • ▼ Show 20 Line(s)
6672 timer.histogram.collect(CollectionContext(10)).numberOfMeasurements should be(1L)
6773 // TODO: check time but we need to mock the time first
6874
6975 And("the failure should be preserved")
7076 result.failed.futureValue should be(failure)
7177 }
7278
7379 "measure a successful source" in {
80 implicit val clock = new SettableClock()
7481 val timer = HistogramTimer("timer")
7582 val promise = Promise[Int]()
76 val sourceFuture = timer.forSource(Source.fromFuture(promise.future)).runWith(Sink.seq)
83 val graph = timer.forSource(Source.fromFuture(promise.future))
84 .toMat(Sink.seq)(Keep.right)
85
86 clock.plus(1.second)
87 val sourceFuture = graph.run
7788 timer.histogram.collect(CollectionContext(10)).numberOfMeasurements should be(0L)
89 clock.plus(1.second)
7890 promise.success(1)
7991 sourceFuture.futureValue should contain theSameElementsAs Seq(1)
80 timer.histogram.collect(CollectionContext(10)).numberOfMeasurements should be(1L)
92 // histograms are not precise, so we check the range
93 timer.histogram.collect(CollectionContext(10)).max shouldEqual (1000000000L +- 100000000L)
8194 }
8295
83 "measure a failed source" in {
96 "measure a failed source (and propagate the exception)" in {
97 implicit val clock = new SettableClock()
8498 val timer = HistogramTimer("timer")
8599 val promise = Promise[Int]()
86 val sourceFuture = timer.forSource(Source.fromFuture(promise.future)).runWith(Sink.seq)
100 val graph = timer.forSource(Source.fromFuture(promise.future))
101 .toMat(Sink.seq)(Keep.right)
102 clock.plus(1.second)
103 val sourceFuture = graph.run
87104 timer.histogram.collect(CollectionContext(10)).numberOfMeasurements should be(0L)
88 val ex = new Exception("")
105 clock.plus(1.second)
106 val ex = new Exception("Very exception!")
89107 promise.failure(ex)
90 sourceFuture.futureValue
91 timer.histogram.collect(CollectionContext(10)).numberOfMeasurements should be(1L)
108 inside(Try(sourceFuture.futureValue)) {
109 case Failure(testFailEx: TestFailedException) =>
110 testFailEx.cause shouldBe Some(ex)
111 }
112
113 timer.histogram.collect(CollectionContext(10)).max shouldEqual (1000000000L +- 100000000L)
92114 }
93115 }
94116}