[Enhancement]: Add Structured Logging with Log Levels Across All Microservices Using Python's logging Module

[divyanshim27]

Summary

Arachnode is a microservice architecture with 7 distinct services (crawler, scraper, aggregator, contact-discovery, email-generator, gateway, scheduler). Currently, logging across these services is inconsistent — some services use print() statements, some use basic logging.info() calls, and the scheduler service has its own logger.py that is not shared. For a distributed system, this makes debugging across services extremely difficult.

Problem

• print() statements are used for debug output in multiple services instead of proper structured logging.
• The scheduler has a dedicated logger.py but this is not shared or replicated across other services — each service invents its own logging approach.
• Log output mixes levels (info, warning, error) without consistent formatting, making it hard to filter logs in production (e.g., in Docker or Cloud logging tools).
• There is no correlation ID or request ID propagated across service boundaries, so tracing a single job from crawler → aggregator → contact-discovery → email-generator requires manually correlating timestamps.
• When running docker compose up, log output from all 7 services is interleaved with no consistent format to distinguish them.

Impact

• Debugging a multi-step workflow failure (e.g., "why didn't an email get generated for this job?") requires manually reading logs from multiple containers.
• Production logging on any cloud platform (GCP Cloud Logging, Datadog, etc.) works best with structured JSON logs — plain text logs are hard to query.
• Inconsistent log levels mean error conditions may be logged at INFO level and go unnoticed.

Proposed Solution

I will create a shared logging utility and integrate it across all services:

shared/logger.py (to be placed in each service or as a shared volume):

import logging
import json
import sys
from datetime import datetime, timezone

class JSONFormatter(logging.Formatter):
    def format(self, record):
        log_entry = {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "level": record.levelname,
            "service": record.name,
            "message": record.getMessage(),
            "module": record.module,
            "line": record.lineno,
        }
        if record.exc_info:
            log_entry["exception"] = self.formatException(record.exc_info)
        return json.dumps(log_entry)

def get_logger(service_name: str, level: str = "INFO") -> logging.Logger:
    logger = logging.getLogger(service_name)
    logger.setLevel(getattr(logging, level.upper(), logging.INFO))
    if not logger.handlers:
        handler = logging.StreamHandler(sys.stdout)
        handler.setFormatter(JSONFormatter())
        logger.addHandler(handler)
    return logger

Usage in each service (replacing print()):

# Before:
print(f"Job inserted: {job_id}")

# After:
from shared.logger import get_logger
logger = get_logger("aggregator-service")
logger.info("Job inserted", extra={"job_id": job_id})

Additional Notes

• I will replace all print() statements across all 7 services with appropriately leveled logger.info(), logger.warning(), or logger.error() calls.
• I will add a LOG_LEVEL environment variable to .env.example so log verbosity can be tuned per environment.
• The JSON format will work natively with GCP Cloud Logging's structured log parsing.

Please assign this issue to me.

Labels: enhancement, devops, backend, observability, GSSoC 2026

[divyanshim27]

Hey @sharmavaibhav31 ,
Please assign it to me under GSSoC 2026.
/assign /claim /claim-assignment