Add worker info proposal/discussion

[spapas]

Hey friends, I've implemented some functionality to know what each of my workers is doing.

If you think it is useful I can provide a PR for this.

We'll use a WorkerInfo model like:

class WorkerInfo(models.Model):
    thread_name = models.CharField(max_length=128, unique=True)
    started_on = models.DateTimeField()
    status = models.CharField(max_length=128, default="WAITING")

    task_name = models.CharField(max_length=128, blank=True, null=True)
    task_started_on = models.DateTimeField(blank=True, null=True)
    task_finished_on = models.DateTimeField(blank=True, null=True)

(and maybe add a few more fields if needed)

Then register the following signals:

@on_startup()
def consumer_startup():
    random.seed()
    tname = threading.current_thread().name
    logger.info(colored(f"Consumer startup (hook): {tname}, {get_hostname()}, {os.getpid()}", "green"))
    with threading.Lock(), transaction.atomic():
        wi, _created = WorkerInfo.objects.get_or_create(thread_name=tname, defaults={"started_on": timezone.now()})
        wi.started_on = timezone.now()
        wi.status = "WAITING"
        wi.save()

@on_shutdown()
def consumer_shutdown():
    time.sleep(random.randint(1, 1000) / 1000)
    tname = threading.current_thread().name
    logger.info(colored(f"Consumer shutdown (hook): {tname}, {get_hostname()}, {os.getpid()}", "red"))
    with threading.Lock(), transaction.atomic():
        WorkerInfo.objects.filter(thread_name=tname).update(status="STOPPED")

@signal(SIGNAL_COMPLETE)
def task_success(signal, task):
    tname = threading.current_thread().name
    with transaction.atomic():
        wi, _created = WorkerInfo.objects.get_or_create(thread_name=tname)
        wi.status = "WAIITING"
        wi.task_finished_on = timezone.now()
        wi.save()

@signal(SIGNAL_EXECUTING)
def task_start(signal, task):
    tname = threading.current_thread().name
    with transaction.atomic():
        wi, _created = WorkerInfo.objects.get_or_create(thread_name=tname)
        wi.status = "RUNNING"
        wi.task_name = str(task)
        wi.task_started_on = timezone.now()
        wi.task_finished_on = None
        wi.save()

Then we'll get sometihng like this on our admin:

I think this is useful (at least for me) to know that my workers are active and to see what they are doing (for example if a task has stuck).

Kind regards,
Serafeim

[jedie]

Interesting idea. Would be great to see more live information about the consumer.

But i think this is only a solution for "thread" worker type, isn't it? See: https://huey.readthedocs.io/en/latest/consumer.html#worker-types

At least you have to add "process id", too.

[cnschn]

To add to this, even with worker type "thread" you can still have multiple worker machines, so you would need the host name as well.

This has (at least in our use case) the additional complication that new worker machines are spawned with changing hostnames, so with every new deploy you would end up with a new set of N workers, with the old ones being "abandoned" in a sense. Maybe That's not actually an issue, and it's okay that the get_or_create sets up new WorkerInfo instances in this case, and the old ones are either not displayed in the overview, or clearly displayed as "expired" when there was no update from them in a while?

For the single worker machine - multiple worker thread setup I like this approach, but I'd love to extend it to the more general case.

[spapas]

Hello friends! I'd be happy to add the process id and the hostname. Also, concerning the "abandoned" workers it should be possible to identify them using the (started_on, task_started_on) tuple. If there are not other comments I'll provide a PR for this and we can discuss it more there.