chore: sync shared modules and metrics across all repositories

- Updated all repos to support prometheus metrics

Author: mobious999

src/app/config_shared.py

@@ -8,47 +8,9 @@
 
 from app.utils.config_utils import get_config_bool, get_config_value
 from app.utils.types import OutputMode
-from app.utils.vault_client import VaultClient, get_secret_or_env
+from app.utils.vault_client import get_secret_or_env
 
-_vault = VaultClient()
-
-# @lru_cache
-# def get_config_value(key: str, default: Optional[str] = None) -> str:
-#     """
-#     Retrieve a configuration value from Vault, environment variable, or fallback.
-
-#     Args:
-#         key (str): The configuration key to retrieve.
-#         default (Optional[str]): Fallback value if the key is not found.
-
-#     Returns:
-#         str: The resolved configuration value.
-
-#     Raises:
-#         ValueError: If no value is found and no default is provided.
-#     """
-#     val = _vault.get(key, os.getenv(key))
-#     if val is None:
-#         if default is not None:
-#             return str(default)
-#         raise ValueError(f"❌ Missing required config for key: {key}")
-#     return str(val)
-
-
-# @lru_cache
-# def get_config_bool(key: str, default: bool = False) -> bool:
-#     """
-#     Retrieve a boolean configuration value with support for multiple true-like values.
-
-#     Args:
-#         key (str): The configuration key to retrieve.
-#         default (bool): Default value if the key is not found.
-
-#     Returns:
-#         bool: The resolved boolean configuration value.
-#     """
-#     val = get_config_value(key, str(default)).strip().lower()
-#     return val in {"1", "true", "yes", "on"}
+# _vault = VaultClient()
 
 
 @lru_cache
@@ -1371,3 +1333,136 @@ def get_healthcheck_port() -> int:
 
     """
     return int(get_config_value("HEALTHCHECK_PORT", "8081"))
+
+
+@lru_cache
+def get_metrics_enabled() -> bool:
+    """Determine whether the Prometheus metrics server should be started.
+
+    This checks the value of METRICS_ENABLED from Vault or the environment.
+    Accepts truthy values: "1", "true", "yes" (case-insensitive).
+
+    Returns:
+        bool: True if metrics are enabled, otherwise False.
+
+    """
+    val = get_config_value("METRICS_ENABLED", "true").lower()
+    return val in ("1", "true", "yes")
+
+
+@lru_cache
+def get_metrics_port() -> int:
+    """Retrieve the TCP port number on which the Prometheus metrics server should listen.
+
+    Returns:
+        int: Port number (default: 8000).
+
+    Raises:
+        ValueError: If METRICS_PORT is not a valid integer.
+
+    """
+    port_str = get_config_value("METRICS_PORT", "8000")
+    try:
+        return int(port_str)
+    except ValueError:
+        raise ValueError(f"❌ Invalid METRICS_PORT value: '{port_str}' must be an integer.")
+
+
+@lru_cache
+def get_metrics_bind_address() -> str:
+    """Get the network address to bind the metrics server.
+
+    Returns:
+        str: Bind address (default: "0.0.0.0").
+
+    """
+    return get_config_value("METRICS_BIND_ADDRESS", "0.0.0.0")
+
+
+@lru_cache
+def get_config_bool(key: str, default: bool = False) -> bool:
+    """Retrieve a boolean configuration value from Vault or environment.
+
+    Interprets common truthy values ("1", "true", "yes", "on") as True.
+    Uses a string fallback if the key is not found.
+
+    Args:
+        key (str): The configuration key to retrieve.
+        default (bool): The fallback value if key is missing or invalid.
+
+    Returns:
+        bool: Boolean value for the requested configuration key.
+
+    """
+    val: str = get_config_value(key, str(default)).lower()
+    return val in ("1", "true", "yes", "on")
+
+
+@lru_cache
+def get_output_modes() -> List[str]:
+    """Retrieve a list of enabled output modes.
+
+    This reads the OUTPUT_MODES config key, splits it by commas, and returns
+    a lowercase list of modes like "s3", "rest", or "database".
+
+    Returns:
+        List[str]: Enabled output modes.
+
+    """
+    modes: str = get_config_value("OUTPUT_MODES", "")
+    return [m.strip().lower() for m in modes.split(",") if m.strip()]
+
+
+@lru_cache
+def get_rest_output_url() -> str:
+    """Get the REST endpoint URL for output dispatch.
+
+    Returns:
+        str: Fully qualified REST API URL.
+
+    """
+    return get_config_value("REST_OUTPUT_URL")
+
+
+@lru_cache
+def get_s3_output_bucket() -> str:
+    """Get the name of the S3 bucket used for output dispatch.
+
+    Returns:
+        str: S3 bucket name.
+
+    """
+    return get_config_value("S3_OUTPUT_BUCKET")
+
+
+@lru_cache
+def get_s3_output_prefix() -> str:
+    """Get the object key prefix used for S3 output files.
+
+    Returns:
+        str: S3 key prefix (can be empty).
+
+    """
+    return get_config_value("S3_OUTPUT_PREFIX", "")
+
+
+@lru_cache
+def get_database_output_url() -> str:
+    """Get the SQLAlchemy-compatible connection string for output database.
+
+    Returns:
+        str: Database URL (e.g., postgresql://user:pass@host/db).
+
+    """
+    return get_config_value("DATABASE_OUTPUT_URL")
+
+
+@lru_cache
+def get_database_insert_sql() -> str:
+    """Get the raw SQL INSERT statement template for database output.
+
+    Returns:
+        str: SQL insert statement.
+
+    """
+    return get_config_value("DATABASE_INSERT_SQL")

src/app/main.py

@@ -1,41 +1,89 @@
 """Main entry point for the service.
 
-This script initializes logging, loads the queue consumer, and begins
-consuming data using the configured processing callback.
+Initializes logging, sets up metrics, validates configuration, and
+starts consuming messages using the configured output handler.
 """
 
 import os
 import sys
-
-# Add 'src/' to Python's module search path
-sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
+import traceback
 
 from app import config_shared
-from app.output_handler import send_to_output
+from app.output_handler import output_handler
 from app.queue_handler import consume_messages
+from app.utils.metrics_server import start_metrics_server
 from app.utils.setup_logger import setup_logger
 
-# Initialize the module-level logger with optional structured logging
+# Add 'src/' to Python's module search path
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
+
+# Initialize structured or plain logger
 logger = setup_logger(
     __name__,
     structured=config_shared.get_config_bool("STRUCTURED_LOGGING", False),
 )
 
+REDACT_LOGS = config_shared.get_config_bool("REDACT_SENSITIVE_LOGS", True)
+
+
+def redact(value: str) -> str:
+    """Redact sensitive values from logs if redaction is enabled.
+
+    Args:
+        value (str): The value to redact.
+
+    Returns:
+        str: Redacted or original string.
+
+    """
+    return "[REDACTED]" if REDACT_LOGS else value
+
+
+def validate_output_config() -> None:
+    """Validate that required config variables are present for selected output modes.
+
+    Logs sanitized values for REST, S3, and database outputs to help with troubleshooting.
+    """
+    output_modes = config_shared.get_output_modes()
+    logger.info("📤 Output modes enabled: %s", output_modes)
+
+    if "rest" in output_modes:
+        url = config_shared.get_rest_output_url()
+        logger.info("🌐 REST output URL: %s", redact(url))
+
+    if "s3" in output_modes:
+        bucket = config_shared.get_s3_output_bucket()
+        prefix = config_shared.get_s3_output_prefix()
+        logger.info("🪣 S3 bucket: %s, prefix: %s", redact(bucket), redact(prefix))
+
+    if "database" in output_modes:
+        db_url = config_shared.get_database_output_url()
+        insert_sql = config_shared.get_database_insert_sql()
+        logger.info("🗄️ DB URL: %s", redact(db_url))
+        logger.debug("📝 Insert SQL: %s", redact(insert_sql))
+
 
 def main() -> None:
     """Start the data processing service.
 
-    This function initializes the service by calling the queue consumer,
-    which begins listening to RabbitMQ or SQS and processes data using
-    the `send_to_output` callback.
+    This function performs startup tasks and begins consuming messages
+    from the configured queue using the output handler.
     """
     logger.info("🚀 Starting processing service...")
-    consume_messages(send_to_output)
+
+    start_metrics_server()
+    validate_output_config()
+
+    logger.info(
+        "✅ Ready. Listening for messages on queue type: %s", config_shared.get_queue_type()
+    )
+    consume_messages(output_handler.send)
 
 
 if __name__ == "__main__":
     try:
         main()
     except Exception as e:
-        logger.exception("❌ Unhandled exception in main: %s", e)
+        logger.exception("❌ Unhandled exception: %s", redact(str(e)))
+        traceback.print_exc()
         sys.exit(1)

src/app/utils/metrics_server.py

@@ -10,12 +10,12 @@
 """
 
 import os
+
 from prometheus_client import start_http_server
 
 
 def start_metrics_server() -> None:
-    """
-    Conditionally start the Prometheus metrics HTTP server.
+    """Conditionally start the Prometheus metrics HTTP server.
 
     This starts a simple HTTP server to expose metrics from the global
     Prometheus client registry on the given port. If the environment
@@ -31,6 +31,7 @@ def start_metrics_server() -> None:
 
     Example:
         start_metrics_server()  # Will start if METRICS_ENABLED is true
+
     """
     enabled = os.getenv("METRICS_ENABLED", "true").lower()
     if enabled not in ("1", "true", "yes"):

src/app/utils/setup_logger.py

@@ -1,7 +1,6 @@
 import logging
 import sys
 from logging import Logger
-from typing import Optional
 
 
 def setup_logger(