Merge pull request spotify#63 from spotify/dsimon/config-exec

[cli, exec, lib] Remove effective-klio-job.yaml

cli/src/klio_cli/cli.py

@@ -139,9 +139,6 @@ def run_job(klio_config, config_meta, **kwargs):
         config_meta.job_dir, kwargs.get("image_tag")
     )
     image_tag = kwargs.get("image_tag") or git_sha
-    if config_meta.config_file:
-        basename = os.path.basename(config_meta.config_file)
-        image_tag = "{}-{}".format(image_tag, basename)
 
     runtime_config = DockerRuntimeConfig(
         image_tag=image_tag,

cli/src/klio_cli/commands/base.py

@@ -15,9 +15,12 @@
 
 import logging
 import os
+import tempfile
 
 import docker
+import yaml
 
+from klio_cli.commands.job import configuration
 from klio_cli.utils import docker_utils
 
 
@@ -28,15 +31,21 @@ class BaseDockerizedPipeline(object):
     CONTAINER_GCP_CRED_PATH = os.path.join("/usr", GCP_CRED_FILE)
     CONTAINER_JOB_DIR = "/usr/src/app"
     DOCKER_LOGGER_NAME = "klio.base_docker_pipeline"
+    # path where the temp config-file is mounted into klio-exec's container
+    MATERIALIZED_CONFIG_PATH = "/usr/src/config/materialized_config.yaml"
 
     def __init__(self, job_dir, klio_config, docker_runtime_config):
         self.job_dir = job_dir
-        # TODO: this should be KlioConfig object
         self.klio_config = klio_config
         self.docker_runtime_config = docker_runtime_config
         self._docker_client = None
         self._docker_logger = self._get_docker_logger()
 
+        # if this is set to true, running the command will generate a temp file
+        # and mount it to the container
+        self.requires_config_file = True
+        self.materialized_config_file = None
+
     @property
     def _full_image_name(self):
         return "{}:{}".format(
@@ -82,7 +91,7 @@ def _get_volumes(self):
         host_cred_path = os.path.join(
             os.environ.get("HOME"), BaseDockerizedPipeline.HOST_GCP_CRED_PATH
         )
-        return {
+        volumes = {
             host_cred_path: {
                 "bind": BaseDockerizedPipeline.CONTAINER_GCP_CRED_PATH,
                 "mode": "rw",  # Fails if no write access
@@ -93,15 +102,33 @@ def _get_volumes(self):
             },
         }
 
+        if self.materialized_config_file is not None:
+            volumes[self.materialized_config_file.name] = {
+                "bind": BaseDockerizedPipeline.MATERIALIZED_CONFIG_PATH,
+                "mode": "rw",
+            }
+
+        return volumes
+
     def _get_command(self, *args, **kwargs):
         raise NotImplementedError
 
+    def _add_base_args(self, command):
+        if self.requires_config_file:
+            command.extend(
+                [
+                    "--config-file",
+                    BaseDockerizedPipeline.MATERIALIZED_CONFIG_PATH,
+                ]
+            )
+        return command
+
     def _get_docker_runflags(self, *args, **kwargs):
         return {
             "image": self._full_image_name,
             # overwrite fnapi image entrypoint
             "entrypoint": self.ENTRYPOINT,
-            "command": self._get_command(*args, **kwargs),
+            "command": self._add_base_args(self._get_command(*args, **kwargs)),
             # mount klio code
             "volumes": self._get_volumes(),
             "environment": self._get_environment(),
@@ -149,10 +176,24 @@ def _check_gcp_credentials_exist(self):
                 )
             )
 
+    def _write_effective_config(self):
+        if self.requires_config_file:
+            self.materialized_config_file = tempfile.NamedTemporaryFile(
+                prefix="/tmp/", mode="w", delete=False
+            )
+            yaml.dump(
+                self.klio_config._raw,
+                stream=self.materialized_config_file,
+                Dumper=configuration.IndentListDumper,
+                default_flow_style=False,
+                sort_keys=False,
+            )
+
     def run(self, *args, **kwargs):
         # bail early
         self._check_gcp_credentials_exist()
         self._check_docker_setup()
+        self._write_effective_config()
 
         self._setup_docker_image()
         runflags = self._get_docker_runflags(*args, **kwargs)

cli/src/klio_cli/commands/job/run.py

@@ -35,6 +35,7 @@ def __init__(
     ):
         super().__init__(job_dir, klio_config, docker_runtime_config)
         self.run_job_config = run_job_config
+        self.requires_config_file = True
 
     @staticmethod
     def _try_container_kill(container):
@@ -115,14 +116,6 @@ def _get_command(self):
         ):  # don't do anything if `None`
             command.append("--no-update")
 
-        if self.docker_runtime_config.config_file_override:
-            command.extend(
-                [
-                    "--config-file",
-                    self.docker_runtime_config.config_file_override,
-                ]
-            )
-
         return command
 
     def _setup_docker_image(self):

cli/tests/commands/job/test_run.py

@@ -197,8 +197,7 @@ def test_get_environment(run_pipeline):
 
 
 @pytest.mark.parametrize(
-    "config_file,exp_config_flags",
-    ((None, []), ("klio-job2.yaml", ["--config-file", "klio-job2.yaml"])),
+    "config_file", (None, "klio-job2.yaml"),
 )
 @pytest.mark.parametrize(
     "image_tag,exp_image_flags",
@@ -219,7 +218,6 @@ def test_get_command(
     image_tag,
     exp_image_flags,
     config_file,
-    exp_config_flags,
     run_pipeline,
     monkeypatch,
 ):
@@ -236,7 +234,6 @@ def test_get_command(
     exp_command.extend(exp_update_flag)
     exp_command.extend(exp_runner_flag)
     exp_command.extend(exp_image_flags)
-    exp_command.extend(exp_config_flags)
 
     assert sorted(exp_command) == sorted(run_pipeline._get_command())
 

cli/tests/commands/test_base.py

@@ -71,6 +71,13 @@ def mock_docker_client(mocker):
     return mock_client
 
 
+@pytest.fixture
+def mock_materialized_config_file(mocker):
+    mock = mocker.Mock()
+    mock.name = "test-config"
+    return mock
+
+
 @pytest.fixture
 def base_pipeline(
     klio_config,
@@ -108,6 +115,10 @@ def expected_volumes():
             "mode": "rw",
         },
         "/test/dir/jobs/test_run_job": {"bind": "/usr/src/app", "mode": "rw"},
+        "test-config": {
+            "bind": base.BaseDockerizedPipeline.MATERIALIZED_CONFIG_PATH,
+            "mode": "rw",
+        },
     }
 
 
@@ -136,7 +147,18 @@ def test_get_environment(base_pipeline, expected_envs):
     assert expected_envs == base_pipeline._get_environment()
 
 
-def test_get_volumes(base_pipeline, expected_volumes):
+def test_get_volumes(
+    base_pipeline,
+    expected_volumes,
+    mocker,
+    monkeypatch,
+    mock_materialized_config_file,
+):
+    monkeypatch.setattr(
+        base_pipeline,
+        "materialized_config_file",
+        mock_materialized_config_file,
+    )
     assert expected_volumes == base_pipeline._get_volumes()
 
 
@@ -145,16 +167,42 @@ def test_get_command(base_pipeline):
         base_pipeline._get_command()
 
 
+@pytest.mark.parametrize("requires_config", (False, True))
 def test_get_docker_runflags(
-    base_pipeline, expected_volumes, expected_envs, mocker, monkeypatch
+    base_pipeline,
+    expected_volumes,
+    expected_envs,
+    mocker,
+    monkeypatch,
+    requires_config,
+    mock_materialized_config_file,
 ):
     mock_get_command = mocker.Mock(return_value=["command"])
     monkeypatch.setattr(base_pipeline, "_get_command", mock_get_command)
 
+    base_pipeline.requires_config_file = requires_config
+
+    expected_command = ["command"]
+
+    if requires_config:
+        monkeypatch.setattr(
+            base_pipeline,
+            "materialized_config_file",
+            mock_materialized_config_file,
+        )
+        expected_command.extend(
+            [
+                "--config-file",
+                base.BaseDockerizedPipeline.MATERIALIZED_CONFIG_PATH,
+            ]
+        )
+    else:
+        expected_volumes.pop("test-config")
+
     exp_runflags = {
         "image": "test-image:foo-123",
         "entrypoint": "klioexec",
-        "command": ["command"],
+        "command": expected_command,
         "volumes": expected_volumes,
         "environment": expected_envs,
         "detach": True,
@@ -204,7 +252,9 @@ def test_setup_docker_image(
         mock_build_docker_image.assert_not_called()
 
 
-def test_check_docker_setup(base_pipeline, mocker, monkeypatch):
+def test_check_docker_setup(
+    base_pipeline, mock_docker_client, mocker, monkeypatch
+):
     mock_check_docker_conn = mocker.Mock()
     monkeypatch.setattr(
         base.docker_utils, "check_docker_connection", mock_check_docker_conn
@@ -216,6 +266,10 @@ def test_check_docker_setup(base_pipeline, mocker, monkeypatch):
         mock_check_dockerfile_present,
     )
 
+    monkeypatch.setattr(
+        base.docker, "from_env", mock_docker_client,
+    )
+
     base_pipeline._check_docker_setup()
 
     mock_check_docker_conn.assert_called_once_with(