vdk-core: Allow logs to be sent to an endpoint using SysLog (#807)

This change makes it so an cloud executions of jobs can have an endpoint configured, where logs will be sent automatically using SysLog. Testing done: ran job with a configured endpoint and observed logs in the endpoint Signed-off-by: Gabriel Georgiev <[email protected]>
vmware · Apr 19, 2022 · c5be8a0 · c5be8a0
1 parent fa96ce9
commit c5be8a0
Show file tree

Hide file tree

Showing 2 changed files with 111 additions and 34 deletions.
diff --git a/projects/vdk-core/src/vdk/internal/builtin_plugins/config/log_config.py b/projects/vdk-core/src/vdk/internal/builtin_plugins/config/log_config.py
@@ -1,6 +1,7 @@
 # Copyright 2021 VMware, Inc.
 # SPDX-License-Identifier: Apache-2.0
 import logging
+import socket
 import types
 from sys import modules
 from typing import cast
@@ -9,14 +10,23 @@
 from vdk.internal.builtin_plugins.config import vdk_config
 from vdk.internal.builtin_plugins.run.job_context import JobContext
 from vdk.internal.core import errors
+from vdk.internal.core.config import ConfigurationBuilder
 from vdk.internal.core.statestore import CommonStoreKeys
 
+SYSLOG_URL = "SYSLOG_URL"
+SYSLOG_PORT = "SYSLOG_PORT"
+SYSLOG_SOCK_TYPE = "SYSLOG_SOCK_TYPE"
+SYSLOG_ENABLED = "SYSLOG_ENABLED"
+
+SYSLOG_SOCK_TYPE_VALUES_DICT = {"UDP": socket.SOCK_DGRAM, "TCP": socket.SOCK_STREAM}
+
 
 def configure_loggers(
     job_name: str = "",
     attempt_id: str = "no-id",
     log_config_type: str = None,
     vdk_logging_level: str = "DEBUG",
+    syslog_args: (str, int, str, bool) = ("localhost", 514, "UDP", False),
 ) -> None:
     """
     Configure default logging configuration
@@ -25,6 +35,7 @@ def configure_loggers(
     :param attempt_id: the id of the current job run attempt
     :param log_config_type: where the job is executed: CLOUD or LOCAL
     :param vdk_logging_level: The level for vdk specific logs.
+    :param syslog_args: Arguments necessary for SysLog logging.
     """
 
     import logging.config
@@ -57,12 +68,38 @@ def configure_loggers(
         "stream": "ext://sys.stderr",
     }
 
+    syslog_url, syslog_port, syslog_sock_type, syslog_enabled = syslog_args
+
+    if syslog_sock_type not in SYSLOG_SOCK_TYPE_VALUES_DICT:
+        errors.log_and_throw(
+            to_be_fixed_by=errors.ResolvableBy.CONFIG_ERROR,
+            log=log,
+            what_happened=f"Provided configuration variable for {SYSLOG_SOCK_TYPE} has invalid value.",
+            why_it_happened=f"VDK was run with {SYSLOG_SOCK_TYPE}={syslog_sock_type}, however {syslog_sock_type} is invalid value for this variable.",
+            consequences=errors.MSG_CONSEQUENCE_DELEGATING_TO_CALLER__LIKELY_EXECUTION_FAILURE,
+            countermeasures=f"Provide a valid value for {SYSLOG_SOCK_TYPE}."
+            f"Currently possible values are {list(SYSLOG_SOCK_TYPE_VALUES_DICT.keys())}",
+        )
+
+    _SYSLOG_HANDLER = {
+        "level": "DEBUG",
+        "class": "logging.handlers.SysLogHandler",
+        "formatter": "detailedFormatter",
+        "address": (syslog_url, syslog_port),
+        "socktype": SYSLOG_SOCK_TYPE_VALUES_DICT[syslog_sock_type],
+        "facility": "user",
+    }
+
+    handlers = ["consoleHandler"]
+    if syslog_enabled:
+        handlers.append("SysLog")
+
     if "CLOUD" == log_config_type:
         CLOUD = {  # @UnusedVariable
             "version": 1,
-            "handlers": {"consoleHandler": _CONSOLE_HANDLER},
+            "handlers": {"consoleHandler": _CONSOLE_HANDLER, "SysLog": _SYSLOG_HANDLER},
             "formatters": _FORMATTERS,
-            "root": {"handlers": ["consoleHandler"]},
+            "root": {"handlers": handlers},
             "loggers": _LOGGERS,
             "disable_existing_loggers": False,
         }
@@ -72,9 +109,9 @@ def configure_loggers(
     else:
         LOCAL = {  # @UnusedVariable
             "version": 1,
-            "handlers": {"consoleHandler": _CONSOLE_HANDLER},
+            "handlers": {"consoleHandler": _CONSOLE_HANDLER, "SysLog": _SYSLOG_HANDLER},
             "formatters": _FORMATTERS,
-            "root": {"handlers": ("consoleHandler",)},
+            "root": {"handlers": handlers},
             "loggers": _LOGGERS,
             "disable_existing_loggers": False,
         }
@@ -125,6 +162,30 @@ class LoggingPlugin:
     Define the logging plugin
     """
 
+    @staticmethod
+    @hookimpl
+    def vdk_configure(config_builder: ConfigurationBuilder):
+        config_builder.add(
+            key=SYSLOG_URL,
+            default_value="localhost",
+            description="The hostname of the endpoint to which VDK logs will be sent through SysLog.",
+        )
+        config_builder.add(
+            key=SYSLOG_PORT,
+            default_value=514,
+            description="The port of the endpoint to which VDK logs will be sent through SysLog.",
+        )
+        config_builder.add(
+            key=SYSLOG_ENABLED,
+            default_value=False,
+            description="If set to True, SysLog log forwarding is enabled.",
+        )
+        config_builder.add(
+            key=SYSLOG_SOCK_TYPE,
+            default_value="UDP",
+            description=f"Socket type for SysLog log forwarding connection. Currently possible values are {list(SYSLOG_SOCK_TYPE_VALUES_DICT.keys())}",
+        )
+
     @hookimpl
     def initialize_job(self, context: JobContext) -> None:
         """
@@ -141,12 +202,19 @@ def initialize_job(self, context: JobContext) -> None:
         vdk_log_level = context.core_context.configuration.get_value(
             vdk_config.LOG_LEVEL_VDK
         )
+        syslog_url = context.core_context.configuration.get_value(SYSLOG_URL)
+        syslog_port = context.core_context.configuration.get_value(SYSLOG_PORT)
+        syslog_sock_type = context.core_context.configuration.get_value(
+            SYSLOG_SOCK_TYPE
+        )
+        syslog_enabled = context.core_context.configuration.get_value(SYSLOG_ENABLED)
         try:  # If logging initialization fails we want to attempt sending telemetry before exiting VDK
             configure_loggers(
                 job_name,
                 attempt_id,
                 log_config_type=log_config_type,
                 vdk_logging_level=vdk_log_level,
+                syslog_args=(syslog_url, syslog_port, syslog_sock_type, syslog_enabled),
             )
             log = logging.getLogger(__name__)
             log.debug(f"Initialized logging for log type {log_config_type}.")

diff --git a/projects/vdk-core/tests/vdk/internal/builtin_plugins/config/test_log_config.py b/projects/vdk-core/tests/vdk/internal/builtin_plugins/config/test_log_config.py
@@ -2,12 +2,16 @@
 # SPDX-License-Identifier: Apache-2.0
 import logging
 import pathlib
-from unittest.mock import MagicMock
+from unittest import mock
 
 import pytest
 from vdk.api.plugin.plugin_registry import IPluginRegistry
 from vdk.internal.builtin_plugins.config import vdk_config
 from vdk.internal.builtin_plugins.config.log_config import LoggingPlugin
+from vdk.internal.builtin_plugins.config.log_config import SYSLOG_ENABLED
+from vdk.internal.builtin_plugins.config.log_config import SYSLOG_PORT
+from vdk.internal.builtin_plugins.config.log_config import SYSLOG_SOCK_TYPE
+from vdk.internal.builtin_plugins.config.log_config import SYSLOG_URL
 from vdk.internal.builtin_plugins.run.data_job import JobArguments
 from vdk.internal.builtin_plugins.run.job_context import JobContext
 from vdk.internal.builtin_plugins.templates.template_impl import TemplatesImpl
@@ -28,32 +32,37 @@
     ),
 )
 def test_log_plugin(log_type, vdk_level, expected_vdk_level):
-    logging.getLogger().setLevel(logging.DEBUG)  # root level
-    logging.getLogger("vdk").setLevel(logging.NOTSET)  # reset vdk log level
-
-    log_plugin = LoggingPlugin()
-
-    store = StateStore()
-    store.set(CommonStoreKeys.ATTEMPT_ID, "attempt_id")
-
-    conf = (
-        ConfigurationBuilder()
-        .add(vdk_config.LOG_CONFIG, log_type)
-        .add(vdk_config.LOG_LEVEL_VDK, vdk_level)
-        .build()
-    )
-    core_context = CoreContext(MagicMock(spec=IPluginRegistry), conf, store)
-
-    job_context = JobContext(
-        "foo",
-        pathlib.Path("foo"),
-        core_context,
-        JobArguments([]),
-        TemplatesImpl("foo", core_context),
-    )
-
-    log_plugin.initialize_job(job_context)
-
-    assert (
-        logging.getLogger("vdk").getEffectiveLevel() == expected_vdk_level
-    ), "internal vdk logs must be set according to configuration option LOG_LEVEL_VDK but are not"
+    with mock.patch("socket.socket.connect"):
+        logging.getLogger().setLevel(logging.DEBUG)  # root level
+        logging.getLogger("vdk").setLevel(logging.NOTSET)  # reset vdk log level
+
+        log_plugin = LoggingPlugin()
+
+        store = StateStore()
+        store.set(CommonStoreKeys.ATTEMPT_ID, "attempt_id")
+
+        conf = (
+            ConfigurationBuilder()
+            .add(vdk_config.LOG_CONFIG, log_type)
+            .add(vdk_config.LOG_LEVEL_VDK, vdk_level)
+            .add(SYSLOG_URL, "localhost")
+            .add(SYSLOG_PORT, 514)
+            .add(SYSLOG_ENABLED, True)
+            .add(SYSLOG_SOCK_TYPE, "UDP")
+            .build()
+        )
+        core_context = CoreContext(mock.MagicMock(spec=IPluginRegistry), conf, store)
+
+        job_context = JobContext(
+            "foo",
+            pathlib.Path("foo"),
+            core_context,
+            JobArguments([]),
+            TemplatesImpl("foo", core_context),
+        )
+
+        log_plugin.initialize_job(job_context)
+
+        assert (
+            logging.getLogger("vdk").getEffectiveLevel() == expected_vdk_level
+        ), "internal vdk logs must be set according to configuration option LOG_LEVEL_VDK but are not"