kubelib.py 文件源码

def apply(self, desc, filename, force=False):
        """Simple apply with secrets support."""
        changes = self.apply_secrets(desc, filename)
        action = "unknown"
        try:
            if force or len(changes):
                LOG.info(
                    'Secret changes detected: %s -- Replacing pod',
                    changes
                )
                action = "replace_path"
                self.replace_path(filename, force=force)
            else:
                action = "apply_file"
                self.apply_file(filename)

        except sh.ErrorReturnCode_1:
            LOG.error('%s failed (forcing)', action)

            if self.exists(desc.metadata.name):
                self.delete_path(filename)
            self.create_path(filename)

        return changes