Add logging feature to the distro

noxfile.py

@@ -1007,9 +1007,35 @@ def integration_tests_redis(
                 kill_process_and_clean_outputs(temp_file, "test_redis", session)
 
 
+@nox.session()
+@nox.parametrize(
+    "python",
+    [(python) for python in python_versions()],
+)
+def integration_tests_logging(session):
+    session.install(".")
+    temp_file = create_it_tempfile("logging")
+    with session.chdir("src/test/integration/logging"):
+        try:
+            session.run(
+                "pytest",
+                "--tb=native",
+                "--log-cli-level=INFO",
+                "--color=yes",
+                "-v",
+                "./tests/test_logging.py",
+                env={
+                    "LUMIGO_DEBUG_LOGDUMP": temp_file,
+                    "LUMIGO_DEBUG": "true",
+                },
+            )
+        finally:
+            kill_process_and_clean_outputs(temp_file, "test_logging", session)
+
+
 def kill_process_and_clean_outputs(full_path: str, process_name: str, session) -> None:
     kill_process(process_name)
-    clean_outputs(full_path, session)
+    # clean_outputs(full_path, session)
 
 
 def clean_outputs(full_path: str, session) -> None:

src/lumigo_opentelemetry/__init__.py

@@ -94,20 +94,34 @@ def init() -> Dict[str, Any]:
     )
 
     from opentelemetry import trace
+    from opentelemetry import _logs
     from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
     from opentelemetry.sdk.trace import SpanLimits, TracerProvider
-
+    from opentelemetry.sdk._logs import LoggerProvider, LoggingHandler
+    from opentelemetry.sdk._logs.export import BatchLogRecordProcessor
+    from opentelemetry.exporter.otlp.proto.http._log_exporter import OTLPLogExporter
     from lumigo_opentelemetry.resources.span_processor import LumigoSpanProcessor
 
     DEFAULT_LUMIGO_ENDPOINT = (
         "https://ga-otlp.lumigo-tracer-edge.golumigo.com/v1/traces"
     )
+    DEFAULT_LUMIGO_LOGS_ENDPOINT = (
+        "https://ga-otlp.lumigo-tracer-edge.golumigo.com/v1/logs"
+    )
     DEFAULT_DEPENDENCIES_ENDPOINT = (
         "https://ga-otlp.lumigo-tracer-edge.golumigo.com/v1/dependencies"
     )
-    lumigo_endpoint = os.getenv("LUMIGO_ENDPOINT", DEFAULT_LUMIGO_ENDPOINT)
+    lumigo_traces_endpoint = os.getenv("LUMIGO_ENDPOINT", DEFAULT_LUMIGO_ENDPOINT)
+    lumigo_logs_endpoint = os.getenv(
+        "LUMIGO_LOGS_ENDPOINT", DEFAULT_LUMIGO_LOGS_ENDPOINT
+    )
     lumigo_token = os.getenv("LUMIGO_TRACER_TOKEN")
-    lumigo_report_dependencies = os.getenv("LUMIGO_REPORT_DEPENDENCIES", "true").lower()
+    lumigo_report_dependencies = (
+        os.getenv("LUMIGO_REPORT_DEPENDENCIES", "true").lower() == "true"
+    )
+    logging_enabled = os.getenv("LUMIGO_LOGS_ENABLED", "").lower() == "true"
+    spandump_file = os.getenv("LUMIGO_DEBUG_SPANDUMP")
+    logdump_file = os.getenv("LUMIGO_DEBUG_LOGDUMP")
 
     # Activate instrumentations
     from lumigo_opentelemetry.instrumentations import instrumentations  # noqa
@@ -122,27 +136,39 @@ def init() -> Dict[str, Any]:
     infrastructure_resource = get_infrastructure_resource()
     process_resource = get_process_resource()
 
-    tracer_resource = get_resource(
+    resource = get_resource(
         infrastructure_resource, process_resource, {"framework": framework}
     )
 
     tracer_provider = TracerProvider(
-        resource=tracer_resource,
+        resource=resource,
         sampler=_get_lumigo_sampler(),
         span_limits=(SpanLimits(max_span_attribute_length=(get_max_size()))),
     )
 
+    logger_provider = LoggerProvider(resource=resource)
+
     if lumigo_token:
         tracer_provider.add_span_processor(
             LumigoSpanProcessor(
                 OTLPSpanExporter(
-                    endpoint=lumigo_endpoint,
+                    endpoint=lumigo_traces_endpoint,
                     headers={"Authorization": f"LumigoToken {lumigo_token}"},
                 ),
             )
         )
 
-        if lumigo_report_dependencies == "true":
+        if logging_enabled:
+            logger_provider.add_log_record_processor(
+                BatchLogRecordProcessor(
+                    OTLPLogExporter(
+                        endpoint=lumigo_logs_endpoint,
+                        headers={"Authorization": f"LumigoToken {lumigo_token}"},
+                    )
+                )
+            )
+
+        if lumigo_report_dependencies:
             from lumigo_opentelemetry.dependencies import report
 
             try:
@@ -161,7 +187,6 @@ def init() -> Dict[str, Any]:
             "no data will be sent to Lumigo"
         )
 
-    spandump_file = os.getenv("LUMIGO_DEBUG_SPANDUMP")
     if spandump_file:
         from opentelemetry.sdk.trace.export import (
             ConsoleSpanExporter,
@@ -184,7 +209,32 @@ def init() -> Dict[str, Any]:
 
     trace.set_tracer_provider(tracer_provider)
 
-    return {"tracer_provider": tracer_provider}
+    if logging_enabled:
+        if logdump_file:
+            from opentelemetry.sdk._logs.export import (
+                ConsoleLogExporter,
+                SimpleLogRecordProcessor,
+            )
+
+            logger_provider.add_log_record_processor(
+                SimpleLogRecordProcessor(
+                    ConsoleLogExporter(
+                        out=open(logdump_file, "w"),
+                        # Print one log per line for ease of parsing, as the file itself
+                        # will not be valid JSON but a sequence of JSON objects (not a valid JSON array)
+                        formatter=lambda log_record: log_record.to_json() + "\n",
+                    )
+                )
+            )
+
+            logger.debug("Storing a copy of the log data under: %s", logdump_file)
+
+        _logs.set_logger_provider(logger_provider)
+
+        # Add the handler to the root logger, hence affecting all loggers created by the app from now on
+        logging.getLogger().addHandler(LoggingHandler(logger_provider=logger_provider))
+
+    return {"tracer_provider": tracer_provider, "logger_provider": logger_provider}
 
 
 def lumigo_wrapped(func: Callable[..., T]) -> Callable[..., T]:
@@ -211,5 +261,13 @@ def wrapper(*args: List[Any], **kwargs: Dict[Any, Any]) -> T:
 init_data = init()
 
 tracer_provider = init_data.get("tracer_provider")
-
-__all__ = ["auto_load", "init", "lumigo_wrapped", "logger", "tracer_provider"]
+logger_provider = init_data.get("logger_provider")
+
+__all__ = [
+    "auto_load",
+    "init",
+    "lumigo_wrapped",
+    "logger",
+    "tracer_provider",
+    "logger_provider",
+]

src/test/integration/conftest.py

@@ -3,6 +3,6 @@
 import pytest
 
 
-@pytest.fixture(autouse=True)
+@pytest.fixture()
 def increment_spans_counter():
     SpansContainer.update_span_offset()

src/test/integration/logging/app/logging_app.py

@@ -0,0 +1,11 @@
+from lumigo_opentelemetry import logger_provider
+import logging
+
+logger = logging.getLogger(__name__)
+
+logger.setLevel(logging.DEBUG)
+
+logger.debug("Hello OTEL!")
+
+logger_provider.force_flush()
+logger_provider.shutdown()

src/test/integration/logging/tests/test_logging.py

@@ -0,0 +1,37 @@
+import os
+import subprocess
+import sys
+import unittest
+from os import path
+from test.test_utils.logs_parser import LogsContainer
+
+
+def run_logging_app():
+    sample_path = path.join(
+        path.dirname(path.abspath(__file__)),
+        "../app/logging_app.py",
+    )
+    subprocess.check_output(
+        [sys.executable, sample_path],
+        env={
+            **os.environ,
+            "AUTOWRAPT_BOOTSTRAP": "lumigo_opentelemetry",
+            "OTEL_SERVICE_NAME": "logging-app",
+            "LUMIGO_LOGS_ENABLED": "true",
+        },
+    )
+
+
+class TestLogging(unittest.TestCase):
+    def test_logging(self):
+        run_logging_app()
+
+        logs_container = LogsContainer.get_logs_from_file()
+
+        self.assertEqual(len(logs_container), 1)
+
+        logs_container[0]["body"] == "Hello OTEL!"
+
+        # "resource" is currently not a proper dictionary, but a string from a repr(...) call over the resource attributes.
+        # Pending a fix in https://github.com/open-telemetry/opentelemetry-python/pull/3346
+        logs_container[0]["resource"]["service.name"] == "logging-app"

src/test/test_utils/logs_parser.py

@@ -0,0 +1,38 @@
+from __future__ import annotations
+
+import json
+import os
+import time
+from dataclasses import dataclass
+from typing import Any, Dict, List, Optional
+
+LOGS_FILE_FULL_PATH = os.environ.get("LUMIGO_DEBUG_LOGDUMP")
+
+
+@dataclass(frozen=True)
+class LogsContainer:
+    logs: List[Dict[str, Any]]
+
+    @staticmethod
+    def get_logs_from_file(
+        path: Optional[str] = LOGS_FILE_FULL_PATH,
+        wait_time_sec: int = 3,
+    ) -> LogsContainer:
+        time.sleep(wait_time_sec)
+
+        try:
+            with open(path) as file:
+                logs = [json.loads(line) for line in file.readlines()]
+
+            return LogsContainer(logs=logs)
+
+        except Exception as err:
+            print(f"Failed to parse logs from file after {wait_time_sec}s: {err}")
+
+        return LogsContainer(logs=[])
+
+    def __len__(self) -> int:
+        return len(self.logs)
+
+    def __getitem__(self, item) -> Dict[str, Any]:
+        return self.logs[item]

src/test/test_utils/spans_parser.py

@@ -6,7 +6,7 @@
 from dataclasses import dataclass
 from typing import Any, Dict, List, Optional
 
-SPANS_FILE_FULL_PATH = os.environ["LUMIGO_DEBUG_SPANDUMP"]
+SPANS_FILE_FULL_PATH = os.environ.get("LUMIGO_DEBUG_SPANDUMP")
 
 
 class SpansCounter:

src/test/unit/test_tracer.py

@@ -12,11 +12,14 @@
 
 
 class TestDistroInit(unittest.TestCase):
-    def test_access_trace_provider(self):
-        from lumigo_opentelemetry import tracer_provider
+    def test_access_trace_and_logging_providers(self):
+        from lumigo_opentelemetry import tracer_provider, logger_provider
 
         self.assertIsNotNone(tracer_provider)
+        self.assertTrue(hasattr(tracer_provider, "force_flush"))
+        self.assertTrue(hasattr(tracer_provider, "shutdown"))
 
+        self.assertIsNotNone(logger_provider)
         self.assertTrue(hasattr(tracer_provider, "force_flush"))
         self.assertTrue(hasattr(tracer_provider, "shutdown"))