Luigi: Luigi failure events with multiple workers

[kashifjt]

I recently started using luigi for some of the data extraction/processing tasks at work and it is working perfectly, using multiple workers. But I am having some issues while executing a callback method for FAILURE events. I am planning to use this method to create some sort of exception logging for all the failed tasks. e.g. I am trying to use the following code, which is provided in the documentation:

@luigi.Task.event_handler(luigi.Event.FAILURE)
def mourn_failure(task, exception):
print(exception.__str__())

This code does not even get executed, even if there are some task failures. But I noticed that if I run my whole pipeline with single worker, then it starts working fine and gets executed, every time a task fails. So, I don't know why, but this issue only occurs when I run my tasks in parallel using multiple workers.

[fjavieralba]

Hi @kashifjt why did you close the issue? did you find a solution?

I'm having trouble too with luigi.Event.FAILURE methods, I cannot get them executed when my task fails.

I have a task failing sometimes with a Luigi Framework Error, and I also have a method like:

@MyTask.event_handler(luigi.Event.FAILURE)
def on_my_task_failure(task, exception):
    logger.info("removing intermediate data")
    os.remove(task.input_file)

But it never gets executed.

Any hint on this?

thanks!

[kwilcox]

Is your task a WrapperTask and failing in the requires() method? That won't trigger a FAILURE event.

[fjavieralba]

mmm yes, thats my case, thank you for your comment.

[kwilcox]

@fjavieralba Here is my current solution that will raise a FAILURE event if a WrapperTask fails in the requires() method. Keep in mind we have multiple people contributing to and adding tasks into our codebase so the first priority was maintaining compatibility with the Luigi documentation. If we were willing to to break compatibility our solution might have been a bit cleaner.

@contextmanager
def wrapper_failure(task):
    try:
        yield
    except GeneratorExit as e:
        raise e  # Don't break luigi
    except BaseException as e:
        task.trigger_event(luigi.Event.FAILURE, task, e)
        raise e

And then on every WrapperTask, we just wrap the entire contents of the requires() method:

class SomeAwesomeWrapperTask(luigi.WrapperTask):
    ...
    def requires(self):
        with wrapper_failure(self):
           # do things

[fjavieralba]

thank you very much for you detailed response @kwilcox!
I will probably use the same approach 😄

Much appreciated!