Add guard against invalid end timestamps (elastic#3363)

CHANGELOG.asciidoc

@@ -31,6 +31,10 @@ Use subheadings with the "=====" level for adding notes for unreleased changes:
 
 === Unreleased
 
+[float]
+===== Features
+* Added protection against invalid timestamps provided by manual instrumentation - {pull}3363[#3363]
+
 [float]
 ===== Bug fixes
 * Fixed too many spans being created for `HTTPUrlConnection` requests with method `HEAD` - {pull}3353[#3353]

apm-agent-core/src/main/java/co/elastic/apm/agent/impl/transaction/AbstractSpan.java

@@ -559,6 +559,13 @@ public void end() {
 
     public final void end(long epochMicros) {
         if (!finished) {
+
+            long startTime = timestamp.get();
+            if(epochMicros < startTime) {
+                logger.warn("End called on {} with a timestamp before start! Using start timestamp as end instead.", this);
+                epochMicros = startTime;
+            }
+
             this.endTimestamp.set(epochMicros);
             childDurations.onSpanEnd(epochMicros);
 

apm-agent-core/src/test/java/co/elastic/apm/agent/impl/ElasticApmTracerTest.java

@@ -505,6 +505,20 @@ void testTimestamps() {
         assertThat(span.getDuration()).isEqualTo(10);
     }
 
+    @Test
+    void testTimestampSanitization() {
+        final Transaction transaction = tracerImpl.startChildTransaction(new HashMap<>(), TextHeaderMapAccessor.INSTANCE, ConstantSampler.of(true), 10, null);
+        final Span span = transaction.createSpan(20);
+        span.end(10);
+
+        transaction.end(5);
+
+        assertThat(transaction.getTimestamp()).isEqualTo(10);
+        assertThat(transaction.getDuration()).isEqualTo(0);
+        assertThat(span.getTimestamp()).isEqualTo(20);
+        assertThat(span.getDuration()).isEqualTo(0);
+    }
+
     @Test
     void testStartSpanAfterTransactionHasEnded() {
         final Transaction transaction = startTestRootTransaction();