proactively drop old spans in the APM reporter

Author: ivantopo

reporters/kamon-apm-reporter/src/main/scala/kamon/apm/KamonApm.scala

@@ -83,20 +83,24 @@ class KamonApm(configPath: String) extends CombinedReporter {
 
   override def reportSpans(spans: Seq[Span.Finished]): Unit = if(spans.nonEmpty) {
     if(isAcceptableApiKey(_settings.apiKey)) {
-      val env = Kamon.environment
-      val apmSpans = spans map convertSpan
-
-      val batch = SpanBatch.newBuilder()
-        .setAgent(_settings.agent)
-        .setServiceName(env.service)
-        .setHost(env.host)
-        .setInstance(env.instance)
-        .setApiKey(_settings.apiKey)
-        .addAllSpans(apmSpans.asJava)
-        .build()
-
-      _httpClient.foreach(_.postSpans(batch))
+      val spansFinishLimit = Kamon.clock().instant().minus(_maxSnapshotAge)
+      val recentSpans = spans.filter(_.to.isAfter(spansFinishLimit))
+
+      if(recentSpans.nonEmpty) {
+        val env = Kamon.environment
+        val apmSpans = recentSpans map convertSpan
+
+        val batch = SpanBatch.newBuilder()
+          .setAgent(_settings.agent)
+          .setServiceName(env.service)
+          .setHost(env.host)
+          .setInstance(env.instance)
+          .setApiKey(_settings.apiKey)
+          .addAllSpans(apmSpans.asJava)
+          .build()
 
+        _httpClient.foreach(_.postSpans(batch))
+      }
     } else
       _logger.error(s"Dropping Spans because an invalid API key has been configured [${_settings.apiKey}]")
   }

reporters/kamon-apm-reporter/src/test/scala/kamon/apm/ReporterSpec.scala

@@ -103,7 +103,7 @@ class ReporterSpec extends TestKit(ActorSystem("MetricReporterSpec"))
 
     "retry span ingestion" in {
       val span = Span.Finished(Identifier.Empty, Trace.Empty, Identifier.Empty, "", false, false,
-        Instant.ofEpochMilli(0), Instant.ofEpochMilli(0), Span.Kind.Unknown, Span.Position.Unknown,
+        Instant.now(), Instant.now(), Span.Kind.Unknown, Span.Position.Unknown,
         TagSet.Empty, TagSet.Empty, Seq.empty, Seq.empty)
 
       reporter.reportSpans(Seq(span))