salt: Avoid duplicating static pod manifests

When using `metalk8s.static_pod_managed`, we call `file.managed` behind
the scenes. This state does a lot of magic, including creating a
temporary file with the new contents before replacing the old file.
This temp file gets created **in the same directory** as the managed
file by default, so it gets picked up by `kubelet` as if it were
another static Pod to manage. If the replacement occurs too late,
`kubelet` may have already created another Pod for the temp file, and
may not be able to "remember" the old Pod, hence not cleaning it up.
This results in "rogue containers", which can create issues (e.g.
preventing new containers from binding some ports on the host).

This commit reimplements the 'file.managed' state in a minimal fashion,
to ensure the temporary file used for making an "atomic replace" is
ignored by kubelet. Note that it requires us to also reimplement the
'file.manage_file' execution function, since it always relies on the
existing "atomic copy" operation from `salt.utils.files.copyfile`.

Fixes: #2840

salt/_modules/metalk8s.py

@@ -3,17 +3,20 @@
 Module for handling MetalK8s specific calls.
 '''
 import functools
+import itertools
 import logging
 import os.path
 import re
 import six
 import socket
+import tempfile
 import time
 
 from salt.pillar import get_pillar
 from salt.exceptions import CommandExecutionError
 import salt.utils.args
 import salt.utils.files
+from salt.utils.hashutils import get_hash
 
 log = logging.getLogger(__name__)
 
@@ -355,3 +358,164 @@ def cmp_sorted(*args, **kwargs):
         kwargs['key'] = functools.cmp_to_key(kwargs.pop('cmp'))
 
     return sorted(*args, **kwargs)
+
+
+def _error(ret, err_msg):
+    ret["result"] = False
+    ret["comment"] = err_msg
+    return ret
+
+
+def _atomic_copy(source, dest, tmp_prefix="."):  # pragma: no cover
+    """Minimalistic implementation of an atomic copy operation.
+
+    First, we write the contents of the source file to a temporary file, which
+    is saved in the same directory as the target, with a custom prefix.
+    Finally, we link the temporary file contents to the destination filename.
+    """
+    base_name = os.path.basename(dest)
+    dir_name = os.path.dirname(dest)
+
+    with open(source, mode='rb') as src_file:
+        contents = src_file.read()
+
+    with tempfile.NamedTemporaryFile(
+        prefix=tmp_prefix, dir=dir_name, delete=False,
+    ) as tmp_file:
+        tmp_file.write(contents)
+
+    try:
+        os.replace(tmp_file.name, dest)
+    except OSError as exc:
+        os.remove(tmp_file.name)
+        raise
+
+
+def manage_static_pod_manifest(
+    name, source_filename, source, source_sum, saltenv='base',
+):
+    """Checks a manifest file and applies changes if necessary.
+
+    Implementation derived from saltstack/salt salt.modules.file.manage_file.
+
+    name:
+        Path to the static pod manifest.
+
+    source_filename:
+        Path to the cached source file on the minion. The hash sum of this
+        file will be compared with the `source_sum` argument to determine
+        whether the source file should be fetched again using `cp.cache_file`.
+
+    source:
+        Reference for the source file (from the master).
+
+    source_sum:
+        Hash sum for the source file.
+
+    CLI Example:
+    .. code-block:: bash
+        salt '*' metalk8s.manage_static_pod_manifest /etc/kubernetes/manifests/etcd.yaml '' salt://metalk8s/kubernetes/etcd/files/manifest.yaml '{hash_type: 'md5', 'hsum': <md5sum>}' saltenv=metalk8s-2.7.0
+    """
+    ret = {"name": name, "changes": {}, "comment": "", "result": True}
+
+    def _clean_tmp(sfn):
+        if sfn.startswith(os.path.join(
+            tempfile.gettempdir(), salt.utils.files.TEMPFILE_PREFIX
+        )):
+            # Don't remove if it exists in file_roots (any saltenv)
+            all_roots = itertools.chain.from_iterable(
+                __opts__["file_roots"].values()
+            )
+            in_roots = any(sfn.startswith(root) for root in all_roots)
+            # Only clean up files that exist
+            if os.path.exists(sfn) and not in_roots:
+                os.remove(sfn)
+
+    target_dir = os.path.dirname(name)
+    target_exists = os.path.isfile(name) or os.path.islink(name)
+    hash_type = source_sum.get("hash_type", __opts__["hash_type"])
+
+    if not source:
+        return _error(ret, "Must provide a source")
+    if not os.path.isdir(target_dir):
+        return _error(
+            ret, "Target directory {} does not exist".format(target_dir)
+        )
+
+    if source_filename:
+        # File should be already cached, verify its checksum
+        cached_hash = get_hash(source_filename, form=hash_type)
+        if source_sum.get("hsum") != cached_hash:
+            log.debug(
+                "Cached source file {} does not match expected checksum, "
+                "will fetch it again".format(source_filename)
+            )
+            source_filename = ''  # Reset source filename to fetch it again
+
+    if not source_filename:
+        # File is not present or outdated, cache it
+        source_filename = __salt__["cp.cache_file"](source, saltenv)
+        if not source_filename:
+            return _error(
+                ret, "Source file '{}' not found".format(source)
+            )
+
+        # Recalculate source sum now that file has been cached
+        source_sum = {
+            "hash_type": hash_type,
+            "hsum": get_hash(source_filename, form=hash_type)
+        }
+
+    # Check changes if the target file exists
+    if target_exists:
+        if os.path.islink(name):
+            real_name = os.path.realpath(name)
+        else:
+            real_name = name
+
+        target_hash = get_hash(real_name, hash_type)
+        source_hash = source_sum.get("hsum")
+
+        # Check if file needs to be replaced
+        if source_hash != target_hash:
+            # Print a diff equivalent to diff -u old new
+            if __salt__["config.option"]("obfuscate_templates"):
+                ret["changes"]["diff"] = "<Obfuscated Template>"
+            else:
+                try:
+                    ret["changes"]["diff"] = __salt__["file.get_diff"](
+                        real_name, source_filename, show_filenames=False
+                    )
+                except CommandExecutionError as exc:
+                    ret["changes"]["diff"] = exc.strerror
+
+    else:  # target file does not exist
+        ret["changes"]["diff"] = "New file"
+        real_name = name
+
+    if ret["changes"] and not __opts__["test"]:
+        # The file needs to be replaced
+        try:
+            _atomic_copy(source_filename, real_name)
+        except OSError as io_error:
+            _clean_tmp(source_filename)
+            return _error(ret, "Failed to commit change: {}".format(io_error))
+
+    # Always enforce perms, even if no changes to contents (this module is
+    # idempotent)
+    ret, _ = __salt__["file.check_perms"](
+        name, ret, user="root", group="root", mode="0600",
+    )
+
+    if ret["changes"]:
+        if __opts__["test"]:
+            ret["comment"] = "File {} would be updated".format(name)
+        else:
+            ret["comment"] = "File {} updated".format(name)
+
+    elif not ret["changes"] and ret["result"]:
+        ret["comment"] = "File {} is in the correct state".format(name)
+
+    if source_filename:
+        _clean_tmp(source_filename)
+    return ret

salt/_states/metalk8s.py

@@ -1,28 +1,39 @@
 """Custom states for MetalK8s."""
 
+import logging
+import tempfile
 import time
+import traceback
 import re
 
+from salt.ext import six
+
 
 __virtualname__ = "metalk8s"
+log = logging.getLogger(__name__)
 
 
 def __virtual__():
     return __virtualname__
 
 
+def _error(ret, err_msg):
+    ret["result"] = False
+    ret["comment"] = err_msg
+    return ret
+
+
 def static_pod_managed(name,
                        source,
                        config_files=None,
                        config_files_opt=None,
-                       context=None,
-                       **kwargs):
+                       context=None):
     """Simple helper to edit a static Pod manifest if configuration changes.
 
     Expects the template to use:
     - `config_digest` variable and store it in the `metadata.annotations`
       section, with the key `metalk8s.scality.com/config-digest`.
-    - `metalk8s_version` variabble and store it in the `metadata.labels`
+    - `metalk8s_version` variable and store it in the `metadata.labels`
       section, with the key `metalk8s.scality.com/version`.
 
     name:
@@ -41,44 +52,96 @@ def static_pod_managed(name,
 
     context:
         Context to use to render the source template.
-
-    kwargs:
-        Any arguments supported by `file.managed` are supported.
     """
+    ret = {"changes": {}, "comment": "", "name": name, "result": True}
+
+    if not name:
+        return _error(ret, "Manifest file name is required")
+
+    if not isinstance(source, six.text_type):
+        return _error(ret, "Source must be a single string")
+
     if not config_files:
         config_files = []
 
     for config_file in config_files_opt or []:
         if __salt__["file.file_exists"](config_file):
             config_files.append(config_file)
+        else:
+            log.debug(
+                "Ignoring optional config file %s: file does not exist",
+                config_file
+            )
+
+    config_file_digests = []
+    for config_file in config_files:
+        try:
+            digest = __salt__["hashutil.digest_file"](
+                config_file, checksum="sha256"
+            )
+        except CommandExecutionError as exc:
+            return _error(
+                ret,
+                "Unable to compute digest of config file {}: {}".format(
+                    config_file, exc
+                )
+            )
+        config_file_digests.append(digest)
 
-    config_file_digests = [
-        __salt__["hashutil.digest_file"](config_file, checksum="sha256")
-        for config_file in config_files
-    ]
     config_digest = __salt__["hashutil.md5_digest"](
         "-".join(config_file_digests)
     )
 
     match = re.search(r'metalk8s-(?P<version>.+)$', __env__)
     metalk8s_version = match.group('version') if match else "unknown"
 
-    return __states__["file.managed"](
-        name,
-        source,
-        template=kwargs.pop("template", "jinja"),
-        user=kwargs.pop("user", "root"),
-        group=kwargs.pop("group", "root"),
-        mode=kwargs.pop("mode", "0600"),
-        makedirs=kwargs.pop("makedirs", False),
-        backup=kwargs.pop("backup", False),
-        context=dict(
-            context or {},
-            config_digest=config_digest, metalk8s_version=metalk8s_version
-        ),
-        **kwargs
+    context_ = dict(
+        context or {},
+        config_digest=config_digest, metalk8s_version=metalk8s_version
     )
 
+    if __opts__["test"]:
+        log.warning("Test functionality is not yet implemented.")
+        ret["comment"] = (
+            "The manifest {} is in the correct state (supposedly)."
+        ).format(name)
+        return ret
+
+    # Gather the source file from the server
+    try:
+        source_filename, source_sum, comment_ = __salt__["file.get_managed"](
+            name,
+            template="jinja",
+            source=source,
+            source_hash="",
+            source_hash_name=None,
+            user="root",
+            group="root",
+            mode="0600",
+            attrs=None,
+            saltenv=__env__,
+            context=context_,
+            defaults=None,
+        )
+    except Exception as exc:  # pylint: disable=broad-except
+        log.debug(traceback.format_exc())
+        return _error(ret, "Unable to get managed file: {}".format(exc))
+
+    if comment_:
+        return _error(ret, comment_)
+    else:
+        try:
+            return __salt__["metalk8s.manage_static_pod_manifest"](
+                name,
+                source_filename,
+                source,
+                source_sum,
+                saltenv=__env__,
+            )
+        except Exception as exc:  # pylint: disable=broad-except
+            log.debug(traceback.format_exc())
+            return _error(ret, "Unable to manage file: {}".format(exc))
+
 
 def module_run(name, attemps=1, sleep_time=10, **kwargs):
     """Classic module.run with a retry logic as it's buggy in salt version