vdk-core: support overriding configs with secrets (#3125)

## 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".

**Note:** The Configuration class and the CoreContext class are
annotated with `@dataclass(frozen=True)`. This enforces encapsulation,
so in order to mutate the Configuration object after it's created, we
have to add more public methods to the Configuration class.

https://docs.python.org/3/library/dataclasses.html#frozen-instances

## How was this tested?

Functional test
CI/CD

## What kind of change is this?

Feature/non-breaking

Signed-off-by: Dilyan Marinov <mdilyan@vmware.com>

Author: DeltaMichael

projects/vdk-core/README.md

@@ -35,6 +35,46 @@ vdk --help
 Plugins are a powerful way of extending or adapting Versatile Data Kit for all kinds of use-cases.
 For more information see [Plugins doc](../vdk-plugins/README.md).
 
+### A note on configs and secrets
+
+[Secrets also double as config keys.](https://github.com/vmware/versatile-data-kit/pull/3125). At the end of the
+`initialize_job` hook, the secrets provider is called and all secrets keys are fetched. Any configs that match a
+secrets key have their values overridden. However, if a plugin caches some config values in the `initialize_job`
+hook and re-uses them later, those values will not be overridden by secrets.
+
+**Example**
+
+This will not work
+
+```shell
+vdk secrets --set-prompt mytoken
+```
+
+```python
+    @hookimpl
+    def initialize_job(self, context: JobContext) -> None:
+        token = context.core_context.configuration.get_value("mytoken") # will be None if not set in config.ini
+
+        if token:
+            myapi.login(token)
+
+        # secrets override configs after initialize_job but before run_job
+```
+
+Plugins should do this instead
+
+```python
+    @hookimpl(hookwrapper=True)
+    def run_job(self, context: JobContext) -> Optional[ExecutionResult]:
+        # config is already overridden by secrets
+        token = context.core_context.configuration.get_value("mytoken")
+
+        if token:
+            myapi.login(token)
+```
+
+This is a limitation on the current implementation, which might change after https://github.com/vmware/versatile-data-kit/issues/3210
+
 ## Public interfaces
 
 Any backwards compatibility guarantees apply only to public interfaces.

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,14 @@
+# Copyright 2023-2024 Broadcom
+# SPDX-License-Identifier: Apache-2.0
+# 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_value(key, value)

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
@@ -70,6 +69,11 @@ def __getitem__(self, key: ConfigKey):
         key = _normalize_config_key(key)
         return self.get_value(key)
 
+    def override_value(self, key: ConfigKey, value: ConfigValue):
+        key = _normalize_config_key(key)
+        if key in self.__config_key_to_default_value:
+            self.__config_key_to_value[key] = value
+
     def get_value(self, key: ConfigKey) -> ConfigValue:
         """
         Return configuration value associated with a given key or None if it does not exists.

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,40 @@ 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"
+            secrets["IRRELEVANT_SECRET"] = "not relevant at all"
+            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)
+        # secrets that match config keys are overridden
+        assert config_log.config.get_value("log_exception_formatter") == "plain"
+        # secrets that do not match any config key do not leak in the config object
+        assert config_log.config.get_value("irrelevant_secret") is None

projects/vdk-plugins/vdk-plugin-control-cli/tests/test_secrets_plugin.py

@@ -61,14 +61,15 @@ def test_secrets_plugin_no_url_configured():
     ):
         runner = CliEntryBasedTestRunner(vdk_plugin_control_cli, secrets_plugin)
 
-        result: Result = runner.invoke(
-            ["run", jobs_path_from_caller_directory("simple-job")]
-        )
-        # job that do not use secrets should succeed
-        cli_assert_equal(0, result)
+        # TODO: Uncomment this after vdk-core release
+        # result: Result = runner.invoke(
+        #     ["run", jobs_path_from_caller_directory("simple-job")]
+        # )
+        # # job that doesn't explicitly use secrets should fail due to secrets overrides in core
+        # cli_assert_equal(1, result)
 
         result: Result = runner.invoke(
             ["run", jobs_path_from_caller_directory("secrets-job")]
         )
-        # but jobs that do use secrets should fail.
+        # job that explicitly uses secrets should also fail
         cli_assert_equal(1, result)