add event.ingested to APM events (elastic#3279)

jalvz · Mar 11, 2020 · af77bbc · af77bbc
1 parent 9cf16dd
commit af77bbc
Show file tree

Hide file tree

Showing 8 changed files with 126 additions and 14 deletions.
diff --git a/ingest/pipeline/definition.json b/ingest/pipeline/definition.json
@@ -35,6 +35,7 @@
             "ignore_failure": true
           }
         }
+<<<<<<< HEAD
       ]
     }
   },
@@ -52,6 +53,47 @@
             "on_failure": [
               {
                 "remove": {
+=======
+      },
+      {
+        "pipeline": {
+          "name": "apm_ingest_timestamp"
+        }
+      }
+    ]
+  }
+},
+{
+  "id": "apm_user_agent",
+  "body": {
+    "description" : "Add user agent information for APM events",
+    "processors" : [
+      {
+        "user_agent" : {
+          "field": "user_agent.original",
+          "target_field": "user_agent",
+          "ignore_missing": true,
+          "ignore_failure": true
+        }
+      }
+    ]
+  }
+},
+{
+  "id": "apm_user_geo",
+  "body": {
+    "description" : "Add user geo information for APM events",
+    "processors" : [
+      {
+        "geoip" : {
+          "database_file": "GeoLite2-City.mmdb",
+          "field": "client.ip",
+          "target_field": "client.geo",
+          "ignore_missing": true,
+          "on_failure": [
+            {
+              "remove": {
+>>>>>>> 75589764... add event.ingested to APM events (#3279)
                   "field": "client.ip",
                   "ignore_missing": true,
                   "ignore_failure": true
@@ -91,4 +133,24 @@
       ]
     }
   }
-]
+<<<<<<< HEAD
+]
+=======
+},
+{
+  "id": "apm_ingest_timestamp",
+  "body": {
+    "description": "Add an ingest timestamp for APM events",
+    "processors": [
+      {
+        "set": {
+          "if": "ctx.processor?.event != 'span'",
+          "field": "event.ingested",
+          "value": "{{_ingest.timestamp}}"
+        }
+      }
+    ]
+  }
+}
+]
+>>>>>>> 75589764... add event.ingested to APM events (#3279)
diff --git a/tests/system/apmserver.py b/tests/system/apmserver.py
@@ -393,6 +393,12 @@ def get_doc_id(doc):
                 for appr in approved:
                     if get_doc_id(appr) == rec_id:
                         rec['observer'] = appr['observer']
+                        # ensure both docs have the same event keys set
+                        self.assertEqual(rec.get("event", {}).keys(), appr.get("event", {}).keys())
+                        # We don't compare the event values between received/approved
+                        # as they are dependent on the environment.
+                        if 'event' in rec:
+                            rec['event'] = appr['event']
                         break
             assert len(received) == len(approved)
             for i, rec in enumerate(received):

diff --git a/tests/system/error.approved.json b/tests/system/error.approved.json
@@ -31,6 +31,9 @@
             "us": 1494342245999000
         },
         "@timestamp": "2017-05-09T15:04:05.999Z",
+        "event": {
+            "ingested": "2017-05-09T15:04:05.998Z"
+        },
         "labels": {
             "tag1": "one",
             "tag2": 2
@@ -121,6 +124,9 @@
             "us": 1494342245999999
         },
         "@timestamp": "2017-05-09T15:04:05.999Z",
+        "event": {
+            "ingested": "2017-05-09T15:04:05.998Z"
+        },
         "labels": {
             "organization_uuid": "9f0e9d64-c185-4d21-a6f4-4673ed561ec8",
             "tag1": "one",
@@ -435,6 +441,9 @@
             "us": 1494342245000000
         },
         "@timestamp": "2017-05-09T15:04:05.000Z",
+        "event": {
+            "ingested": "2017-05-09T15:04:05.001Z"
+        },
         "labels": {
             "tag1": "one",
             "tag2": 2
@@ -519,6 +528,9 @@
             "us": 1494342245000000
         },
         "@timestamp": "2017-05-09T15:04:05.000Z",
+        "event": {
+            "ingested": "2017-05-09T15:04:05.001Z"
+        },
         "labels": {
             "tag1": "one",
             "tag2": 2
@@ -572,4 +584,4 @@
             "event": "error"
         }
     }
-]
+]
diff --git a/tests/system/jaeger_batch_0.approved.json b/tests/system/jaeger_batch_0.approved.json
@@ -25,6 +25,9 @@
             "us": 1576827704953864
         },
         "@timestamp": "2019-12-20T07:41:44.953Z",
+        "event": {
+            "ingested": "2019-12-20T07:41:45.954Z"
+        },
         "labels": {
             "peer_port": 50535,
             "as": "thrift",
@@ -83,6 +86,9 @@
             "us": 1576827705007552
         },
         "@timestamp": "2019-12-20T07:41:45.007Z",
+        "event": {
+            "ingested": "2019-12-20T07:41:45.008Z"
+        },
         "agent": {
             "ephemeral_id": "624386e9c81d2980",
             "name": "Jaeger/Go",
@@ -144,6 +150,9 @@
             "us": 1576827705089431
         },
         "@timestamp": "2019-12-20T07:41:45.089Z",
+        "event": {
+            "ingested": "2019-12-20T07:41:45.090Z"
+        },
         "agent": {
             "ephemeral_id": "624386e9c81d2980",
             "name": "Jaeger/Go",
@@ -205,6 +214,9 @@
             "us": 1576827705172530
         },
         "@timestamp": "2019-12-20T07:41:45.172Z",
+        "event": {
+            "ingested": "2019-12-20T07:41:45.173Z"
+        },
         "agent": {
             "ephemeral_id": "624386e9c81d2980",
             "name": "Jaeger/Go",
@@ -243,4 +255,4 @@
             "event": "error"
         }
     }
-]
+]
diff --git a/tests/system/jaeger_span.approved.json b/tests/system/jaeger_span.approved.json
@@ -25,6 +25,9 @@
             "us": 1578451731616515
         },
         "@timestamp": "2020-01-08T02:48:51.616Z",
+        "event": {
+            "ingested": "2020-01-08T02:48:51.617Z"
+        },
         "labels": {
             "sampler_type": "const",
             "sampler_param": true
@@ -56,4 +59,4 @@
             "event": "transaction"
         }
     }
-]
+]
diff --git a/tests/system/test_integration.py b/tests/system/test_integration.py
@@ -67,14 +67,12 @@ def test_load_docs_with_template_and_add_transaction(self):
         self.assert_no_logged_warnings()
 
         # compare existing ES documents for transactions with new ones
-        rs = self.es.search(index=index_transaction)
-        assert rs['hits']['total']['value'] == 4, "found {} documents".format(rs['count'])
-        self.approve_docs('transaction', rs['hits']['hits'])
+        transaction_docs = self.wait_for_events('transaction', 4, index=index_transaction)
+        self.approve_docs('transaction', transaction_docs)
 
         # compare existing ES documents for spans with new ones
-        rs = self.es.search(index=index_span)
-        assert rs['hits']['total']['value'] == 5, "found {} documents".format(rs['count'])
-        self.approve_docs('spans', rs['hits']['hits'])
+        span_docs = self.wait_for_events('transaction', 5, index=index_span)
+        self.approve_docs('spans', span_docs)
 
     def test_load_docs_with_template_and_add_error(self):
         """
@@ -85,9 +83,8 @@ def test_load_docs_with_template_and_add_error(self):
         self.assert_no_logged_warnings()
 
         # compare existing ES documents for errors with new ones
-        rs = self.es.search(index=index_error)
-        assert rs['hits']['total']['value'] == 4, "found {} documents".format(rs['count'])
-        self.approve_docs('error', rs['hits']['hits'])
+        error_docs = self.wait_for_events('error', 4, index=index_error)
+        self.approve_docs('error', error_docs)
 
         self.check_backend_error_sourcemap(index_error, count=4)
 

diff --git a/tests/system/test_pipelines.py b/tests/system/test_pipelines.py
@@ -1,3 +1,5 @@
+import datetime
+
 from apmserver import ElasticTest, SubCommandTest, TimeoutError, integration_test
 from elasticsearch import Elasticsearch
 from nose.tools import raises
@@ -67,6 +69,12 @@ def test_pipeline_registered_and_applied(self):
         ua_found = False
         for e in entries:
             src = e['_source']
+
+            # ingest timestamp pipeline
+            ingest_ts = src.get("event", {}).get("ingested")
+            assert datetime.datetime.strptime(ingest_ts[:19], "%Y-%m-%dT%H:%M:%S")
+
+            # user agent pipeline
             if 'user_agent' in src:
                 ua_found = True
                 ua = src['user_agent']

diff --git a/tests/system/transaction.approved.json b/tests/system/transaction.approved.json
@@ -44,6 +44,9 @@
             "us": 1496170422281000
         },
         "@timestamp": "2017-05-30T18:53:42.281Z",
+        "event": {
+            "ingested": "2017-05-30T18:53:42.282Z"
+        },
         "agent": {
             "version": "3.14.0",
             "name": "elastic-node"
@@ -136,6 +139,9 @@
             "us": 1496170422000000
         },
         "@timestamp": "2017-05-30T18:53:42.000Z",
+        "event": {
+            "ingested": "2017-05-30T18:53:42.001Z"
+        },
         "agent": {
             "version": "3.14.0",
             "name": "elastic-node"
@@ -229,6 +235,9 @@
             "us": 1496170422281999
         },
         "@timestamp": "2017-05-30T18:53:42.281Z",
+        "event": {
+            "ingested": "2017-05-30T18:53:42.282Z"
+        },
         "agent": {
             "version": "3.14.0",
             "name": "elastic-node"
@@ -315,6 +324,9 @@
             "us": 1496170407154000
         },
         "@timestamp": "2017-05-30T18:53:27.154Z",
+        "event": {
+            "ingested": "2017-05-30T18:53:27.155Z"
+        },
         "labels": {
             "organization_uuid": "9f0e9d64-c185-4d21-a6f4-4673ed561ec8",
             "bool_error": false,
@@ -485,4 +497,4 @@
             "email": "[email protected]"
         }
     }
-]
+]