[citrix_adc] Improve timestamp parsing (elastic#11698)

- Improve parsing of various timestamps and improve handling of
custom date formats.
- Custom date formats are now handled first, falling back to a
standard set of patterns using a date processor otherwise.
- Consolidated parsing of timestamps to default pipeline.

packages/citrix_adc/_dev/deploy/docker/sample_logs/citrix-adc.log

@@ -15,4 +15,3 @@ Oct 6 14:03:30 <local0.info> 81.2.69.144 10/06/2014:14:03:30 GMT ns1 0-PPE-0 : T
 Oct 6 14:03:30 <local0.info> 81.2.69.144 10/06/2014:14:03:30 GMT ns1 0-PPE-0 : TCP CONN_TERMINATE 4473 0 : Source 127.0.0.1:7776 - Destination 127.0.0.2:55623 - Start Time 10/06/2014:14:02:45 GMT - End Time 10/06/2014:14:03:30 GMT - Total_bytes_send 1 - Total_bytes_recv 1
 Oct 6 14:03:30 <local0.info> 81.2.69.144 10/06/2014:14:03:30 GMT ns1 0-PPE-0 : TCP CONN_TERMINATE 4474 0 : Source 127.0.0.1:80 - Destination 127.0.0.2:39771 - Start Time 10/06/2014:14:02:46 GMT - End Time 10/06/2014:14:03:30 GMT - Total_bytes_send 1 - Total_bytes_recv 1
 <131> 10/08/2024:09:38:41  SYSLOGHOST 0-PPE-1 : default TCP CONN_DELINK 6715345 0 :  Source 127.1.2.1:80 - Vserver 81.2.69.144:80 - NatIP 192.168.10.10:52187 - Destination 81.2.69.144:80 - Delink Time 10/08/2024:09:38:41  - Total_bytes_send 0 - Total_bytes_recv 3118
-<131> 21/08/2024:09:38:41  SYSLOGHOST 0-PPE-1 : default TCP CONN_DELINK 6715345 0 :  Source 127.1.2.1:80 - Vserver 81.2.69.144:80 - NatIP 192.168.10.10:52187 - Destination 81.2.69.144:80 - Delink Time 10/08/2024:09:38:41  - Total_bytes_send 0 - Total_bytes_recv 3118

packages/citrix_adc/changelog.yml

@@ -1,4 +1,9 @@
 # newer versions go on top
+- version: "1.11.0"
+  changes:
+    - description: "Improve timestamp parsing"
+      type: bugfix
+      link: https://github.com/elastic/integrations/pull/11698
 - version: "1.10.0"
   changes:
     - description: "Parse additional sslvpn fields"

packages/citrix_adc/data_stream/log/_dev/test/pipeline/test-citrix-native-with-delink.json-expected.json

@@ -45,7 +45,7 @@
             ]
         },
         {
-            "@timestamp": "2024-08-10:38:41.000Z",
+            "@timestamp": "2024-08-10T09:38:41.000Z",
             "citrix": {
                 "cef_format": false,
                 "default_class": true,
@@ -59,7 +59,7 @@
             },
             "citrix_adc": {
                 "log": {
-                    "delink_time": "2024-08-10T09:38:41",
+                    "delink_time": "2024-08-10T09:38:41.000Z",
                     "destination": {
                         "ip": "81.2.69.144",
                         "port": 80
@@ -93,7 +93,7 @@
                 "category": [
                     "network"
                 ],
-                "end": "2024-08-10T09:38:41",
+                "end": "2024-08-10T09:38:41.000Z",
                 "id": "6715345",
                 "original": "<131> 10/08/2024:09:38:41  SYSLOGHOST 0-PPE-1 : default TCP CONN_DELINK 6715345 0 :  Source 127.1.2.1:80 - Vserver 81.2.69.144:80 - NatIP 192.168.10.10:52187 - Destination 81.2.69.144:80 - Delink Time 10/08/2024:09:38:41  - Total_bytes_send 0 - Total_bytes_recv 3118\n",
                 "severity": 0,
@@ -192,7 +192,7 @@
             },
             "citrix_adc": {
                 "log": {
-                    "delink_time": "2024-08-21T09:38:41",
+                    "delink_time": "2024-08-21T09:38:41.000Z",
                     "destination": {
                         "ip": "81.2.69.144",
                         "port": 80
@@ -226,7 +226,7 @@
                 "category": [
                     "network"
                 ],
-                "end": "2024-08-21T09:38:41",
+                "end": "2024-08-21T09:38:41.000Z",
                 "id": "6715345",
                 "original": "<131> 21/08/2024:09:38:41  SYSLOGHOST 0-PPE-1 : default TCP CONN_DELINK 6715345 0 :  Source 127.1.2.1:80 - Vserver 81.2.69.144:80 - NatIP 192.168.10.10:52187 - Destination 81.2.69.144:80 - Delink Time 21/08/2024:09:38:41  - Total_bytes_send 0 - Total_bytes_recv 3118\n",
                 "severity": 0,

packages/citrix_adc/data_stream/log/_dev/test/pipeline/test-citrix-native.json

@@ -75,6 +75,10 @@
 		{
             "@timestamp": "2024-08-21T09:38:41.000Z",
             "message": "<123> 21/08/2024:09:38:41  SYSLOGHOST 0-PPE-1 : default SSLVPN CHANGEME 600000 0 : \n"
+        },
+        {
+          "@timestamp": "2024-08-21T09:38:41.000Z",
+          "message": "<134> 21/08/2024:13:25:41  SYSLOGHOST 0-PPE-1 : default SSLVPN ICAEND_CONNSTAT 600000 0 : [TCP] [CGP][ICAUUID=00033ef4-29bb-172b-9678-0022480fced0] Source 67.43.156.1:50385 - Destination 10.0.10.75:2598 - customername  - username:domainname user_name:domain_name - startTime \"11/08/2024:10:37:13 \" - endTime \"11/08/2024:10:43:39 \" - Duration 00:06:26 - Total_bytes_send 8379078 - Total_bytes_recv 2761789 - Total_compressedbytes_send 0 - Total_compressedbytes_recv 0 - Compression_ratio_send 0.00% - Compression_ratio_recv 0.00% - connectionId 20459456 \n"
         }
 	]
 }

packages/citrix_adc/data_stream/log/_dev/test/pipeline/test-citrix-native.json-config.yml

@@ -0,0 +1,6 @@
+fields:
+  tags:
+    - preserve_original_event
+    - preserve_duplicate_custom_fields
+  _conf:
+    custom_date_format: "dd/MM/yyyy:HH:mm:ss"

packages/citrix_adc/data_stream/log/elasticsearch/ingest_pipeline/default.yml

@@ -33,7 +33,7 @@ processors:
       pattern_definitions:
         LEVEL: '<?%{IDENT:citrix.facility:keyword}\.%{IDENT:citrix.priority:keyword}>?'
         IDENT: '[a-zA-Z][a-zA-Z0-9]*'
-        SYSLOG_TIMESTAMP: '(?:%{SYSLOGTIMESTAMP:_tmp.timestamp}|%{TIMESTAMP_ISO8601:_tmp.timestamp8601})'
+        SYSLOG_TIMESTAMP: '(?:%{SYSLOGTIMESTAMP:_tmp.syslog_timestamp}|%{TIMESTAMP_ISO8601:_tmp.syslog_timestamp8601})'
         TIMESTAMP_ISO8601: '%{YEAR}-%{MONTHNUM}-%{MONTHDAY}[T ]%{HOUR}:?%{MINUTE}(?::?%{SECOND})?%{ISO8601_TIMEZONE:event.timezone}?'
   - pipeline:
       name: '{{ IngestPipeline "cef" }}'
@@ -48,6 +48,9 @@ processors:
       tag: convert_event_severity_to_long
       type: long
       ignore_missing: true
+
+# Time zone
+
   - set:
       field: _conf.tz_offset
       tag: set_tz_offset_utc
@@ -58,22 +61,33 @@ processors:
       tag: set_event_timezone_from_tz_offset
       copy_from: _conf.tz_offset
       if: ctx.event?.timezone == null || ctx.event?.timezone == ""
+
+# Syslog timestamp
+
   - date:
       if: ctx._tmp?.timestamp8601 != null
-      tag: date_tmp_timestamp8601
-      field: _tmp.timestamp8601
+      tag: date_syslog_timestamp8601
+      field: _tmp.syslog_timestamp8601
       timezone: '{{{event.timezone}}}'
       formats:
         - ISO8601
+      on_failure:
+        - append:
+            field: error.message
+            value: 'Processor {{{_ingest.on_failure_processor_type}}} with tag {{{_ingest.on_failure_processor_tag}}} in pipeline {{{_ingest.on_failure_pipeline}}} failed with message: {{{_ingest.on_failure_message}}}'
   - set:
-      field: _tmp.timestamp
-      tag: set_tmp_timestamp
-      if: ctx._tmp?.timestamp != null && ctx.citrix?.event_year != null
-      value: "{{{citrix.event_year}}} {{{_tmp.timestamp}}}"
+      field: _tmp.syslog_timestamp
+      tag: enrich_syslog_timestamp_with_year
+      if: ctx._tmp?.syslog_timestamp != null && ctx.citrix?.event_year != null
+      value: "{{{citrix.event_year}}} {{{_tmp.syslog_timestamp}}}"
+  - remove:
+      field: citrix.event_year
+      tag: remove_event_year
+      ignore_missing: true
   - date:
-      if: ctx._tmp?.timestamp != null
-      tag: date_tmp_timestamp
-      field: _tmp.timestamp
+      if: ctx._tmp?.syslog_timestamp != null
+      tag: date_syslog_timestamp
+      field: _tmp.syslog_timestamp
       timezone: '{{{event.timezone}}}'
       formats:
         - MMM d HH:mm:ss
@@ -88,35 +102,168 @@ processors:
         - yyyy MMMM d HH:mm:ss
         - yyyy MMMM  d HH:mm:ss
         - yyyy MMMM dd HH:mm:ss
+      on_failure:
+        - append:
+            field: error.message
+            value: 'Processor {{{_ingest.on_failure_processor_type}}} with tag {{{_ingest.on_failure_processor_tag}}} in pipeline {{{_ingest.on_failure_pipeline}}} failed with message: {{{_ingest.on_failure_message}}}'
+
+  - script:
+      tag: trim_time_fields
+      lang: painless
+      params:
+        fields:
+          - timestamp_native
+          - timestamp
+          - delink_time
+          - start_time
+          - end_time
+      source: |-
+        params.fields.forEach(field -> {
+          if (!ctx._tmp?.containsKey(field) || !(ctx._tmp[field] instanceof String)) {
+            return true;
+          }
+
+          String val = ctx._tmp[field];
+          ctx._tmp[field] = val.trim();
+        });
+
   - script:
-      description: Convert timestamp_native via custom format
-      tag: date_tmp_timestamp_native_custom_format
+      tag: parse_custom_time_formats
+      if: ctx._conf?.custom_date_format != null
       lang: painless
-      source: >-
-        def dateFormat = ctx["_conf"]["custom_date_format"];
-        def formatter = DateTimeFormatter.ofPattern(dateFormat);
-        def text = ctx["_tmp"]["timestamp_native"];
-        def parsedDate = LocalDateTime.parse(text, formatter);
-        ctx["_tmp"]["timestamp_native"] = parsedDate.toString();
-
-      if: ctx._tmp?.timestamp_native != null && ctx?._conf?.custom_date_format != null && ctx?._conf?.custom_date_format != ''
+      params:
+        fields:
+          - timestamp_native
+          - timestamp
+          - delink_time
+          - start_time
+          - end_time
+      source: |-
+        def zone = ctx.event?.timezone != null ? ZoneId.of(ctx.event.timezone) : null;
+        def formatter = DateTimeFormatter.ofPattern(ctx._conf.custom_date_format);
+        def outFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSSXXX");
+
+        params.fields.forEach(field -> {
+          if (!ctx._tmp?.containsKey(field)) {
+            return true;
+          }
+
+          try {
+            def localDateTime = LocalDateTime.parse(ctx._tmp[field], formatter);
+            ctx.citrix_adc.log[field] = outFormatter.format(ZonedDateTime.of(localDateTime, zone));
+          } catch (Exception e) {
+            /* Intentionally ignored */
+            return true;
+          }
+        });
+
+# Native-format timestamp
+
+  - date:
+      tag: date_timestamp_native
+      field: _tmp.timestamp_native
+      target_field: citrix_adc.log.timestamp_native
+      if: ctx._tmp?.timestamp_native != null && ctx.citrix_adc?.log?.timestamp_native == null
+      formats:
+        - ISO8601
+        - yyyy/MM/dd:HH:mm:ss
+        - yyyy/MM/dd:HH:mm:ss z
+        - MM/dd/yyyy:HH:mm:ss
+        - MM/dd/yyyy:HH:mm:ss z
       on_failure:
         - append:
             field: error.message
-            value: 'Processor {{{_ingest.on_failure_processor_type}}} with tag {{{_ingest.on_failure_processor_tag}}} in pipeline {{{_ingest.on_failure_pipeline}}} failed with message: {{{_ingest.on_failure_message}}}. Offending format: {{{_conf.custom_date_format}}}.'
+            value: 'Processor {{{_ingest.on_failure_processor_type}}} with tag {{{_ingest.on_failure_processor_tag}}} in pipeline {{{_ingest.on_failure_pipeline}}} failed with message: {{{_ingest.on_failure_message}}}'
   - date:
-      if: ctx._tmp?.timestamp_native != null && ctx?._conf?.custom_date_format == null
-      tag: date_tmp_timestamp_native
-      field: _tmp.timestamp_native
+      tag: date_timestamp
+      field: _tmp.timestamp
+      target_field: citrix_adc.log.timestamp
+      if: ctx._tmp?.timestamp != null && ctx.citrix_adc?.log?.timestamp == null
       formats:
+        - ISO8601
+        - yyyy/MM/dd:HH:mm:ss
+        - yyyy/MM/dd:HH:mm:ss z
         - MM/dd/yyyy:HH:mm:ss
+        - MM/dd/yyyy:HH:mm:ss z
+      on_failure:
+        - append:
+            field: error.message
+            value: 'Processor {{{_ingest.on_failure_processor_type}}} with tag {{{_ingest.on_failure_processor_tag}}} in pipeline {{{_ingest.on_failure_pipeline}}} failed with message: {{{_ingest.on_failure_message}}}'
+  - date:
+      tag: date_start_time
+      field: _tmp.start_time
+      target_field: citrix_adc.log.start_time
+      if: ctx._tmp?.start_time != null && ctx.citrix_adc?.log?.start_time == null
+      formats:
+        - ISO8601
         - yyyy/MM/dd:HH:mm:ss
-        - dd/MM/yyyy:HH:mm:ss
-      timezone: '{{{event.timezone}}}'
+        - yyyy/MM/dd:HH:mm:ss z
+        - MM/dd/yyyy:HH:mm:ss
+        - MM/dd/yyyy:HH:mm:ss z
+      on_failure:
+        - append:
+            field: error.message
+            value: 'Processor {{{_ingest.on_failure_processor_type}}} with tag {{{_ingest.on_failure_processor_tag}}} in pipeline {{{_ingest.on_failure_pipeline}}} failed with message: {{{_ingest.on_failure_message}}}'
+  - date:
+      tag: date_end_time
+      field: _tmp.end_time
+      target_field: citrix_adc.log.end_time
+      if: ctx._tmp?.end_time != null && ctx.citrix_adc?.log?.end_time == null
+      formats:
+        - ISO8601
+        - yyyy/MM/dd:HH:mm:ss
+        - yyyy/MM/dd:HH:mm:ss z
+        - MM/dd/yyyy:HH:mm:ss
+        - MM/dd/yyyy:HH:mm:ss z
+      on_failure:
+        - append:
+            field: error.message
+            value: 'Processor {{{_ingest.on_failure_processor_type}}} with tag {{{_ingest.on_failure_processor_tag}}} in pipeline {{{_ingest.on_failure_pipeline}}} failed with message: {{{_ingest.on_failure_message}}}'
+  - date:
+      tag: date_delink_time
+      field: _tmp.delink_time
+      target_field: citrix_adc.log.delink_time
+      if: ctx._tmp?.delink_time != null && ctx.citrix_adc?.log?.delink_time == null
+      formats:
+        - ISO8601
+        - yyyy/MM/dd:HH:mm:ss
+        - yyyy/MM/dd:HH:mm:ss z
+        - MM/dd/yyyy:HH:mm:ss
+        - MM/dd/yyyy:HH:mm:ss z
+      on_failure:
+        - append:
+            field: error.message
+            value: 'Processor {{{_ingest.on_failure_processor_type}}} with tag {{{_ingest.on_failure_processor_tag}}} in pipeline {{{_ingest.on_failure_pipeline}}} failed with message: {{{_ingest.on_failure_message}}}'
+
+# Move vendor time fields to ECS.
+
+  - set:
+      tag: set_@timestamp_from_citrix_native
+      field: '@timestamp'
+      copy_from: citrix_adc.log.timestamp_native
+      ignore_empty_value: true
   - remove:
-      field: citrix.event_year
-      tag: remove_event_year
+      tag: remove_citrix_timestamp_native
+      field: citrix_adc.log.timestamp_native
       ignore_missing: true
+
+  - set:
+      tag: set_event_start_from_citrix_start_time
+      field: event.start
+      copy_from: citrix_adc.log.start_time
+      ignore_empty_value: true
+
+  - set:
+      tag: set_event_end_from_citrix_delink_time
+      field: event.end
+      copy_from: citrix_adc.log.delink_time
+      ignore_empty_value: true
+  - set:
+      tag: set_event_end_from_citrix_end_time
+      field: event.end
+      copy_from: citrix_adc.log.end_time
+      ignore_empty_value: true
+
   - geoip:
       field: client.ip
       tag: geoip_client_ip_to_client_geo