Worker

CUDOS

Inspired by Majordomo Protocol Worker API, ZeroMQ, Python version.

Original MDP/Worker spec

Location: http://rfc.zeromq.org/spec:7.

Author: Min RK benjaminrk@gmail.com

Based on Java example by Arkadiusz Orzechowski

WorkerWatchDog(worker)

Bases: Thread

Class to monitor worker performance

Source code in norfab\core\worker.py

def __init__(self, worker):
    super().__init__()
    self.worker = worker
    self.worker_process = psutil.Process(os.getpid())

    # extract inventory attributes
    self.watchdog_interval = worker.inventory.get("watchdog_interval", 30)
    self.memory_threshold_mbyte = worker.inventory.get(
        "memory_threshold_mbyte", 1000
    )
    self.memory_threshold_action = worker.inventory.get(
        "memory_threshold_action", "log"
    )

    # initiate variables
    self.runs = 0
    self.watchdog_tasks = []

get_ram_usage()

Return RAM usage in Mbyte

Source code in norfab\core\worker.py

def get_ram_usage(self):
    """Return RAM usage in Mbyte"""
    return self.worker_process.memory_info().rss / 1024000

Result(result=None, failed=False, errors=None, task=None, messages=None, juuid=None)

Result of running individual tasks.

Attributes/Arguments:

Parameters:

Name	Type	Description	Default
changed		True if the task is changing the system	required
result	Any	Result of the task execution, see task's documentation for details	None
failed	bool	Whether the execution failed or not	False
(logging.LEVEL)	severity_level	Severity level associated to the result of the execution	required
errors	Optional[List[str]]	exception thrown during the execution of the task (if any)	None
task	str	Task function name that produced the results	None
messages	Optional[List[str]]	List of messages produced by the task	None
juuid	Optional[str]	Job UUID associated with the task	None

Source code in norfab\core\worker.py

def __init__(
    self,
    result: Any = None,
    failed: bool = False,
    errors: Optional[List[str]] = None,
    task: str = None,
    messages: Optional[List[str]] = None,
    juuid: Optional[str] = None,
) -> None:
    self.task = task
    self.result = result
    self.failed = failed
    self.errors = errors or []
    self.messages = messages or []
    self.juuid = juuid

raise_for_status(message='')

Method to raise error if job failed

Source code in norfab\core\worker.py

def raise_for_status(self, message=""):
    """Method to raise error if job failed"""
    if self.failed:
        if message:
            raise NorfabJobFailedError(
                f"{message}; Errors: {'; '.join(self.errors)}"
            )
        else:
            raise NorfabJobFailedError(f"Errors: {'; '.join(self.errors)}")

dictionary()

Method to serialize result to a dictionary

Source code in norfab\core\worker.py

def dictionary(self):
    """Method to serialize result to a dictionary"""
    if not isinstance(self.errors, list):
        self.errors = [self.errors]
    if not isinstance(self.messages, list):
        self.messages = [self.messages]

    return {
        "task": self.task,
        "failed": self.failed,
        "errors": self.errors,
        "result": self.result,
        "messages": self.messages,
        "juuid": self.juuid,
    }

NFPWorker(inventory, broker, service, name, exit_event, log_level=None, log_queue=None, multiplier=6, keepalive=2500)

Parameters:

Name	Type	Description	Default
broker	str	str, broker endpoint e.g. tcp://127.0.0.1:5555	required
service	str	str, service name	required
name	str	str, worker name	required
exist_event		obj, threading event, if set signal worker to stop	required
multiplier	int	int, number of keepalives lost before consider other party dead	6
keepalive	int	int, keepalive interval in milliseconds	2500

Source code in norfab\core\worker.py

def __init__(
    self,
    inventory: NorFabInventory,
    broker: str,
    service: str,
    name: str,
    exit_event,
    log_level: str = None,
    log_queue: object = None,
    multiplier: int = 6,
    keepalive: int = 2500,
):
    self.setup_logging(log_queue, log_level)
    self.inventory = inventory
    self.broker = broker
    self.service = service.encode("utf-8") if isinstance(service, str) else service
    self.name = name
    self.exit_event = exit_event
    self.broker_socket = None
    self.multiplier = multiplier
    self.keepalive = keepalive
    self.socket_lock = (
        threading.Lock()
    )  # used for keepalives to protect socket object

    # create base directories
    self.base_dir = os.path.join(
        self.inventory.base_dir, "__norfab__", "files", "worker", self.name
    )
    self.base_dir_jobs = os.path.join(self.base_dir, "jobs")
    os.makedirs(self.base_dir, exist_ok=True)
    os.makedirs(self.base_dir_jobs, exist_ok=True)

    # create events and queues
    self.destroy_event = threading.Event()
    self.request_thread = None
    self.reply_thread = None
    self.close_thread = None
    self.recv_thread = None
    self.event_thread = None

    self.post_queue = queue.Queue(maxsize=0)
    self.get_queue = queue.Queue(maxsize=0)
    self.delete_queue = queue.Queue(maxsize=0)
    self.event_queue = queue.Queue(maxsize=0)

    # generate certificates and create directories
    generate_certificates(
        self.base_dir,
        cert_name=self.name,
        broker_keys_dir=os.path.join(
            self.inventory.base_dir, "__norfab__", "files", "broker", "public_keys"
        ),
        inventory=self.inventory,
    )
    self.public_keys_dir = os.path.join(self.base_dir, "public_keys")
    self.secret_keys_dir = os.path.join(self.base_dir, "private_keys")

    self.ctx = zmq.Context()
    self.poller = zmq.Poller()
    self.reconnect_to_broker()

    # create queue file
    self.queue_filename = os.path.join(self.base_dir_jobs, f"{self.name}.queue.txt")
    if not os.path.exists(self.queue_filename):
        with open(self.queue_filename, "w") as f:
            pass
    self.queue_done_filename = os.path.join(
        self.base_dir_jobs, f"{self.name}.queue.done.txt"
    )
    if not os.path.exists(self.queue_done_filename):
        with open(self.queue_done_filename, "w") as f:
            pass

    self.client = NFPClient(
        self.inventory,
        self.broker,
        name=f"{self.name}-NFPClient",
        exit_event=self.exit_event,
    )

setup_logging(log_queue, log_level)

Method to apply logging configuration

Source code in norfab\core\worker.py

def setup_logging(self, log_queue, log_level: str) -> None:
    """Method to apply logging configuration"""
    logging_config_producer["handlers"]["queue"]["queue"] = log_queue
    if log_level is not None:
        logging_config_producer["root"]["level"] = log_level
    logging.config.dictConfig(logging_config_producer)

reconnect_to_broker()

Connect or reconnect to broker

Source code in norfab\core\worker.py

def reconnect_to_broker(self):
    """Connect or reconnect to broker"""
    if self.broker_socket:
        self.send_to_broker(NFP.DISCONNECT)
        self.poller.unregister(self.broker_socket)
        self.broker_socket.close()

    self.broker_socket = self.ctx.socket(zmq.DEALER)
    self.broker_socket.setsockopt_unicode(zmq.IDENTITY, self.name, "utf8")
    self.broker_socket.linger = 0

    # We need two certificates, one for the client and one for
    # the server. The client must know the server's public key
    # to make a CURVE connection.
    client_secret_file = os.path.join(
        self.secret_keys_dir, f"{self.name}.key_secret"
    )
    client_public, client_secret = zmq.auth.load_certificate(client_secret_file)
    self.broker_socket.curve_secretkey = client_secret
    self.broker_socket.curve_publickey = client_public

    # The client must know the server's public key to make a CURVE connection.
    server_public_file = os.path.join(self.public_keys_dir, "broker.key")
    server_public, _ = zmq.auth.load_certificate(server_public_file)
    self.broker_socket.curve_serverkey = server_public

    self.broker_socket.connect(self.broker)
    self.poller.register(self.broker_socket, zmq.POLLIN)

    # Register service with broker
    self.send_to_broker(NFP.READY)

    # start keepalives
    if self.keepaliver is not None:
        self.keepaliver.restart(self.broker_socket)
    else:
        self.keepaliver = KeepAliver(
            address=None,
            socket=self.broker_socket,
            multiplier=self.multiplier,
            keepalive=self.keepalive,
            exit_event=self.destroy_event,
            service=self.service,
            whoami=NFP.WORKER,
            name=self.name,
            socket_lock=self.socket_lock,
        )
        self.keepaliver.start()

    self.stats_reconnect_to_broker += 1
    log.info(
        f"{self.name} - registered to broker at '{self.broker}', "
        f"service '{self.service.decode('utf-8')}'"
    )

send_to_broker(command, msg=None)

Send message to broker.

If no msg is provided, creates one internally

Source code in norfab\core\worker.py

def send_to_broker(self, command, msg: list = None):
    """Send message to broker.

    If no msg is provided, creates one internally
    """
    if command == NFP.READY:
        msg = [b"", NFP.WORKER, NFP.READY, self.service]
    elif command == NFP.DISCONNECT:
        msg = [b"", NFP.WORKER, NFP.DISCONNECT, self.service]
    elif command == NFP.RESPONSE:
        msg = [b"", NFP.WORKER, NFP.RESPONSE] + msg
    elif command == NFP.EVENT:
        msg = [b"", NFP.WORKER, NFP.EVENT] + msg
    else:
        log.error(
            f"{self.name} - cannot send '{command}' to broker, command unsupported"
        )
        return

    log.debug(f"{self.name} - sending '{msg}'")

    with self.socket_lock:
        self.broker_socket.send_multipart(msg)

load_inventory()

Function to load inventory from broker for this worker name.

Source code in norfab\core\worker.py

def load_inventory(self):
    """
    Function to load inventory from broker for this worker name.
    """
    inventory_data = self.client.get(
        "sid.service.broker", "get_inventory", kwargs={"name": self.name}
    )

    log.debug(f"{self.name} - worker received invenotry data {inventory_data}")

    if inventory_data["results"]:
        return inventory_data["results"]
    else:
        return {}

worker_exit()

Method to override in child classes with a set of actions to perform on exit call

Source code in norfab\core\worker.py

def worker_exit(self) -> None:
    """Method to override in child classes with a set of actions to perform on exit call"""
    return None

get_inventory()

Method to override in child classes to retrieve worker inventory

Source code in norfab\core\worker.py

def get_inventory(self) -> Dict:
    """Method to override in child classes to retrieve worker inventory"""
    raise NotImplementedError

get_version()

Method to override in child classes to retrieve worker version report

Source code in norfab\core\worker.py

def get_version(self) -> Dict:
    """Method to override in child classes to retrieve worker version report"""
    raise NotImplementedError

fetch_file(url, raise_on_fail=False, read=True)

Function to download file from broker File Sharing Service

Parameters:

Name	Type	Description	Default
url	str	file location string in nf://<filepath> format	required
raise_on_fail	bool	raise FIleNotFoundError if download fails	False
read	bool	if True returns file content, return OS path to saved file otherwise	True

Source code in norfab\core\worker.py

def fetch_file(
    self, url: str, raise_on_fail: bool = False, read: bool = True
) -> str:
    """
    Function to download file from broker File Sharing Service

    :param url: file location string in ``nf://<filepath>`` format
    :param raise_on_fail: raise FIleNotFoundError if download fails
    :param read: if True returns file content, return OS path to saved file otherwise
    """
    status, file_content = self.client.fetch_file(url=url, read=read)
    msg = f"{self.name} - worker '{url}' fetch file failed with status '{status}'"

    if status == "200":
        return file_content
    elif raise_on_fail is True:
        raise FileNotFoundError(msg)
    else:
        log.error(msg)
        return None

fetch_jinja2(url)

Helper function to recursively download Jinja2 template together with other templates referenced using "include" statements

Parameters:

Name	Type	Description	Default
url	str	nf://file/path like URL to download file	required

Source code in norfab\core\worker.py

def fetch_jinja2(self, url: str) -> str:
    """
    Helper function to recursively download Jinja2 template together with
    other templates referenced using "include" statements

    :param url: ``nf://file/path`` like URL to download file
    """
    filepath = self.fetch_file(url, read=False)
    if filepath is None:
        msg = f"{self.name} - file download failed '{url}'"
        raise FileNotFoundError(msg)

    # download Jinja2 template "include"-ed files
    content = self.fetch_file(url, read=True)
    j2env = Environment(loader="BaseLoader")
    try:
        parsed_content = j2env.parse(content)
    except Exception as e:
        msg = f"{self.name} - Jinja2 template parsing failed '{url}', error: '{e}'"
        raise Exception(msg)

    # run recursion on include statements
    for node in parsed_content.find_all(Include):
        include_file = node.template.value
        base_path = os.path.split(url)[0]
        self.fetch_jinja2(os.path.join(base_path, include_file))

    return filepath

job_details(uuid, data=True, result=True, events=True)

Method to get job details by UUID for completed jobs.

Parameters:

Name	Type	Description	Default
uuid	str	str, job UUID to return details for	required
data	bool	bool, if True return job data	True
result	bool	bool, if True return job result	True
events	bool	bool, if True return job events	True

Returns:

Type	Description
Result	Result object with job details

Source code in norfab\core\worker.py

def job_details(
    self, uuid: str, data: bool = True, result: bool = True, events: bool = True
) -> Result:
    """
    Method to get job details by UUID for completed jobs.

    :param uuid: str, job UUID to return details for
    :param data: bool, if True return job data
    :param result: bool, if True return job result
    :param events: bool, if True return job events
    :return: Result object with job details
    """
    job = None
    with queue_file_lock:
        with open(self.queue_done_filename, "rb+") as f:
            for entry in f.readlines():
                job_data, job_result, job_events = None, None, []
                job_entry = entry.decode("utf-8").strip()
                suuid, start, end = job_entry.split("--")  # {suuid}--start--end
                if suuid != uuid:
                    continue
                # load job request details
                client_address, empty, juuid, job_data_bytes = loader(
                    request_filename(suuid, self.base_dir_jobs)
                )
                if data:
                    job_data = json.loads(job_data_bytes.decode("utf-8"))
                # load job result details
                if result:
                    rep_filename = reply_filename(suuid, self.base_dir_jobs)
                    if os.path.exists(rep_filename):
                        job_result = loader(rep_filename)
                        job_result = json.loads(job_result[-1].decode("utf-8"))
                        job_result = job_result[self.name]
                # load event details
                if events:
                    events_filename = event_filename(suuid, self.base_dir_jobs)
                    if os.path.exists(events_filename):
                        job_events = loader(events_filename)
                        job_events = [e[-1] for e in job_events]

                job = {
                    "uuid": suuid,
                    "client": client_address.decode("utf-8"),
                    "received_timestamp": start,
                    "done_timestamp": end,
                    "status": "COMPLETED",
                    "job_data": job_data,
                    "job_result": job_result,
                    "job_events": job_events,
                }

    if job:
        return Result(
            task=f"{self.name}:job_details",
            result=job,
        )
    else:
        raise FileNotFoundError(f"{self.name} - job with UUID '{uuid}' not found")

job_list(pending=True, completed=True, task=None, last=None, client=None, uuid=None)

Method to list worker jobs completed and pending.

Parameters:

Name	Type	Description	Default
pending	bool	bool, if True or None return pending jobs, if False skip pending jobs	True
completed	bool	bool, if True or None return completed jobs, if False skip completed jobs	True
task	str	str, if provided return only jobs with this task name	None
last	int	int, if provided return only last N completed and last N pending jobs	None
client	str	str, if provided return only jobs submitted by this client	None
uuid	str	str, if provided return only job with this UUID	None

Returns:

Type	Description
Result	Result object with list of jobs

Source code in norfab\core\worker.py

def job_list(
    self,
    pending: bool = True,
    completed: bool = True,
    task: str = None,
    last: int = None,
    client: str = None,
    uuid: str = None,
) -> Result:
    """
    Method to list worker jobs completed and pending.

    :param pending: bool, if True or None return pending jobs, if
        False skip pending jobs
    :param completed: bool, if True or None return completed jobs,
        if False skip completed jobs
    :param task: str, if provided return only jobs with this task name
    :param last: int, if provided return only last N completed and
        last N pending jobs
    :param client: str, if provided return only jobs submitted by this client
    :param uuid: str, if provided return only job with this UUID
    :return: Result object with list of jobs
    """
    job_pending = []
    # load pending jobs
    if pending is True:
        with queue_file_lock:
            with open(self.queue_filename, "rb+") as f:
                for entry in f.readlines():
                    job_entry = entry.decode("utf-8").strip()
                    suuid, start = job_entry.split("--")  # {suuid}--start
                    if uuid and suuid != uuid:
                        continue
                    client_address, empty, juuid, data = loader(
                        request_filename(suuid, self.base_dir_jobs)
                    )
                    if client and client_address.decode("utf-8") != client:
                        continue
                    job_task = json.loads(data.decode("utf-8"))["task"]
                    # check if need to skip this job
                    if task and job_task != task:
                        continue
                    job_pending.append(
                        {
                            "uuid": suuid,
                            "client": client_address.decode("utf-8"),
                            "received_timestamp": start,
                            "done_timestamp": None,
                            "task": job_task,
                            "status": "PENDING",
                            "worker": self.name,
                            "service": self.service.decode("utf-8"),
                        }
                    )
    job_completed = []
    # load done jobs
    if completed is True:
        with queue_file_lock:
            with open(self.queue_done_filename, "rb+") as f:
                for entry in f.readlines():
                    job_entry = entry.decode("utf-8").strip()
                    suuid, start, end = job_entry.split("--")  # {suuid}--start--end
                    if uuid and suuid != uuid:
                        continue
                    client_address, empty, juuid, data = loader(
                        request_filename(suuid, self.base_dir_jobs)
                    )
                    if client and client_address.decode("utf-8") != client:
                        continue
                    job_task = json.loads(data.decode("utf-8"))["task"]
                    # check if need to skip this job
                    if task and job_task != task:
                        continue
                    job_completed.append(
                        {
                            "uuid": suuid,
                            "client": client_address.decode("utf-8"),
                            "received_timestamp": start,
                            "done_timestamp": end,
                            "task": job_task,
                            "status": "COMPLETED",
                            "worker": self.name,
                            "service": self.service.decode("utf-8"),
                        }
                    )
    if last:
        return Result(
            task=f"{self.name}:job_list",
            result=job_completed[len(job_completed) - last :]
            + job_pending[len(job_pending) - last :],
        )
    else:
        return Result(
            task=f"{self.name}:job_list",
            result=job_completed + job_pending,
        )

request_filename(suuid, base_dir_jobs)

Returns freshly allocated request filename for given UUID str

Source code in norfab\core\worker.py

def request_filename(suuid: Union[str, bytes], base_dir_jobs: str):
    """Returns freshly allocated request filename for given UUID str"""
    suuid = suuid.decode("utf-8") if isinstance(suuid, bytes) else suuid
    return os.path.join(base_dir_jobs, f"{suuid}.req")

reply_filename(suuid, base_dir_jobs)

Returns freshly allocated reply filename for given UUID str

Source code in norfab\core\worker.py

def reply_filename(suuid: Union[str, bytes], base_dir_jobs: str):
    """Returns freshly allocated reply filename for given UUID str"""
    suuid = suuid.decode("utf-8") if isinstance(suuid, bytes) else suuid
    return os.path.join(base_dir_jobs, f"{suuid}.rep")

event_filename(suuid, base_dir_jobs)

Returns freshly allocated event filename for given UUID str

Source code in norfab\core\worker.py

def event_filename(suuid: Union[str, bytes], base_dir_jobs: str):
    """Returns freshly allocated event filename for given UUID str"""
    suuid = suuid.decode("utf-8") if isinstance(suuid, bytes) else suuid
    return os.path.join(base_dir_jobs, f"{suuid}.event")

_post(worker, post_queue, queue_filename, destroy_event, base_dir_jobs)

Thread to receive POST requests and save them to hard disk

Source code in norfab\core\worker.py

def _post(worker, post_queue, queue_filename, destroy_event, base_dir_jobs):
    """Thread to receive POST requests and save them to hard disk"""
    # Ensure message directory exists
    if not os.path.exists(base_dir_jobs):
        os.mkdir(base_dir_jobs)

    while not destroy_event.is_set():
        try:
            work = post_queue.get(block=True, timeout=0.1)
        except queue.Empty:
            continue
        timestamp = time.ctime()
        client_address = work[0]
        suuid = work[2]
        filename = request_filename(suuid, base_dir_jobs)
        dumper(work, filename)

        # write reply for this job indicating it is pending
        filename = reply_filename(suuid, base_dir_jobs)
        dumper(
            [
                client_address,
                b"",
                suuid,
                b"300",
                json.dumps(
                    {
                        "worker": worker.name,
                        "uuid": suuid.decode("utf-8"),
                        "status": "PENDING",
                        "service": worker.service.decode("utf-8"),
                    }
                ).encode("utf-8"),
            ],
            filename,
        )
        log.debug(f"{worker.name} - '{suuid}' job, saved PENDING reply filename")

        # add job request to the queue_filename
        with queue_file_lock:
            with open(queue_filename, "ab") as f:
                f.write(f"{suuid.decode('utf-8')}--{timestamp}\n".encode("utf-8"))
        log.debug(f"{worker.name} - '{suuid}' job, added job to queue filename")

        # ack job back to client
        worker.send_to_broker(
            NFP.RESPONSE,
            [
                client_address,
                b"",
                suuid,
                b"202",
                json.dumps(
                    {
                        "worker": worker.name,
                        "uuid": suuid.decode("utf-8"),
                        "status": "ACCEPTED",
                        "service": worker.service.decode("utf-8"),
                    }
                ).encode("utf-8"),
            ],
        )
        log.debug(
            f"{worker.name} - '{suuid}' job, sent ACK back to client '{client_address}'"
        )

        post_queue.task_done()

_get(worker, get_queue, destroy_event, base_dir_jobs)

Thread to receive GET requests and retrieve results from the hard disk

Source code in norfab\core\worker.py

def _get(worker, get_queue, destroy_event, base_dir_jobs):
    """Thread to receive GET requests and retrieve results from the hard disk"""
    while not destroy_event.is_set():
        try:
            work = get_queue.get(block=True, timeout=0.1)
        except queue.Empty:
            continue

        client_address = work[0]
        suuid = work[2]
        rep_filename = reply_filename(suuid, base_dir_jobs)

        if os.path.exists(rep_filename):
            reply = loader(rep_filename)
        else:
            reply = [
                client_address,
                b"",
                suuid,
                b"400",
                json.dumps(
                    {
                        "worker": worker.name,
                        "uuid": suuid.decode("utf-8"),
                        "status": "JOB RESULTS NOT FOUND",
                        "service": worker.service.decode("utf-8"),
                    }
                ).encode("utf-8"),
            ]

        worker.send_to_broker(NFP.RESPONSE, reply)

        get_queue.task_done()

_event(worker, event_queue, destroy_event)

Thread function to emit events to Clients

Source code in norfab\core\worker.py

def _event(worker, event_queue, destroy_event):
    """Thread function to emit events to Clients"""
    while not destroy_event.is_set():
        try:
            work = event_queue.get(block=True, timeout=0.1)
        except queue.Empty:
            continue

        client_address = work[0]
        suuid = work[1]
        task = work[2]
        timeout = work[3]
        data = work[4]

        event = [
            client_address,
            b"",
            suuid,
            b"200",
            json.dumps(
                {
                    "worker": worker.name,
                    "service": worker.service.decode("utf-8"),
                    "uuid": suuid.decode("utf-8"),
                    "task": task,
                    "timeout": timeout,
                    **data,
                }
            ).encode("utf-8"),
        ]

        worker.send_to_broker(NFP.EVENT, event)

        event_queue.task_done()

recv(worker, destroy_event)

Thread to process receive messages from broker.

Source code in norfab\core\worker.py

def recv(worker, destroy_event):
    """Thread to process receive messages from broker."""
    while not destroy_event.is_set():
        # Poll socket for messages every second
        try:
            items = worker.poller.poll(1000)
        except KeyboardInterrupt:
            break  # Interrupted
        if items:
            msg = worker.broker_socket.recv_multipart()
            log.debug(f"{worker.name} - received '{msg}'")
            empty = msg.pop(0)
            header = msg.pop(0)
            command = msg.pop(0)

            if command == NFP.POST:
                worker.post_queue.put(msg)
            elif command == NFP.DELETE:
                worker.delete_queue.put(msg)
            elif command == NFP.GET:
                worker.get_queue.put(msg)
            elif command == NFP.KEEPALIVE:
                worker.keepaliver.received_heartbeat([header] + msg)
            elif command == NFP.DISCONNECT:
                worker.reconnect_to_broker()
            else:
                log.debug(
                    f"{worker.name} - invalid input, header '{header}', command '{command}', message '{msg}'"
                )

        if not worker.keepaliver.is_alive():
            log.warning(f"{worker.name} - '{worker.broker}' broker keepalive expired")
            worker.reconnect_to_broker()