Sarc 368 loki connect

config/sarc-dev.json

@@ -1,4 +1,9 @@
 {
+    "logging": {
+        "log_level": "WARNING",
+        "OTLP_endpoint": "http://loki01.server.mila.quebec:3100/otlp/v1/logs",
+        "service_name": "sarc-dev"
+    },
     "mongo": {
         "connection_string": "mongodb://localhost:27017/sarc-dev",
         "database_name": "sarc-dev"

config/sarc-prod.json

@@ -1,4 +1,9 @@
 {
+    "logging": {
+        "log_level": "WARNING",
+        "OTLP_endpoint": "http://loki01.server.mila.quebec:3100/otlp/v1/logs",
+        "service_name": "sarc-dev"
+    },
     "mongo": {
         "connection_string": "mongodb://writeuser:REPLACEME@localhost:27017/sarc",
         "database_name": "sarc"

pyproject.toml

@@ -25,6 +25,7 @@ sphinx-rtd-theme = "^1.2.2"
 nbsphinx = "^0.9.3"
 opentelemetry-api = "^1.23.0"
 opentelemetry-sdk = "^1.23.0"
+opentelemetry-exporter-otlp = "^1.23.0"
 requests-mock = "^1.11.0"
 pandas = "< 2.1.0"
 gifnoc = "^0.2.3"
@@ -39,6 +40,7 @@ mypy = ">= 0.991"
 pytest = ">= 7.2.1"
 pytest-cov = ">= 4.0.0"
 pytest-regressions = ">= 2.4.2"
+pytest-httpserver = ">= 1.1.0"
 tox = ">= 4.4.5"
 pytest-freezegun = ">= 0.4.2"
 fabric = {extras = ["testing"], version = "^3.0.0"}

sarc/cli/__init__.py

@@ -8,13 +8,12 @@
 
 from simple_parsing import ArgumentParser, field, subparsers
 
+from sarc.logging import setupLogging
+
 from .acquire import Acquire
 from .db import Db
 from .health import Health
 
-logger = logging.getLogger(__name__)
-
-
 colors = SimpleNamespace(
     grey="\033[38;21m",
     blue="\033[38;5;39m",
@@ -64,27 +63,35 @@ class CLI:
     )
 
     def execute(self) -> int:
-        levels = {0: logging.WARNING, 1: logging.INFO, 2: logging.DEBUG}
+        # levels = {0: logging.WARNING, 1: logging.INFO, 2: logging.DEBUG}
 
-        if self.color:
-            logging.basicConfig(
-                handlers=[NiceHandler()],
-                level=levels.get(self.verbose, logging.DEBUG),
-            )
+        # if self.color:
+        #     logging.basicConfig(
+        #         handlers=[NiceHandler(), getHandler()],
+        #         level=levels.get(self.verbose, logging.DEBUG),
+        #     )
 
-        else:
-            logging.basicConfig(
-                format="%(asctime)-15s::%(levelname)s::%(name)s::%(message)s",
-                level=levels.get(self.verbose, logging.DEBUG),
-            )
+        # else:
+        #     logging.basicConfig(
+        #         handlers=[getHandler()],
+        #         format="%(asctime)-15s::%(levelname)s::%(name)s::%(message)s",
+        #         level=levels.get(self.verbose, logging.DEBUG),
+        #     )
 
-        # logger.debug("SARC version : %s", sarc.__version__)
+        # logger = logging.getLogger(__name__)
+
+        # # logger.debug("SARC version : %s", sarc.__version__)
+        # logger.debug(f"Running command: {self.command}")
+        # logger.warning(f"Test warning log")
 
         return self.command.execute()
 
 
 def main(argv: list[Any] | None = None) -> int:
     """Main commandline for SARC"""
+
+    setupLogging()
+
     parser = ArgumentParser()
     parser.add_arguments(CLI, dest="command")
     args = parser.parse_args(argv)

sarc/config.py

@@ -178,6 +178,12 @@ class AccountMatchingConfig(BaseModel):
     make_matches_config: Path
 
 
+class LoggingConfig(BaseModel):
+    log_level: str
+    OTLP_endpoint: str
+    service_name: str
+
+
 # pylint: disable=unused-argument,redefined-outer-name
 def _absolute_path(value, values, config, field):
     return value and value.expanduser().absolute()
@@ -201,6 +207,9 @@ class ScraperConfig(BaseModel):
     account_matching: AccountMatchingConfig = None
     sshconfig: Path = None
     clusters: dict[str, ClusterConfig] = None
+    logging: LoggingConfig = None
+    loki: LokiConfig = None
+    tempo: TempoConfig = None
 
     _abs_path = validator("cache", "sshconfig", allow_reuse=True)(_absolute_path)
 

sarc/jobs/sacct.py

@@ -263,7 +263,7 @@ def sacct_mongodb_import(
     """
     collection = _jobs_collection()
     scraper = SAcctScraper(cluster, day)
-    logger.info("Getting the sacct data...")
+    logger.info(f"Getting the sacct data for cluster {cluster.name}, date {day}...")
     scraper.get_raw()
     logger.info(
         f"Saving into mongodb collection '{collection.Meta.collection_name}'..."

sarc/logging.py

@@ -0,0 +1,48 @@
+import logging
+import os
+
+from sarc.config import config
+
+
+def setupLogging():
+    from opentelemetry._logs import set_logger_provider
+    from opentelemetry.exporter.otlp.proto.http._log_exporter import OTLPLogExporter
+    from opentelemetry.sdk._logs import LoggerProvider, LoggingHandler
+    from opentelemetry.sdk._logs.export import BatchLogRecordProcessor
+    from opentelemetry.sdk.resources import Resource
+
+    logging_levels = {
+        "DEBUG": logging.DEBUG,
+        "INFO": logging.INFO,
+        "WARNING": logging.WARNING,
+        "ERROR": logging.ERROR,
+        "CRITICAL": logging.CRITICAL,
+    }
+
+    if config().logging:
+        log_level = logging_levels.get(config().logging.log_level, logging.WARNING)
+        logger_provider = LoggerProvider(
+            resource=Resource.create(
+                {
+                    "service.name": config().logging.service_name,
+                    "service.instance.id": os.uname().nodename,
+                }
+            ),
+        )
+        set_logger_provider(logger_provider)
+
+        otlp_exporter = OTLPLogExporter(config().logging.OTLP_endpoint)
+        logger_provider.add_log_record_processor(BatchLogRecordProcessor(otlp_exporter))
+        handler = LoggingHandler(level=log_level, logger_provider=logger_provider)
+
+        logging.basicConfig(
+            handlers=[handler, logging.StreamHandler()],
+            format="%(asctime)-15s::%(levelname)s::%(name)s::%(message)s",
+            level=log_level,
+        )
+    else:
+        logging.basicConfig(
+            handlers=[logging.StreamHandler()],
+            format="%(asctime)-15s::%(levelname)s::%(name)s::%(message)s",
+            level=logging.WARNING,  # Default log level
+        )

tests/functional/jobs/test_func_sacct.py

@@ -938,7 +938,10 @@ def _create_session(cluster_name, cmd_template, datetimes):
             caplog.text,
         )
     )
-    assert "Getting the sacct data..." in caplog.text
+    assert (
+        "Getting the sacct data for cluster raisin, date 2023-02-15 00:00:00..."
+        in caplog.text
+    )
     assert "Saving into mongodb collection '" in caplog.text
     assert bool(
         re.search(

tests/functional/test_loki_logging.py

@@ -0,0 +1,44 @@
+# test logging to loki service
+
+import logging
+
+import pytest
+import requests_mock
+from pytest_httpserver import HTTPServer
+
+from sarc.config import config
+from sarc.logging import setupLogging
+
+
+@pytest.fixture
+def httpserver(httpserver: HTTPServer):
+
+    # Configurer le mock pour l'endpoint Loki
+    httpserver.expect_request("/otlp/v1/logs").respond_with_json(
+        {"status": "success", "data": {"result": []}}
+    )
+    return httpserver
+
+
+from sarc.logging import setupLogging
+
+
+# @pytest.mark.no_capture
+# @pytest.mark.skip(reason="cannot make HTTPServer running without error in this situation")
+def test_loki_logging(
+    standard_config,
+):
+    # Configurer l'URL de l'endpoint Loki
+    # print(f"http://{httpserver.host}:{httpserver.port}/otlp/v1/logs")
+    # print(httpserver.url_for("/otlp/v1/logs"))
+    loki_url = httpserver.url_for("/otlp/v1/logs")
+    config().logging.OTLP_endpoint = loki_url
+
+    setupLogging()
+
+    logging.warning("test Warning message")
+
+    # assert loki_mock.called
+    # assert loki_mock.call_count == 1
+    # assert loki_mock.request_history[0].url == 'http://loki01.server.raisin.quebec:3100/otlp/v1/logs'
+    assert True

tests/sarc-test.json

@@ -1,4 +1,9 @@
 {
+    "logging": {
+        "log_level": "WARNING",
+        "OTLP_endpoint": "http://127.0.0.1/otlp/v1/logs",
+        "service_name": "sarc-test"
+    },
     "mongo": {
         "connection_string": "localhost:27017",
         "database_name": "sarc-test"