vdk-core: support overriding configs with secrets

Why?

VDK doesn't provide a way to set sensitive configuration like passwords, such as trino_password.
The only way to currently do this is by adding config keys and fetching the values from secrets.

What?

Add a plugin that reconfigures the Configuration object in CoreContext based on secrets. Do this
in the initialize_job hook. In this setup, secrets override options set by regular configs. For example
if you set trino_password to "password" in config.ini, but also have a secret called trino_passowrd="another password",
the value of trino_password will be "another_passowrd".

How was this tested?

Functional test
CI/CD

What kind of change is this?

Feature/non-breaking

Signed-off-by: Dilyan Marinov <[email protected]>

projects/vdk-core/src/vdk/internal/builtin_plugins/builtin_hook_impl.py

@@ -16,6 +16,7 @@
 from vdk.internal.builtin_plugins.config import vdk_config
 from vdk.internal.builtin_plugins.config.config_help import ConfigHelpPlugin
 from vdk.internal.builtin_plugins.config.log_config import LoggingPlugin
+from vdk.internal.builtin_plugins.config.secrets_config import SecretsConfigPlugin
 from vdk.internal.builtin_plugins.config.vdk_config import CoreConfigDefinitionPlugin
 from vdk.internal.builtin_plugins.config.vdk_config import EnvironmentVarsConfigPlugin
 from vdk.internal.builtin_plugins.config.vdk_config import JobConfigIniPlugin
@@ -130,6 +131,7 @@ def vdk_start(plugin_registry: PluginRegistry, command_line_args: List) -> None:
     # TODO: should be in run package only
     plugin_registry.add_hook_specs(JobRunHookSpecs)
     plugin_registry.load_plugin_with_hooks_impl(JobConfigIniPlugin())
+    plugin_registry.load_plugin_with_hooks_impl(SecretsConfigPlugin())
     plugin_registry.load_plugin_with_hooks_impl(TerminationMessageWriterPlugin())
     plugin_registry.load_plugin_with_hooks_impl(JobRunSummaryOutputPlugin())
     # connection plugins

projects/vdk-core/src/vdk/internal/builtin_plugins/config/secrets_config.py

@@ -0,0 +1,13 @@
+# Copyright 2021-2024 VMware, Inc.
+# SPDX-License-Identifier: Apache-2.0
+from vdk.api.plugin.hook_markers import hookimpl
+from vdk.internal.builtin_plugins.run.job_context import JobContext
+
+
+class SecretsConfigPlugin:
+    @hookimpl(trylast=True)
+    def initialize_job(self, context: JobContext):
+        secrets = context.job_input.get_all_secrets()
+        for key, value in secrets.items():
+            context.core_context.configuration.override_with_secret(key, value)
+        context.core_context.configuration.lock_overrides()

projects/vdk-core/src/vdk/internal/core/config.py

@@ -3,7 +3,6 @@
 from __future__ import annotations
 
 import logging
-from collections import OrderedDict
 from dataclasses import dataclass
 from dataclasses import field
 from typing import Any
@@ -65,11 +64,25 @@ class Configuration:
         default_factory=dict
     )
     __config_key_to_sensitive: dict[ConfigKey, bool] = field(default_factory=dict)
+    __config_key_to_can_override: dict[ConfigKey, bool] = field(default_factory=dict)
 
     def __getitem__(self, key: ConfigKey):
         key = _normalize_config_key(key)
         return self.get_value(key)
 
+    def __can_override(self, key: ConfigKey):
+        return self.__config_key_to_can_override.get(key, False)
+
+    def override_with_secret(self, key: ConfigKey, value: ConfigValue):
+        key = _normalize_config_key(key)
+        if self.__can_override(key):
+            self.__config_key_to_value[key] = value
+            self.__config_key_to_can_override[key] = False
+
+    def lock_overrides(self):
+        for key in self.__config_key_to_can_override.keys():
+            self.__config_key_to_can_override[key] = False
+
     def get_value(self, key: ConfigKey) -> ConfigValue:
         """
         Return configuration value associated with a given key or None if it does not exists.
@@ -157,12 +170,14 @@ class ConfigurationBuilder:
     __config_key_to_value: dict[ConfigKey, ConfigValue]
     __config_key_to_default_value: dict[ConfigKey, ConfigValue]
     __config_key_to_sensitive: dict[ConfigKey, bool]
+    __config_key_to_can_override: dict[ConfigKey, bool]
 
     def __init__(self):
         self.__config_key_to_description = dict()
         self.__config_key_to_default_value = dict()
         self.__config_key_to_value = dict()
         self.__config_key_to_sensitive = dict()
+        self.__config_key_to_can_override = dict()
 
     def add(
         self,
@@ -188,6 +203,7 @@ def add(
         """
         key = _normalize_config_key(key)
         self.__config_key_to_default_value[key] = default_value
+        self.__config_key_to_can_override[key] = True
         self.__config_key_to_sensitive[key] = is_sensitive
         if description and show_default_value:
             self.__add_public(key, description, is_sensitive, default_value)
@@ -258,4 +274,5 @@ def build(self) -> Configuration:
             self.__config_key_to_value,
             self.__config_key_to_default_value,
             self.__config_key_to_sensitive,
+            self.__config_key_to_can_override,
         )

projects/vdk-core/tests/functional/run/test_run_configuration.py

@@ -13,6 +13,7 @@
 from vdk.internal.core.statestore import CommonStoreKeys
 from vdk.plugin.test_utils.util_funcs import cli_assert_equal
 from vdk.plugin.test_utils.util_funcs import CliEntryBasedTestRunner
+from vdk.plugin.test_utils.util_plugins import TestSecretsServiceClient
 
 
 @mock.patch.dict(
@@ -100,3 +101,36 @@ def run_job(self, context: JobContext) -> Optional[ExecutionResult]:
             config_log.config.get_value("other_config")
             == "other-config-from-config-ini"
         )
+
+
+def test_run_secrets_override_config_vars():
+    class DebugConfigLog:
+        def __init__(self):
+            self.config = None
+            self.secrets_client = TestSecretsServiceClient()
+
+        @hookimpl(tryfirst=True)
+        def initialize_job(self, context: JobContext):
+            context.secrets.set_secrets_factory_method(
+                "default", lambda: self.secrets_client
+            )
+            secrets = context.job_input.get_all_secrets()
+            secrets["LOG_EXCEPTION_FORMATTER"] = "plain"
+            context.job_input.set_all_secrets(secrets)
+
+        @hookimpl(trylast=True)
+        def run_job(self, context: JobContext) -> Optional[ExecutionResult]:
+            self.config = context.core_context.configuration
+            return None  # continue with next hook impl.
+
+    with mock.patch.dict(
+        os.environ,
+        {
+            "VDK_LOG_EXCEPTION_FORMATTER": "pretty",
+        },
+    ):
+        config_log = DebugConfigLog()
+        runner = CliEntryBasedTestRunner(config_log)
+        result: Result = runner.invoke(["run", util.job_path("simple-job")])
+        cli_assert_equal(0, result)
+        assert config_log.config.get_value("log_exception_formatter") == "plain"