Enhance debug logging for output client initialization and add environment variable aliases for debug settings

Should have caught this on previous fix for since. the current time is used on line 2145: connection.fetch_messages(reports_folder, since=current_time)
if that code is called and it usually won't be depending upon configuration it will fail  with the time format being wrong: yyyy-mm-ddThh:mm:ss.zzzzzz+00:00Z  ---     this removes the extra "Z" that is not needed since utc offset is already specified and becomes invalid.

.vscode/settings.json

@@ -26,11 +26,13 @@
       "boto",
       "brakhane",
       "Brightmail",
+      "cafile",
       "CEST",
       "CHACHA",
       "checkdmarc",
       "Codecov",
       "confnew",
+      "creds",
       "dateparser",
       "dateutil",
       "Davmail",
@@ -130,6 +132,7 @@
       "sdist",
       "Servernameone",
       "setuptools",
+      "signum",
       "smartquotes",
       "SMTPTLS",
       "sortlists",

CHANGELOG.md

@@ -1,5 +1,18 @@
 # Changelog
 
+## 9.5.5
+
+### Fixed
+
+- Output client initialization now retries up to 4 times with exponential backoff before exiting. This fixes persistent `Connection refused` errors in Docker when OpenSearch or Elasticsearch is momentarily unavailable at startup.
+- Use tuple format for `http_auth` in OpenSearch and Elasticsearch connections, matching the documented convention and avoiding potential issues if the password contains a colon.
+- Fix current_time format for MSGraphConnection (current-time) (PR #708)
+
+### Changes
+
+- Added debug logging to all output client initialization (S3, syslog, Splunk HEC, Kafka, GELF, webhook, Elasticsearch, OpenSearch).
+- `DEBUG=true` and `PARSEDMARC_DEBUG=true` are now accepted as short aliases for `PARSEDMARC_GENERAL_DEBUG=true`.
+
 ## 9.5.4
 
 ### Fixed

parsedmarc/__init__.py

@@ -1956,7 +1956,7 @@ def get_dmarc_reports_from_mailbox(
             current_time = datetime.now(timezone.utc).strftime("%d-%b-%Y")
         elif isinstance(connection, MSGraphConnection):
             since = (datetime.now(timezone.utc) - timedelta(minutes=_since)).isoformat()
-            current_time = datetime.now(timezone.utc).isoformat() + "Z"
+            current_time = datetime.now(timezone.utc).isoformat()
         elif isinstance(connection, GmailConnection):
             since = (datetime.now(timezone.utc) - timedelta(minutes=_since)).strftime(
                 "%s"

parsedmarc/cli.py

@@ -9,6 +9,7 @@ import logging
 import os
 import signal
 import sys
+import time
 from argparse import ArgumentParser, Namespace
 from configparser import ConfigParser
 from glob import glob
@@ -135,12 +136,20 @@ def _apply_env_overrides(config: ConfigParser) -> None:
     """
     prefix = "PARSEDMARC_"
 
-    for env_key, env_value in os.environ.items():
-        if not env_key.startswith(prefix) or env_key == "PARSEDMARC_CONFIG_FILE":
-            continue
+    # Short aliases that don't follow the PARSEDMARC_{SECTION}_{KEY} pattern.
+    _ENV_ALIASES = {
+        "DEBUG": ("general", "debug"),
+        "PARSEDMARC_DEBUG": ("general", "debug"),
+    }
 
-        suffix = env_key[len(prefix) :]
-        section, key = _resolve_section_key(suffix)
+    for env_key, env_value in os.environ.items():
+        if env_key in _ENV_ALIASES:
+            section, key = _ENV_ALIASES[env_key]
+        elif env_key.startswith(prefix) and env_key != "PARSEDMARC_CONFIG_FILE":
+            suffix = env_key[len(prefix) :]
+            section, key = _resolve_section_key(suffix)
+        else:
+            continue
 
         if section is None:
             logger.debug("Ignoring unrecognized env var: %s", env_key)
@@ -988,6 +997,7 @@ def _init_output_clients(opts):
 
     try:
         if opts.s3_bucket:
+            logger.debug("Initializing S3 client: bucket=%s", opts.s3_bucket)
             clients["s3_client"] = s3.S3Client(
                 bucket_name=opts.s3_bucket,
                 bucket_path=opts.s3_path,
@@ -1001,6 +1011,11 @@ def _init_output_clients(opts):
 
     try:
         if opts.syslog_server:
+            logger.debug(
+                "Initializing syslog client: server=%s:%s",
+                opts.syslog_server,
+                opts.syslog_port,
+            )
             clients["syslog_client"] = syslog.SyslogClient(
                 server_name=opts.syslog_server,
                 server_port=int(opts.syslog_port),
@@ -1025,6 +1040,7 @@ def _init_output_clients(opts):
                 "HEC token and HEC index are required when using HEC URL"
             )
         try:
+            logger.debug("Initializing Splunk HEC client: url=%s", opts.hec)
             verify = True
             if opts.hec_skip_certificate_verification:
                 verify = False
@@ -1036,6 +1052,7 @@ def _init_output_clients(opts):
 
     try:
         if opts.kafka_hosts:
+            logger.debug("Initializing Kafka client: hosts=%s", opts.kafka_hosts)
             ssl_context = None
             if opts.kafka_skip_certificate_verification:
                 logger.debug("Skipping Kafka certificate verification")
@@ -1053,6 +1070,11 @@ def _init_output_clients(opts):
 
     try:
         if opts.gelf_host:
+            logger.debug(
+                "Initializing GELF client: host=%s:%s",
+                opts.gelf_host,
+                opts.gelf_port,
+            )
             clients["gelf_client"] = gelf.GelfClient(
                 host=opts.gelf_host,
                 port=int(opts.gelf_port),
@@ -1067,6 +1089,7 @@ def _init_output_clients(opts):
             or opts.webhook_forensic_url
             or opts.webhook_smtp_tls_url
         ):
+            logger.debug("Initializing webhook client")
             clients["webhook_client"] = webhook.WebhookClient(
                 aggregate_url=opts.webhook_aggregate_url,
                 forensic_url=opts.webhook_forensic_url,
@@ -1079,11 +1102,16 @@ def _init_output_clients(opts):
     # Elasticsearch and OpenSearch mutate module-level global state via
     # connections.create_connection(), which cannot be rolled back if a later
     # step fails.  Initialise them last so that all other clients are created
-    # successfully first; this minimises the window for partial-init problems
+    # successfully first; this minimizes the window for partial-init problems
     # during config reload.
     if opts.save_aggregate or opts.save_forensic or opts.save_smtp_tls:
         try:
             if opts.elasticsearch_hosts:
+                logger.debug(
+                    "Initializing Elasticsearch client: hosts=%s, ssl=%s",
+                    opts.elasticsearch_hosts,
+                    opts.elasticsearch_ssl,
+                )
                 es_aggregate_index = "dmarc_aggregate"
                 es_forensic_index = "dmarc_forensic"
                 es_smtp_tls_index = "smtp_tls"
@@ -1122,6 +1150,11 @@ def _init_output_clients(opts):
 
         try:
             if opts.opensearch_hosts:
+                logger.debug(
+                    "Initializing OpenSearch client: hosts=%s, ssl=%s",
+                    opts.opensearch_hosts,
+                    opts.opensearch_ssl,
+                )
                 os_aggregate_index = "dmarc_aggregate"
                 os_forensic_index = "dmarc_forensic"
                 os_smtp_tls_index = "smtp_tls"
@@ -1849,15 +1882,31 @@ def _main():
 
     logger.info("Starting parsedmarc")
 
-    # Initialize output clients
-    try:
-        clients = _init_output_clients(opts)
-    except ConfigurationError as e:
-        logger.critical(str(e))
-        exit(1)
-    except Exception as error_:
-        logger.error("Output client error: {0}".format(error_))
-        exit(1)
+    # Initialize output clients (with retry for transient connection errors)
+    clients = {}
+    max_retries = 4
+    retry_delay = 5
+    for attempt in range(max_retries + 1):
+        try:
+            clients = _init_output_clients(opts)
+            break
+        except ConfigurationError as e:
+            logger.critical(str(e))
+            exit(1)
+        except Exception as error_:
+            if attempt < max_retries:
+                logger.warning(
+                    "Output client error (attempt %d/%d, retrying in %ds): %s",
+                    attempt + 1,
+                    max_retries + 1,
+                    retry_delay,
+                    error_,
+                )
+                time.sleep(retry_delay)
+                retry_delay *= 2
+            else:
+                logger.error("Output client error: {0}".format(error_))
+                exit(1)
 
     file_paths = []
     mbox_paths = []

parsedmarc/constants.py

@@ -1,3 +1,3 @@
-__version__ = "9.5.4"
+__version__ = "9.5.5"
 
 USER_AGENT = f"parsedmarc/{__version__}"

parsedmarc/elastic.py

@@ -299,7 +299,7 @@ def set_hosts(
         else:
             conn_params["verify_certs"] = True
     if username and password:
-        conn_params["http_auth"] = username + ":" + password
+        conn_params["http_auth"] = (username, password)
     if api_key:
         conn_params["api_key"] = api_key
     connections.create_connection(**conn_params)

parsedmarc/opensearch.py

@@ -298,6 +298,7 @@ def set_hosts(
     """
     if not isinstance(hosts, list):
         hosts = [hosts]
+    logger.debug("Connecting to OpenSearch: hosts=%s, use_ssl=%s", hosts, use_ssl)
     conn_params = {"hosts": hosts, "timeout": timeout}
     if use_ssl:
         conn_params["use_ssl"] = True
@@ -323,7 +324,7 @@ def set_hosts(
         conn_params["connection_class"] = RequestsHttpConnection
     elif normalized_auth_type == "basic":
         if username and password:
-            conn_params["http_auth"] = username + ":" + password
+            conn_params["http_auth"] = (username, password)
         if api_key:
             conn_params["api_key"] = api_key
     else: