Add configurable log level (#68)

config.sample.toml

@@ -3,6 +3,7 @@ source_collector_dir = "path/to/source_collector_dir/"
 host_modifier_dir = "path/to/host_modifier_dir/"
 db_uri = "dbname='zac' user='zabbix' host='localhost' password='secret' port=5432 connect_timeout=2"
 health_file = "/tmp/zac_health.json"
+log_level = "DEBUG"
 
 [zabbix]
 map_dir = "path/to/map_dir/"

tests/test_models.py

@@ -1,3 +1,4 @@
+import logging
 import pytest
 from pydantic import ValidationError
 from zabbix_auto_config import models
@@ -114,3 +115,65 @@ def test_host_merge_invalid(full_hosts):
     h1 = models.Host(**host)
     with pytest.raises(TypeError):
         h1.merge(object())
+
+
+@pytest.mark.parametrize(
+    "level,expect",
+    [
+        ["notset", logging.NOTSET],
+        ["debug", logging.DEBUG],
+        ["info", logging.INFO],
+        ["warn", logging.WARN],
+        ["warning", logging.WARNING],
+        ["error", logging.ERROR],
+        ["fatal", logging.FATAL],
+        ["critical", logging.CRITICAL],
+    ],
+)
+@pytest.mark.parametrize("upper", [True, False])
+def test_zacsettings_log_level_str(level: str, expect: int, upper: bool) -> None:
+    settings = models.ZacSettings(
+        db_uri="",
+        source_collector_dir="",
+        host_modifier_dir="",
+        log_level=level.upper() if upper else level.lower(),
+    )
+    assert settings.log_level == expect
+
+
+@pytest.mark.parametrize(
+    "level,expect",
+    [
+        [0, logging.NOTSET],
+        [10, logging.DEBUG],
+        [20, logging.INFO],
+        [30, logging.WARN],
+        [30, logging.WARNING],
+        [40, logging.ERROR],
+        [50, logging.FATAL],
+        [50, logging.CRITICAL],
+    ],
+)
+def test_zacsettings_log_level_int(level: str, expect: int) -> None:
+    settings = models.ZacSettings(
+        db_uri="",
+        source_collector_dir="",
+        host_modifier_dir="",
+        log_level=level,
+    )
+    assert settings.log_level == expect
+
+
+def test_zacsettings_log_level_serialize() -> None:
+    settings = models.ZacSettings(
+        db_uri="", source_collector_dir="", host_modifier_dir="", log_level=logging.INFO
+    )
+    assert settings.log_level == logging.INFO == 20  # sanity check
+
+    # Serialize to dict:
+    settings_dict = settings.model_dump()
+    assert settings_dict["log_level"] == "INFO"
+
+    # Serialize to JSON:
+    settings_json = settings.model_dump_json()
+    assert '"log_level":"INFO"' in settings_json

zabbix_auto_config/__init__.py

@@ -114,14 +114,13 @@ def log_process_status(processes):
 
 
 def main():
+    multiprocessing_logging.install_mp_handler()
     logging.basicConfig(format='%(asctime)s %(levelname)s [%(processName)s %(process)d] [%(name)s] %(message)s', datefmt="%Y-%m-%dT%H:%M:%S%z", level=logging.DEBUG)
     config = get_config()
-
-    multiprocessing_logging.install_mp_handler()
+    logging.getLogger().setLevel(config.zac.log_level)
     logging.getLogger("urllib3.connectionpool").setLevel(logging.ERROR)
 
     logging.info("Main start (%d) version %s", os.getpid(), __version__)
-
     stop_event = multiprocessing.Event()
     state_manager = multiprocessing.Manager()
     processes = []
@@ -198,3 +197,7 @@ def main():
             alive_processes = [process for process in processes if process.is_alive()]
 
     logging.info("Main exit")
+
+
+if __name__ == "__main__":
+    main()

zabbix_auto_config/models.py

@@ -4,7 +4,13 @@
 
 from pydantic import BaseModel
 from pydantic import BaseModel as PydanticBaseModel
-from pydantic import ConfigDict, Field, field_validator, model_validator
+from pydantic import (
+    ConfigDict,
+    Field,
+    field_validator,
+    field_serializer,
+    model_validator,
+)
 from typing_extensions import Annotated
 
 from . import utils
@@ -60,11 +66,43 @@ class ZabbixSettings(ConfigBaseModel):
     # These groups are not managed by ZAC beyond creating them.
     extra_siteadmin_hostgroup_prefixes: Set[str] = set()
 
+
 class ZacSettings(ConfigBaseModel):
     source_collector_dir: str
     host_modifier_dir: str
     db_uri: str
     health_file: Optional[Path] = None
+    log_level: int = Field(logging.DEBUG, description="The log level to use.")
+
+    @field_serializer("log_level")
+    def _serialize_log_level(self, v: str) -> str:
+        """Serializes the log level as a string.
+        Ensures consistent semantics between loading/storing log level in config.
+        E.g. we dump `"INFO"` instead of `20`.
+        """
+        return logging.getLevelName(v)
+
+    @field_validator("log_level", mode="before")
+    @classmethod
+    def _validate_log_level(cls, v: Any) -> int:
+        """Validates the log level and converts it to an integer.
+        The log level can be specified as an integer or a string."""
+        if isinstance(v, int):
+            if v not in logging._levelToName:
+                raise ValueError(
+                    f"Invalid log level: {v} is not a valid log level integer."
+                )
+            return v
+        elif isinstance(v, str):
+            v = v.upper()
+            level_int = logging._nameToLevel.get(v, None)
+            if level_int is None:
+                raise ValueError(
+                    f"Invalid log level: {v} is not a valid log level name."
+                )
+            return level_int
+        else:
+            raise TypeError("Log level must be an integer or string.")
 
 
 class SourceCollectorSettings(ConfigBaseModel, extra="allow"):