fix: set RunStepTaskAction end time to asset sync completion time (#11)

Signed-off-by: Gahyun Suh <[email protected]>

src/deadline_worker_agent/sessions/actions/sync_input_job_attachments.py

@@ -127,8 +127,6 @@ def _on_done(
             )
         except Exception as e:
             session.logger.exception(e)
-            # We need to directly complete the action. Other actions rely on the Open Job Description session's
-            # callback to complete the action
             action_status = ActionStatus(
                 state=ActionState.FAILED,
                 fail_message=str(e),

src/deadline_worker_agent/sessions/log_config.py

@@ -198,7 +198,7 @@ def create_remote_handler(
         if not (log_group := self.options.get(LOG_CONFIG_OPTION_GROUP_NAME_KEY, None)):
             raise KeyError(f'No "{LOG_CONFIG_OPTION_GROUP_NAME_KEY}" in logConfiguration.options')
         elif not (log_stream := self.options.get(LOG_CONFIG_OPTION_STREAM_NAME_KEY, None)):
-            raise KeyError('No "{LOG_CONFIG_OPTION_STREAM_NAME_KEY}" in logConfiguration.options')
+            raise KeyError(f'No "{LOG_CONFIG_OPTION_STREAM_NAME_KEY}" in logConfiguration.options')
 
         return CloudWatchHandler(
             log_group_name=log_group,

src/deadline_worker_agent/sessions/session.py

@@ -962,14 +962,17 @@ def _action_updated_impl(
         ):
             # Synchronizing job output attachments is currently bundled together with the
             # RunStepTaskAction. The synchronization happens after the task run succeeds, and both
-            # must be successful in order to mark the action as SUCCEEDED.
+            # must be successful in order to mark the action as SUCCEEDED. The time when
+            # the action is completed should be the moment when the sunchronization have
+            # been finished.
             try:
                 self._sync_asset_outputs(current_action=current_action)
+                now = datetime.now(tz=timezone.utc)
             except Exception as e:
                 # Log and fail the task run action if we are unable to sync output job
                 # attachments
                 fail_message = f"Failed to sync job output attachments for {current_action.definition.human_readable()}: {e}"
-                logger.warning(fail_message)
+                self.logger.warning(fail_message)
                 action_status = ActionStatus(state=ActionState.FAILED, fail_message=fail_message)
                 is_unsuccessful = True
 
@@ -1083,7 +1086,7 @@ def _sync_asset_outputs(
         from .actions import RunStepTaskAction
 
         assert isinstance(current_action.definition, RunStepTaskAction)
-        self._asset_sync.sync_outputs(
+        upload_summary_statistics = self._asset_sync.sync_outputs(
             s3_settings=s3_settings,
             attachments=attachments,
             queue_id=self._queue_id,
@@ -1096,6 +1099,9 @@ def _sync_asset_outputs(
             storage_profiles_path_mapping_rules=storage_profiles_path_mapping_rules_dict,
             on_uploading_files=partial(self._notifier_callback, current_action),
         )
+
+        ASSET_SYNC_LOGGER.info(f"Summary Statistics for file uploads:\n{upload_summary_statistics}")
+
         ASSET_SYNC_LOGGER.info("Finished syncing outputs using Job Attachments")
 
     def run_task(

test/unit/sessions/test_session.py

@@ -29,6 +29,7 @@
 
 from deadline_worker_agent.api_models import EnvironmentAction, TaskRunAction
 from deadline_worker_agent.sessions import Session
+from deadline_worker_agent.sessions import session as session_module
 from deadline_worker_agent.sessions.session import (
     CurrentAction,
     SessionActionStatus,
@@ -1087,7 +1088,14 @@ def test_success_task_run(
             end_time=action_complete_time,
         )
 
-        with patch.object(session, "_sync_asset_outputs") as mock_sync_asset_outputs:
+        def mock_now(*arg, **kwarg) -> datetime:
+            return action_complete_time
+
+        with patch.object(session_module, "datetime") as mock_datetime, patch.object(
+            session, "_sync_asset_outputs"
+        ) as mock_sync_asset_outputs:
+            mock_datetime.now.side_effect = mock_now
+
             # Assert that reporting the action update happens AFTER syncing the output job
             # attachments.
             def sync_asset_outputs_side_effect(*, current_action: CurrentAction) -> None: