ULS v1.5 (#28)

Version 1.5.0

Author: bitonio

.gitignore

@@ -1 +1,3 @@
 .DS_Store
+__pycache__
+/ext/*

Dockerfile

@@ -1,4 +1,4 @@
-FROM            python:3.10.4-slim-bullseye
+FROM            python:3.10.5-slim-bullseye
 LABEL 	        MAINTAINER="Mike Schiessl - [email protected]"
 LABEL	        APP_LONG="Akamai Universal Log Streamer"
 LABEL           APP_SHORT="ULS"
@@ -11,7 +11,7 @@ ARG             ULS_DIR="$HOMEDIR/uls"
 ARG             EXT_DIR="$ULS_DIR/ext"
 
 ARG             ETP_CLI_VERSION="0.3.8"
-ARG             EAA_CLI_VERSION="0.5.0"
+ARG             EAA_CLI_VERSION="0.5.0.1"
 ARG             MFA_CLI_VERSION="0.0.9"
 
 # ENV VARS

bin/config/global_config.py

@@ -1,7 +1,7 @@
 #!/usr/bin/env python3
 
 # Common global variables / constants
-__version__ = "1.4.0"
+__version__ = "1.5.0"
 __tool_name_long__ = "Akamai Unified Log Streamer"
 __tool_name_short__ = "ULS"
 
@@ -52,6 +52,7 @@
 input_rerun_delay = 1                           # Time in seconds between rerun attempts
 input_disable_stderr = True                     # Enable STDERR output disabling (see value below to specify when this should happen)
 input_disable_stderr_after = 25                 # Disable stderr output after x input_cli cycles --> to prevent buffer overflow
+input_queue_size = 10000                        # Maximum number of events we want to store in-memory, default is 10000
 
 # OUTPUT Configuration
 output_reconnect_retries = 10                   # Number of reconnect attempts before giving up
@@ -64,11 +65,13 @@
     ## HTTP
 output_http_header = {'User-Agent': f'{__tool_name_long__}/{__version__}'}  # HTTP Additional Headers to send (requests module KV pairs)
 output_http_timeout = 10                        # Timeout after which a request will be considered as failed
+output_http_aggregate_count = 50                # Number of events to aggregate in POST request to HTTP Collector. 1 mean no aggregation
+output_http_aggregate_idle = 5                  # Aggregate will send the data regardless of the count if the previous event was x secs ago
     ## FILE
 output_file_encoding = "utf-8"                  # FILE Encoding setting
 output_file_handler_choices = ['SIZE', 'TIME']  # Available Choices for the file handler
 output_file_default_backup_count = 3                # Default number of backup files (after rotation)
-output_file_default_maxbytes = 50 * 1024 * 1024      # Default maximum size of a file when rotated by the FILE - handler
+output_file_default_maxbytes = 50 * 1024 * 1024     # Default maximum size of a file when rotated by the FILE - handler
 output_file_default_time_use_utc = False            # Use UTC instead of local system time (Default: False)
 output_file_time_choices = ['S','M','H','D','W0','W1','W2','W3','W4','W5','W6','MIDNIGHT']      # Available choices for the time unit
 output_file_time_default = 'M'                      # Default value for the time unit (Minutes)
@@ -83,10 +86,10 @@
 edgerc_eaa_legacy = ["eaa_api_host", "eaa_api_key", "eaa_api_secret"]               # required for EAA - Legacy
 edgerc_mfa = ["mfa_integration_id", "mfa_signing_key"]                              # Required for MFA
 edgerc_documentation_url = "https://github.com/akamai/uls/blob/main/docs/AKAMAI_API_CREDENTIALS.md"
-edgerc_mock_file = "ext/edgerc"                 # Required for display the version if no edgercfile was given
+edgerc_mock_file = "ext/edgerc"                  # Required for display the version if no edgercfile was given
 
 # Autoresume Configuration
 autoresume_checkpoint_path = "var/"              # (Default) Path, where the checkpointfiles should be stored to
-autoresume_supported_inputs = ['ETP', 'EAA']           # Internal Var only, to adjust supported inputs
+autoresume_supported_inputs = ['ETP', 'EAA']     # Internal Var only, to adjust supported inputs
 autoresume_write_after = 1000                    # Write checkpoint only every ${autoresume_write_every} loglines
 

bin/modules/UlsInputCli.py

@@ -17,6 +17,8 @@
 import time
 import shlex
 import os
+import threading
+import queue
 
 # ULS modules
 import config.global_config as uls_config
@@ -272,9 +274,10 @@ def proc_create(self):
                 self.proc = self.cli_proc
                 self.proc_output = self.cli_proc.stdout
 
-                # Unblocking on windows causes trouble so we're avoiding it
-                if not os.name == 'nt':
-                    os.set_blocking(self.proc_output.fileno(), False)
+                # Non-blocking on windows causes trouble so we're avoiding it
+                # 2022-07-08: Disabled completely see EME-588
+                # if not os.name == 'nt':
+                #     os.set_blocking(self.proc_output.fileno(), False)
 
                 time.sleep(1)
 
@@ -338,4 +341,49 @@ def check_proc(self):
             self.proc_create()
             return False
 
+    def ingest(self, stopEvent, event_q, monitor):
+        """
+        Ingest CLI incoming data asynchronously
+        The function will return immediately
+        Args:
+            stopEvent (_type_): Stop Event that will stop the current thread
+            event_q (_type_): The queue where to put the message
+            monitor (): Monitor
+        """
+        self.stopEvent = stopEvent
+        self.event_queue = event_q
+        self.monitor = monitor
+        self.ingest_thread = threading.Thread(target=self.ingest_loop)
+        self.ingest_thread.setName("ingest_loop")
+        self.ingest_thread.start()
+
+    def ingest_loop(self):
+
+        # When reading CLI output, if no data, pause 10ms before retrying
+        # if still no data, then it will backoff exponentially till 60s
+        wait_default = uls_config.main_wait_default
+        wait_max = uls_config.main_wait_max
+        wait = wait_default
+
+        while not self.stopEvent.is_set():
+            try:
+
+                # Ensure the Input handler is still running (rebump if stale)
+                self.check_proc()
+
+                input_data = self.proc_output.readline()
+                if input_data:
+                    self.event_queue.put(input_data, timeout=0.05)
+                    self.monitor.increase_message_ingested()
+                    wait = wait_default  # back to 10ms wait in case of bursty content
+                else:
+                    aka_log.log.debug(f"ingest_loop, wait {wait} seconds [{self.monitor.get_stats()}]")
+                    self.stopEvent.wait(wait)
+                    wait = min(wait * 2, wait_max)  # double the wait till a max of 60s
+            except queue.Full:
+                aka_log.log.fatal("Capacity exceeded, too many incoming data vs. slow output")
+                self.stopEvent.set()
+            except Exception:
+                aka_log.log.exception("Error in ingest_loop")
+
 # EOF

bin/modules/UlsMonitoring.py

@@ -51,6 +51,8 @@ def __init__(self, stopEvent, product, feed, output):
         self.name = "UlsMonitoring"                          # Class Human readable name
         self.overall_messages_handled = 0                    # Define overall number of messages handled
         self.window_messages_handled = 0                     # Define mon_window number of messages handled
+        self.window_messages_bytes = 0                       # Total bytes processed during the window
+        self.window_messages_ingested = 0                    # Message ingested from UlsInputCli module
         self.init_time = time.time()                         # Define the init time
 
         # Define the working thread, daemon allows us to offload
@@ -84,6 +86,8 @@ def display(self):
                        'uls_runtime': self._runtime(),
                        'event_count': self.overall_messages_handled,
                        'event_count_interval': self.window_messages_handled,
+                       'event_ingested_interval': self.window_messages_ingested,
+                       'event_bytes_interval': self.window_messages_bytes,
                        'event_rate': round(self.window_messages_handled / self.monitoring_interval, 2),
                        'mon_interval': self.monitoring_interval
                     }
@@ -94,13 +98,21 @@ def display(self):
                     # Reset window based vars
                     with self._metricLock:
                         self.window_messages_handled = 0
+                        self.window_messages_bytes = 0
+                        self.window_messages_ingested = 0
         except Exception as e:
             aka_log.log.exception(e)
 
-    def increase_message_count(self):
+    def increase_message_count(self, bytes=0):
         with self._metricLock:
             self.overall_messages_handled = self.overall_messages_handled + 1
             self.window_messages_handled = self.window_messages_handled + 1
+            self.window_messages_bytes += bytes
+
+    def increase_message_ingested(self):
+        with self._metricLock:
+            self.window_messages_ingested += 1
+
 
     def get_message_count(self):
         return self.overall_messages_handled

bin/modules/UlsOutput.py

@@ -21,6 +21,7 @@
 import requests
 import logging
 import logging.handlers
+import random
 
 # ULS specific modules
 import config.global_config as uls_config
@@ -93,6 +94,12 @@ def __init__(self, output_type: str,
 
         # HTTP parameters
         elif self.output_type in ['HTTP'] and http_url:
+
+            # ---- Begin change for EME-588 ----
+            self.aggregateList = list()
+            self.aggregateListTick = None # Last time we added items in the list
+            # ---- End change for EME-588 ----
+
             self.http_url = http_url
             # apply other variables if SET
 
@@ -375,7 +382,7 @@ def send_data(self, data):
         :return: True on successful send, False on error
         """
         try:
-            aka_log.log.info(f"{self.name} Trying to send data via {self.output_type}")
+            aka_log.log.debug(f"{self.name} Trying to send data via {self.output_type}")
 
             if self.output_type == "TCP":
                 self.clientSocket.sendall(data)
@@ -384,30 +391,41 @@ def send_data(self, data):
                 self.clientSocket.sendto(data, (self.host, self.port))
 
             elif self.output_type == "HTTP":
-                response = self.httpSession.post(url=self.http_url,
-                                                 data=self.http_out_format % (data.decode()),
-                                                 verify=self.http_verify_tls,
-                                                 timeout=self.http_timeout)
-                aka_log.log.debug(f"{self.name} DATA Send response {response.status_code},"
-                                  f" {response.text} ")
+                self.aggregateList.append(data)
+                if len(self.aggregateList) == uls_config.output_http_aggregate_count or (
+                    self.aggregateListTick is not None and
+                    self.aggregateListTick < time.time() - uls_config.output_http_aggregate_idle
+                ):
+                    data = uls_config.output_line_breaker.join(
+                        self.http_out_format % (event.decode()) for event in self.aggregateList)
+                    request = requests.Request('POST', url=self.http_url, data=data)
+                    prepped = self.httpSession.prepare_request(request)
+                    payload_length = prepped.headers["Content-Length"]
+                    response = self.httpSession.send(prepped, verify=self.http_verify_tls, timeout=self.http_timeout)
+                    response.close()  # Free up the underlying TCP connection in the connection pool
+                    aka_log.log.info(f"{self.name} HTTP POST of {len(self.aggregateList)} event(s) "
+                                     f"completed in {(response.elapsed.total_seconds()*1000):.3f} ms, "
+                                     f"payload={payload_length} bytes, HTTP response {response.status_code}, "
+                                     f"response={response.text} ")
+                    self.aggregateList.clear()
+                self.aggregateListTick = time.time()
 
             elif self.output_type == "RAW":
                 sys.stdout.write(data.decode())
                 sys.stdout.flush()
 
-
             elif self.output_type == "FILE":
                 self.my_file_writer.info(f"{data.decode().rstrip()}")
 
             else:
                 aka_log.log.critical(f"{self.name} target was not defined {self.output_type} ")
                 sys.exit(1)
 
-            aka_log.log.info(f"{self.name} Data successfully sent via {self.output_type}")
+            aka_log.log.debug(f"{self.name} Data successfully sent via {self.output_type}")
             return True
 
         except Exception as my_error:
-            aka_log.log.error(f"{self.name} Issue sending data {my_error}")
+            aka_log.log.exception(f"{self.name} Issue sending data {my_error}")
             self.connected = False
             self.connect()
             return False