Skip to content

Commit

Permalink
vdk-structlog: add default logging format values
Browse files Browse the repository at this point in the history 
Why?

There should be separate default configs for local
and cloud runs

What?

Refactor structlog plugin to eliminate repeating code
Remove syslog config tests
Fix bug with vdk fields crashing logging for custom format
Add tests for vdk fields with custom format
Add default logging config to the data job base image
Note: The default config will not have effect unless vdk-structlog is installed

How was this tested?

Functional tests
CI

What kind of change is this?

Feature/non-breaking

Signed-off-by: Dilyan Marinov <[email protected]>
  • Loading branch information
Dilyan Marinov committed Jan 26, 2024
1 parent 59d2a85 commit 950d9d9
Show file tree
Hide file tree
Showing 6 changed files with 222 additions and 336 deletions.
11 changes: 8 additions & 3 deletions projects/control-service/projects/job-base-image/Dockerfile-data-job-base
View file
Original file line number Diff line number Diff line change
Expand Up @@ -11,9 +11,9 @@ WORKDIR /job
# Install native dependencies so that requirements in requirements.txt can be installed
# some (like openssl) should be pre-installed in the base image but let's be explicit
RUN set -ex \
&& apt-get update \
&& apt-get -y install --no-install-recommends \
build-essential openssl g++
&& apt-get update \
&& apt-get -y install --no-install-recommends \
build-essential openssl g++

# Install the native dependencies necessary for cx_Oracle python library
# See https://oracle.github.io/odpi/doc/installation.html#linux
Expand All @@ -27,6 +27,11 @@ RUN set -ex \
&& ldconfig \
&& apt-get purge -y --auto-remove curl unzip

# Setup default logging with vdk-structlog
ENV VDK_STRUCTLOG_FORMAT="console"
ENV VDK_STRUCTLOG_CONSOLE_CUSTOM_FORMAT="%(asctime)s [VDK] %(vdk_job_name)s [%(levelname)-5.5s] %(name)-30.30s %(filename)20.20s:%(lineno)-4.4s %(funcName)-16.16s[id:%(attempt_id)s] - %(message)s"


# libaio1 - LGPL-2.1+ https://developer.puri.sm/licenses/Librem5/Birch/libaio1/copyright
# curl - MIT/X* modified https://curl.se/docs/copyright.html
# unzip - MIT https://github.com/vipsoft/Unzip/blob/master/LICENSE
Expand Down
2 changes: 2 additions & 0 deletions projects/vdk-plugins/vdk-structlog/src/vdk/plugin/structlog/constants.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@
STRUCTLOG_LOGGING_FORMAT_DEFAULT = "console"

STRUCTLOG_LOGGING_METADATA_JOB = {
"attempt_id": "%(attempt_id)s",
"vdk_job_name": "%(vdk_job_name)s",
"vdk_step_name": "%(vdk_step_name)s",
"vdk_step_type": "%(vdk_step_type)s",
Expand Down Expand Up @@ -44,6 +45,7 @@
}

CONSOLE_STRUCTLOG_LOGGING_METADATA_JOB = {
"attempt_id": "[id:%(attempt_id)s]",
"vdk_job_name": "%(vdk_job_name)s",
"vdk_step_name": "%(vdk_step_name)s",
"vdk_step_type": "%(vdk_step_type)s",
Expand Down
291 changes: 125 additions & 166 deletions projects/vdk-plugins/vdk-structlog/src/vdk/plugin/structlog/structlog_plugin.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,7 @@
from vdk.plugin.structlog.filters import AttributeAdder
from vdk.plugin.structlog.formatters import create_formatter
from vdk.plugin.structlog.log_level_utils import set_non_root_log_levels
from vdk.plugin.structlog.syslog_config import configure_syslog_handler
from vdk.plugin.structlog.syslog_config import create_syslog_handler

"""
Handlers
Expand Down Expand Up @@ -77,6 +77,74 @@ def process(self, msg, kwargs):
"""


def _get_syslog_config(context: CoreContext):
syslog_enabled = context.configuration.get_value(
SYSLOG_ENABLED_KEY
)
syslog_host = context.configuration.get_value(SYSLOG_HOST_KEY)
syslog_port = context.configuration.get_value(SYSLOG_PORT_KEY)
syslog_protocol = context.configuration.get_value(
SYSLOG_PROTOCOL_KEY
)

return syslog_enabled, syslog_host, syslog_port, syslog_protocol


def _create_formatter_and_metadata_filter(context: CoreContext):
metadata_keys = context.configuration.get_value(STRUCTLOG_LOGGING_METADATA_KEY)
logging_formatter = context.configuration.get_value(
STRUCTLOG_LOGGING_FORMAT_KEY
)
custom_format_string = context.configuration.get_value(
STRUCTLOG_CONSOLE_LOG_PATTERN
)

formatter, metadata_filter = create_formatter(
logging_formatter, metadata_keys, custom_format_string
)
return formatter, metadata_filter


def _configure_non_root_log_levels(context: CoreContext):
if logging.getLogger().getEffectiveLevel() == logging.NOTSET:
log_level = os.environ.get(
"LOG_LEVEL_VDK", os.environ.get("VDK_LOG_LEVEL_VDK", "INFO")
).upper()
logging.getLogger().setLevel(log_level)
vdk_log_level = context.configuration.get_value(vdk_config.LOG_LEVEL_VDK)
if vdk_log_level is None:
vdk_log_level = logging.getLogger().getEffectiveLevel()
log_level_module = context.configuration.get_value(vdk_config.LOG_LEVEL_MODULE)
set_non_root_log_levels(vdk_log_level, log_level_module)


def _clear_root_logger_handlers():
root_logger = logging.getLogger()
handlers_to_remove = root_logger.handlers
for handler in handlers_to_remove:
root_logger.removeHandler(handler)


def _configure_root_logger(context: CoreContext, formatter, *filters):
_clear_root_logger_handlers()
root_logger = logging.getLogger()

handlers = [logging.StreamHandler(sys.stderr)]

syslog_config = _get_syslog_config(context)
syslog_handler = create_syslog_handler(
*syslog_config
)
if syslog_handler:
handlers.append(syslog_handler)

for handler in handlers:
for f in filters:
handler.addFilter(f)
handler.setFormatter(formatter)
root_logger.addHandler(handler)


class StructlogPlugin:
@hookimpl(tryfirst=True)
def vdk_configure(self, config_builder: ConfigurationBuilder):
Expand Down Expand Up @@ -134,208 +202,99 @@ def vdk_configure(self, config_builder: ConfigurationBuilder):

@hookimpl
def vdk_initialize(self, context: CoreContext):
if logging.getLogger().getEffectiveLevel() == logging.NOTSET:
log_level = os.environ.get(
"LOG_LEVEL_VDK", os.environ.get("VDK_LOG_LEVEL_VDK", "INFO")
).upper()
logging.getLogger().setLevel(log_level)
vdk_log_level = context.configuration.get_value(vdk_config.LOG_LEVEL_VDK)
if vdk_log_level is None:
vdk_log_level = logging.getLogger().getEffectiveLevel()
log_level_module = context.configuration.get_value(vdk_config.LOG_LEVEL_MODULE)

set_non_root_log_levels(vdk_log_level, log_level_module)

metadata_keys = context.configuration.get_value(STRUCTLOG_LOGGING_METADATA_KEY)
logging_formatter = context.configuration.get_value(
STRUCTLOG_LOGGING_FORMAT_KEY
)
custom_format_string = context.configuration.get_value(
STRUCTLOG_CONSOLE_LOG_PATTERN
)
_configure_non_root_log_levels(context)
formatter, metadata_filter = _create_formatter_and_metadata_filter(context)

formatter, metadata_filter = create_formatter(
logging_formatter, metadata_keys, custom_format_string
)
attempt_id = context.state.get(CommonStoreKeys.ATTEMPT_ID)
attempt_id_adder = AttributeAdder("attempt_id", attempt_id)

root_logger = logging.getLogger()
root_logger.removeHandler(root_logger.handlers[0])
# Add placeholder values in case of custom format string
job_name_adder = AttributeAdder("vdk_job_name", "")
step_name_adder = AttributeAdder("vdk_step_name", "")
step_type_adder = AttributeAdder("vdk_step_type", "")

handler = logging.StreamHandler(sys.stderr)
handler.addFilter(metadata_filter)
handler.setFormatter(formatter)

root_logger.addHandler(handler)
_configure_root_logger(context, formatter, attempt_id_adder, job_name_adder, step_name_adder,
step_type_adder, metadata_filter)

@hookimpl(hookwrapper=True)
def initialize_job(self, context: JobContext) -> None:
metadata_keys = context.core_context.configuration.get_value(
STRUCTLOG_LOGGING_METADATA_KEY
)
job_name = context.name
if logging.getLogger().getEffectiveLevel() == logging.NOTSET:
log_level = os.environ.get(
"LOG_LEVEL_VDK", os.environ.get("VDK_LOG_LEVEL_VDK", "INFO")
).upper()
logging.getLogger().setLevel(log_level)
vdk_log_level = context.core_context.configuration.get_value(
vdk_config.LOG_LEVEL_VDK
)
if vdk_log_level is None:
vdk_log_level = logging.getLogger().getEffectiveLevel()
log_level_module = context.core_context.configuration.get_value(
vdk_config.LOG_LEVEL_MODULE
)

set_non_root_log_levels(vdk_log_level, log_level_module)
_configure_non_root_log_levels(context.core_context)
formatter, metadata_filter = _create_formatter_and_metadata_filter(context.core_context)

logging_formatter = context.core_context.configuration.get_value(
STRUCTLOG_LOGGING_FORMAT_KEY
)
custom_format_string = context.core_context.configuration.get_value(
STRUCTLOG_CONSOLE_LOG_PATTERN
)
attempt_id = context.core_context.state.get(CommonStoreKeys.ATTEMPT_ID)
attempt_id_adder = AttributeAdder("attempt_id", attempt_id)

formatter, metadata_filter = create_formatter(
logging_formatter, metadata_keys, custom_format_string
)
job_name = context.name
job_name_adder = AttributeAdder("vdk_job_name", job_name)

root_logger = logging.getLogger()
root_logger.removeHandler(root_logger.handlers[0])
# Add placeholder values in case of custom format string
step_name_adder = AttributeAdder("vdk_step_name", "")
step_type_adder = AttributeAdder("vdk_step_type", "")

handler = logging.StreamHandler(sys.stderr)
handler.setFormatter(formatter)
handler.addFilter(job_name_adder)
handler.addFilter(metadata_filter)

root_logger.addHandler(handler)

syslog_enabled = context.core_context.configuration.get_value(
SYSLOG_ENABLED_KEY
)
syslog_host = context.core_context.configuration.get_value(SYSLOG_HOST_KEY)
syslog_port = context.core_context.configuration.get_value(SYSLOG_PORT_KEY)
syslog_protocol = context.core_context.configuration.get_value(
SYSLOG_PROTOCOL_KEY
)
attempt_id = context.core_context.state.get(CommonStoreKeys.ATTEMPT_ID)
syslog_handler = configure_syslog_handler(
syslog_enabled,
syslog_host,
syslog_port,
syslog_protocol,
job_name,
attempt_id,
)
if syslog_handler:
root_logger.addHandler(syslog_handler)
_configure_root_logger(context.core_context, formatter, attempt_id_adder, job_name_adder, step_name_adder,
step_type_adder, metadata_filter)

out: HookCallResult
out = yield

root_logger.removeHandler(handler)
if syslog_handler:
root_logger.removeHandler(syslog_handler)
_clear_root_logger_handlers()

@hookimpl(hookwrapper=True)
def run_job(self, context: JobContext) -> Optional[ExecutionResult]:
metadata_keys = context.core_context.configuration.get_value(
STRUCTLOG_LOGGING_METADATA_KEY
)
job_name = context.name
if logging.getLogger().getEffectiveLevel() == logging.NOTSET:
log_level = os.environ.get(
"LOG_LEVEL_VDK", os.environ.get("VDK_LOG_LEVEL_VDK", "INFO")
).upper()
logging.getLogger().setLevel(log_level)
vdk_log_level = context.core_context.configuration.get_value(
vdk_config.LOG_LEVEL_VDK
)
if vdk_log_level is None:
vdk_log_level = logging.getLogger().getEffectiveLevel()
log_level_module = context.core_context.configuration.get_value(
vdk_config.LOG_LEVEL_MODULE
)

set_non_root_log_levels(vdk_log_level, log_level_module)
_configure_non_root_log_levels(context.core_context)
formatter, metadata_filter = _create_formatter_and_metadata_filter(context.core_context)

logging_formatter = context.core_context.configuration.get_value(
STRUCTLOG_LOGGING_FORMAT_KEY
)
custom_format_string = context.core_context.configuration.get_value(
STRUCTLOG_CONSOLE_LOG_PATTERN
)
attempt_id = context.core_context.state.get(CommonStoreKeys.ATTEMPT_ID)
attempt_id_adder = AttributeAdder("attempt_id", attempt_id)

formatter, metadata_filter = create_formatter(
logging_formatter, metadata_keys, custom_format_string
)
job_name = context.name
job_name_adder = AttributeAdder("vdk_job_name", job_name)

root_logger = logging.getLogger()
root_logger.removeHandler(root_logger.handlers[0])

handler = logging.StreamHandler(sys.stderr)
handler.setFormatter(formatter)
handler.addFilter(job_name_adder)
handler.addFilter(metadata_filter)

root_logger.addHandler(handler)
# Add placeholder values in case of custom format string
step_name_adder = AttributeAdder("vdk_step_name", "")
step_type_adder = AttributeAdder("vdk_step_type", "")

syslog_enabled = context.core_context.configuration.get_value(
SYSLOG_ENABLED_KEY
)
syslog_host = context.core_context.configuration.get_value(SYSLOG_HOST_KEY)
syslog_port = context.core_context.configuration.get_value(SYSLOG_PORT_KEY)
syslog_protocol = context.core_context.configuration.get_value(
SYSLOG_PROTOCOL_KEY
)
attempt_id = context.core_context.state.get(CommonStoreKeys.ATTEMPT_ID)
syslog_handler = configure_syslog_handler(
syslog_enabled,
syslog_host,
syslog_port,
syslog_protocol,
job_name,
attempt_id,
)
if syslog_handler:
root_logger.addHandler(syslog_handler)
_configure_root_logger(context.core_context, formatter, attempt_id_adder, job_name_adder, step_name_adder,
step_type_adder, metadata_filter)

out: HookCallResult
out = yield

# do not remove the handler, we need it until the end
# do not remove handlers, we need them until the end

@hookimpl(hookwrapper=True)
def run_step(self, context: JobContext, step: Step) -> Optional[StepResult]:
root_logger = logging.getLogger()
handler = root_logger.handlers[0]
metadata_filter = None
# make sure the metadata filter executes last
# so that step_name and step_type are filtered if necessary
metadata_filter_result = [
f for f in handler.filters if f.name == "metadata_filter"
]
if metadata_filter_result:
metadata_filter = metadata_filter_result[0]

if metadata_filter:
handler.removeFilter(metadata_filter)

root_logger = logging.getLogger()
step_name_adder = AttributeAdder("vdk_step_name", step.name)
step_type_adder = AttributeAdder("vdk_step_type", step.type)
handler.addFilter(step_name_adder)
handler.addFilter(step_type_adder)

# make sure the metadata filter executes last
# so that step_name and step_type are filtered if necessary
if metadata_filter:
handler.addFilter(metadata_filter)
for handler in root_logger.handlers:
metadata_filter = None
# make sure the metadata filter executes last
# so that step_name and step_type are filtered if necessary
metadata_filter_result = [
f for f in handler.filters if f.name == "metadata_filter"
]
if metadata_filter_result:
metadata_filter = metadata_filter_result[0]

if metadata_filter:
handler.removeFilter(metadata_filter)

handler.addFilter(step_name_adder)
handler.addFilter(step_type_adder)

# make sure the metadata filter executes last
# so that step_name and step_type are filtered if necessary
if metadata_filter:
handler.addFilter(metadata_filter)
out: HookCallResult
out = yield
handler.removeFilter(step_name_adder)
handler.removeFilter(step_type_adder)
for handler in root_logger.handlers:
handler.removeFilter(step_name_adder)
handler.removeFilter(step_type_adder)


@hookimpl
Expand Down
Loading

0 comments on commit 950d9d9

Please sign in to comment.