Merge pull request #1573 from pypeclub/bugfix/170-farm-publishing-check-if-published-items-do-exist

Farm publishing: check if published items do exist

openpype/cli.py

@@ -115,7 +115,9 @@ def extractenvironments(output_json_path, project, asset, task, app):
 @main.command()
 @click.argument("paths", nargs=-1)
 @click.option("-d", "--debug", is_flag=True, help="Print debug messages")
-def publish(debug, paths):
+@click.option("-t", "--targets", help="Targets module", default=None,
+              multiple=True)
+def publish(debug, paths, targets):
     """Start CLI publishing.
 
     Publish collects json from paths provided as an argument.
@@ -123,7 +125,7 @@ def publish(debug, paths):
     """
     if debug:
         os.environ['OPENPYPE_DEBUG'] = '3'
-    PypeCommands.publish(list(paths))
+    PypeCommands.publish(list(paths), targets)
 
 
 @main.command()

openpype/lib/abstract_submit_deadline.py

@@ -18,6 +18,48 @@ import pyblish.api
 from .abstract_metaplugins import AbstractMetaInstancePlugin
 
 
+def requests_post(*args, **kwargs):
+    """Wrap request post method.
+
+    Disabling SSL certificate validation if ``DONT_VERIFY_SSL`` environment
+    variable is found. This is useful when Deadline or Muster server are
+    running with self-signed certificates and their certificate is not
+    added to trusted certificates on client machines.
+
+    Warning:
+        Disabling SSL certificate validation is defeating one line
+        of defense SSL is providing and it is not recommended.
+
+    """
+    if 'verify' not in kwargs:
+        kwargs['verify'] = False if os.getenv("OPENPYPE_DONT_VERIFY_SSL",
+                                              True) else True  # noqa
+    # add 10sec timeout before bailing out
+    kwargs['timeout'] = 10
+    return requests.post(*args, **kwargs)
+
+
+def requests_get(*args, **kwargs):
+    """Wrap request get method.
+
+    Disabling SSL certificate validation if ``DONT_VERIFY_SSL`` environment
+    variable is found. This is useful when Deadline or Muster server are
+    running with self-signed certificates and their certificate is not
+    added to trusted certificates on client machines.
+
+    Warning:
+        Disabling SSL certificate validation is defeating one line
+        of defense SSL is providing and it is not recommended.
+
+    """
+    if 'verify' not in kwargs:
+        kwargs['verify'] = False if os.getenv("OPENPYPE_DONT_VERIFY_SSL",
+                                              True) else True  # noqa
+    # add 10sec timeout before bailing out
+    kwargs['timeout'] = 10
+    return requests.get(*args, **kwargs)
+
+
 @attr.s
 class DeadlineJobInfo(object):
     """Mapping of all Deadline *JobInfo* attributes.
@@ -579,7 +621,7 @@ class AbstractSubmitDeadline(pyblish.api.InstancePlugin):
 
         """
         url = "{}/api/jobs".format(self._deadline_url)
-        response = self._requests_post(url, json=payload)
+        response = requests_post(url, json=payload)
         if not response.ok:
             self.log.error("Submission failed!")
             self.log.error(response.status_code)
@@ -592,41 +634,3 @@ class AbstractSubmitDeadline(pyblish.api.InstancePlugin):
         self._instance.data["deadlineSubmissionJob"] = result
 
         return result["_id"]
-
-    def _requests_post(self, *args, **kwargs):
-        """Wrap request post method.
-
-        Disabling SSL certificate validation if ``DONT_VERIFY_SSL`` environment
-        variable is found. This is useful when Deadline or Muster server are
-        running with self-signed certificates and their certificate is not
-        added to trusted certificates on client machines.
-
-        Warning:
-            Disabling SSL certificate validation is defeating one line
-            of defense SSL is providing and it is not recommended.
-
-        """
-        if 'verify' not in kwargs:
-            kwargs['verify'] = False if os.getenv("OPENPYPE_DONT_VERIFY_SSL", True) else True  # noqa
-        # add 10sec timeout before bailing out
-        kwargs['timeout'] = 10
-        return requests.post(*args, **kwargs)
-
-    def _requests_get(self, *args, **kwargs):
-        """Wrap request get method.
-
-        Disabling SSL certificate validation if ``DONT_VERIFY_SSL`` environment
-        variable is found. This is useful when Deadline or Muster server are
-        running with self-signed certificates and their certificate is not
-        added to trusted certificates on client machines.
-
-        Warning:
-            Disabling SSL certificate validation is defeating one line
-            of defense SSL is providing and it is not recommended.
-
-        """
-        if 'verify' not in kwargs:
-            kwargs['verify'] = False if os.getenv("OPENPYPE_DONT_VERIFY_SSL", True) else True  # noqa
-        # add 10sec timeout before bailing out
-        kwargs['timeout'] = 10
-        return requests.get(*args, **kwargs)

openpype/modules/deadline/plugins/publish/submit_publish_job.py

@@ -231,7 +231,8 @@ class ProcessSubmittedJobOnFarm(pyblish.api.InstancePlugin):
 
         args = [
             'publish',
-            roothless_metadata_path
+            roothless_metadata_path,
+            "--targets {}".format("deadline")
         ]
 
         # Generate the payload for Deadline submission

openpype/modules/deadline/plugins/publish/validate_expected_and_rendered_files.py

@@ -0,0 +1,186 @@
+import os
+import json
+import pyblish.api
+
+from avalon.vendor import requests
+
+from openpype.api import get_system_settings
+from openpype.lib.abstract_submit_deadline import requests_get
+from openpype.lib.delivery import collect_frames
+
+
+class ValidateExpectedFiles(pyblish.api.InstancePlugin):
+    """Compare rendered and expected files"""
+
+    label = "Validate rendered files from Deadline"
+    order = pyblish.api.ValidatorOrder
+    families = ["render"]
+    targets = ["deadline"]
+
+    # check if actual frame range on render job wasn't different
+    # case when artists wants to render only subset of frames
+    allow_user_override = True
+
+    def process(self, instance):
+        frame_list = self._get_frame_list(instance.data["render_job_id"])
+
+        for repre in instance.data["representations"]:
+            expected_files = self._get_expected_files(repre)
+
+            staging_dir = repre["stagingDir"]
+            existing_files = self._get_existing_files(staging_dir)
+
+            expected_non_existent = expected_files.difference(
+                existing_files)
+            if len(expected_non_existent) != 0:
+                self.log.info("Some expected files missing {}".format(
+                    expected_non_existent))
+
+                if self.allow_user_override:
+                    file_name_template, frame_placeholder = \
+                        self._get_file_name_template_and_placeholder(
+                            expected_files)
+
+                    if not file_name_template:
+                        return
+
+                    real_expected_rendered = self._get_real_render_expected(
+                        file_name_template,
+                        frame_placeholder,
+                        frame_list)
+
+                    real_expected_non_existent = \
+                        real_expected_rendered.difference(existing_files)
+                    if len(real_expected_non_existent) != 0:
+                        raise RuntimeError("Still missing some files {}".
+                                           format(real_expected_non_existent))
+                    self.log.info("Update range from actual job range")
+                    repre["files"] = sorted(list(real_expected_rendered))
+                else:
+                    raise RuntimeError("Some expected files missing {}".format(
+                        expected_non_existent))
+
+    def _get_frame_list(self, original_job_id):
+        """
+            Returns list of frame ranges from all render job.
+
+            Render job might be requeried so job_id in metadata.json is invalid
+            GlobalJobPreload injects current ids to RENDER_JOB_IDS.
+
+            Args:
+                original_job_id (str)
+            Returns:
+                (list)
+        """
+        all_frame_lists = []
+        render_job_ids = os.environ.get("RENDER_JOB_IDS")
+        if render_job_ids:
+            render_job_ids = render_job_ids.split(',')
+        else:  # fallback
+            render_job_ids = [original_job_id]
+
+        for job_id in render_job_ids:
+            job_info = self._get_job_info(job_id)
+            frame_list = job_info["Props"]["Frames"]
+            if frame_list:
+                all_frame_lists.extend(frame_list.split(','))
+
+        return all_frame_lists
+
+    def _get_real_render_expected(self, file_name_template, frame_placeholder,
+                                  frame_list):
+        """
+            Calculates list of names of expected rendered files.
+
+            Might be different from job expected files if user explicitly and
+            manually change frame list on Deadline job.
+        """
+        real_expected_rendered = set()
+        src_padding_exp = "%0{}d".format(len(frame_placeholder))
+        for frames in frame_list:
+            if '-' not in frames:  # single frame
+                frames = "{}-{}".format(frames, frames)
+
+            start, end = frames.split('-')
+            for frame in range(int(start), int(end) + 1):
+                ren_name = file_name_template.replace(
+                    frame_placeholder, src_padding_exp % frame)
+                real_expected_rendered.add(ren_name)
+
+        return real_expected_rendered
+
+    def _get_file_name_template_and_placeholder(self, files):
+        """Returns file name with frame replaced with # and this placeholder"""
+        sources_and_frames = collect_frames(files)
+
+        file_name_template = frame_placeholder = None
+        for file_name, frame in sources_and_frames.items():
+            frame_placeholder = "#" * len(frame)
+            file_name_template = os.path.basename(
+                file_name.replace(frame, frame_placeholder))
+            break
+
+        return file_name_template, frame_placeholder
+
+    def _get_job_info(self, job_id):
+        """
+            Calls DL for actual job info for 'job_id'
+
+            Might be different than job info saved in metadata.json if user
+            manually changes job pre/during rendering.
+        """
+        deadline_url = (
+            get_system_settings()
+            ["modules"]
+            ["deadline"]
+            ["DEADLINE_REST_URL"]
+        )
+        assert deadline_url, "Requires DEADLINE_REST_URL"
+
+        url = "{}/api/jobs?JobID={}".format(deadline_url, job_id)
+        try:
+            response = requests_get(url)
+        except requests.exceptions.ConnectionError:
+            print("Deadline is not accessible at {}".format(deadline_url))
+            # self.log("Deadline is not accessible at {}".format(deadline_url))
+            return {}
+
+        if not response.ok:
+            self.log.error("Submission failed!")
+            self.log.error(response.status_code)
+            self.log.error(response.content)
+            raise RuntimeError(response.text)
+
+        json_content = response.json()
+        if json_content:
+            return json_content.pop()
+        return {}
+
+    def _parse_metadata_json(self, json_path):
+        if not os.path.exists(json_path):
+            msg = "Metadata file {} doesn't exist".format(json_path)
+            raise RuntimeError(msg)
+
+        with open(json_path) as fp:
+            try:
+                return json.load(fp)
+            except Exception as exc:
+                self.log.error(
+                    "Error loading json: "
+                    "{} - Exception: {}".format(json_path, exc)
+                )
+
+    def _get_existing_files(self, out_dir):
+        """Returns set of existing file names from 'out_dir'"""
+        existing_files = set()
+        for file_name in os.listdir(out_dir):
+            existing_files.add(file_name)
+        return existing_files
+
+    def _get_expected_files(self, repre):
+        """Returns set of file names from metadata.json"""
+        expected_files = set()
+
+        for file_name in repre["files"]:
+            expected_files.add(file_name)
+        return expected_files

openpype/plugins/publish/collect_rendered_files.py

@@ -87,11 +87,14 @@ class CollectRenderedFiles(pyblish.api.ContextPlugin):
             instance = self._context.create_instance(
                 instance_data.get("subset")
             )
-            self.log.info("Filling stagignDir...")
+            self.log.info("Filling stagingDir...")
 
             self._fill_staging_dir(instance_data, anatomy)
             instance.data.update(instance_data)
 
+            # stash render job id for later validation
+            instance.data["render_job_id"] = data.get("job").get("_id")
+
             representations = []
             for repre_data in instance_data.get("representations") or []:
                 self._fill_staging_dir(repre_data, anatomy)

openpype/pype_commands.py

@@ -46,16 +46,18 @@ class PypeCommands:
         standalonepublish.main()
 
     @staticmethod
-    def publish(paths):
+    def publish(paths, targets=None):
         """Start headless publishing.
 
         Publish use json from passed paths argument.
 
         Args:
             paths (list): Paths to jsons.
+            targets (string): What module should be targeted
+                (to choose validator for example)
 
         Raises:
-            RuntimeError: When there is no pathto process.
+            RuntimeError: When there is no path to process.
         """
         if not any(paths):
             raise RuntimeError("No publish paths specified")
@@ -82,6 +84,10 @@ class PypeCommands:
         pyblish.api.register_target("filesequence")
         pyblish.api.register_host("shell")
 
+        if targets:
+            for target in targets:
+                pyblish.api.register_target(target)
+
         os.environ["OPENPYPE_PUBLISH_DATA"] = os.pathsep.join(paths)
 
         log.info("Running publish ...")

openpype/settings/defaults/project_settings/deadline.json

@@ -1,5 +1,12 @@
 {
     "publish": {
+        "ValidateExpectedFiles": {
+            "enabled": true,
+            "active": true,
+            "families": ["render"],
+            "targets": ["deadline"],
+            "allow_user_override": true
+        },
         "MayaSubmitDeadline": {
             "enabled": true,
             "optional": false,

openpype/settings/entities/schemas/projects_schema/schema_project_deadline.json

@@ -11,6 +11,47 @@
             "key": "publish",
             "label": "Publish plugins",
             "children": [
+                {
+                    "type": "dict",
+                    "collapsible": true,
+                    "key": "ValidateExpectedFiles",
+                    "label": "Validate Expected Files",
+                    "checkbox_key": "enabled",
+                    "children": [
+                        {
+                            "type": "boolean",
+                            "key": "enabled",
+                            "label": "Enabled"
+                        },
+                        {
+                            "type": "boolean",
+                            "key": "active",
+                            "label": "Active"
+                        },
+                        {
+                            "type": "label",
+                            "label": "Validate if all expected files were rendered"
+                        },
+                        {
+                            "type": "boolean",
+                            "key": "allow_user_override",
+                            "object_type": "text",
+                            "label": "Allow user change frame range"
+                        },
+                        {
+                            "type": "list",
+                            "key": "families",
+                            "object_type": "text",
+                            "label": "Trigger on families"
+                        },
+                        {
+                            "type": "list",
+                            "key": "targets",
+                            "object_type": "text",
+                            "label": "Trigger for plugins"
+                        }
+                    ]
+                },
                 {
                     "type": "dict",
                     "collapsible": true,

vendor/deadline/custom/plugins/GlobalJobPreLoad.py

@@ -9,6 +9,10 @@ from Deadline.Scripting import RepositoryUtils, FileUtils
 
 
 def inject_openpype_environment(deadlinePlugin):
+    """ Pull env vars from OpenPype and push them to rendering process.
+
+        Used for correct paths, configuration from OpenPype etc.
+    """
     job = deadlinePlugin.GetJob()
     job = RepositoryUtils.GetJob(job.JobId, True)  # invalidates cache
 
@@ -73,6 +77,21 @@ def inject_openpype_environment(deadlinePlugin):
         raise
 
 
+def inject_render_job_id(deadlinePlugin):
+    """Inject dependency ids to publish process as env var for validation."""
+    print("inject_render_job_id start")
+    job = deadlinePlugin.GetJob()
+    job = RepositoryUtils.GetJob(job.JobId, True)  # invalidates cache
+
+    dependency_ids = job.JobDependencyIDs
+    print("dependency_ids {}".format(dependency_ids))
+    render_job_ids = ",".join(dependency_ids)
+
+    deadlinePlugin.SetProcessEnvironmentVariable("RENDER_JOB_IDS",
+                                                 render_job_ids)
+    print("inject_render_job_id end")
+
+
 def pype_command_line(executable, arguments, workingDirectory):
     """Remap paths in comand line argument string.
 
@@ -156,8 +175,7 @@ def __main__(deadlinePlugin):
                            "render and publish.")
 
     if openpype_publish_job == '1':
-        print("Publish job, skipping inject.")
-        return
+        inject_render_job_id(deadlinePlugin)
     elif openpype_render_job == '1':
         inject_openpype_environment(deadlinePlugin)
     else: