Function heartbeat

examples/task-samples/task-byoo-sample/main.py:198–238

The task container must periodically update 'lastUpdatedAt' field in the progress file as a heartbeat that worker container monitors. The task will fail if there's no change in 'lastUpdatedAt' over 5 min. Parameters: :param progress_file_path: progress file path

(progress_file_path: str)

def heartbeat(progress_file_path: str):
    """
    The task container must periodically update 'lastUpdatedAt' field
    in the progress file as a heartbeat that worker container monitors.
    The task will fail if there's no change in 'lastUpdatedAt' over 5 min.

    Parameters:
    :param progress_file_path: progress file path
    """
    while TASK_RUNNING:
        ts = datetime.datetime.now(datetime.timezone.utc).strftime(
            "%Y-%m-%dT%H:%M:%S.%fZ"
        )
        if not os.path.isfile(progress_file_path):
            logger.info("Creating progress file")
            progress = Progress(
                taskId=os.getenv("NVCT_TASK_ID", ""),
                percentComplete=0,
                name="",
                lastUpdatedAt=ts,
                metadata=dict(),
            )
            with open(progress_file_path, "w") as f:
                f.write(progress.output_to_json())

        else:
            progress = parse_progress_file(progress_file_path)
            progress.lastUpdatedAt = ts

            temp_file = progress_file_path + ".tmp"
            with open(temp_file, "w") as f:
                f.write(progress.output_to_json())

            try:
                os.rename(temp_file, progress_file_path)
            except Exception as e:
                logger.exception(f"Failed to write to progress file: {e}")
                raise e

        logger.info(f"Updated timestamp in progress file {ts}")
        time.sleep(60 + random.randint(1, 10))


def wait_for_secrets_file(file_path, max_retries=30, retry_delay=5.0):