trixterfilm / jobtronaut Goto Github PK

Goal

Instead of using a second command that neutralizes all commands of the task it is running on we want to handle it within a single command when Task.NO_RETRY is set.

Motivation

Other than expected Tractor doesn't schedula a task's command immediately after the prior one. This leads to the scnerario, that when a job gets restarted entirely another expansion would happen.

Goal

Provide TRACTOR_ENGINE in the jobtronaut python configuration.

Motivation

Make it easier to spool jobs to another Tractor engine server.

Implementation

Job.spool() exposes hostname and port arguments. Within our submit method we can easily resolve the configuration.

class Job(author.Job):
    ...
    def submit(self, dump_job=True, **kwargs):
       ...
       # resolve configuration.TRACTOR_ENGINE
       # split hostname and port, declare `resolved_hostname` and `resolved_port`
        job_id = self.spool(owner=getpass.getuser(), hostname=kwargs.get("hostname", resolved_hostname), port=kwargs.get("port", resolved_port))
       ...

Goal

The get_arguments_objects() utillity function finds the arguments of a task even after that task "command" got 'neutralized'.

Motivation

I'd like to inspect argument of 'neutralized' tasks for debugging purposes.

Potential Implementation

a) The original command gets stored on the task when being neutralized so that one can read that instead.
b) The original command gets decoded from the 'neutralized' command on the fly.

Goal

Exend the jobtronaut cli submit command and provide --expandchunk as bool flag.

Motivation

For interoperability between other tractor facing apis it is great to use the jobtronaut cli to build the underlying .alf representation of a job/root task and use that for expansion.

Potential Implementation

jobtronaut submit --task SomeTask --expandchunk would not submit, but only print TR_EXPAND_CHUNK <temp_alf_file> to STDOUT.

Goal

A simple commandline call to print arguments used in a particular task/command.

Motivation

As the arguments are encoded a user would have to manually decode it. Although there is a python convenience function we're lacking a way of doing that in the CLI which would be very handy.

Potential Implementation

Expample usage:

jobtronaut arguments <TR_JID>:<TR_TID>

Goal

Implement functionality to inherit, override and extend argument processors at TaskWithOverrides.

Motivation

'TaskWithOverrides' allows customization of existing tasks and override different parameters like 'argument_processors'. The 'argument_processors' argument accepts a list of 'ProcessDefinitions' which replaces the original entirely.
To not only replace all processors but rather modify existing list we can define operations similar to list operations - 'remove, replace, prepend, append' and apply them accordingly. This way we can finer control overrides.

Potential Implementation

Example Usage

TaskWithOverrides(
    "FooBarTask",
    argument_processors=TaskWithOverrides.ArgumentProcessorOverrides(
        {
            "prepend": [ProcessorDefinition1, ProcessorDefinition2],  # accepts list/tuple of ProcessorDefinitions or single ProcessorDefinition
            "replace": {0: ProcessorDefinition3, 2: ProcessorDefinition4},  # requires mapping of {index: ProcessorDefinition}
            "remove": [1, 3],  #accepts list/tuple of indices or single index as int
            "append": [ProcessorDefinition5, ProcessorDefinition6]  # accepts list/tuple of ProcessorDefinitions or single ProcessorDefinition
        }
    )
)

Define object ('ArgumentProcessorOverrides') which holds processor definitions per operation and use this instead of a list. e.g. argument_processor = ArgumentProcessorOverrides(..).

Extend TaskWithOverrides.get in order to decide if a list or 'ArgumentProcessorOverrides' object was provided and modify the overrides by implementing the operations 'remove, replace, prepend, append'. In case of 'remove' and 'replace' the operation are based on the index of the original defined processor list.

Goal

The jobtronaut commandline interface's submit command is missing the maxactive option.

Goal

Introduce an env method on Task that would return a dict that will be added converted and added as envkey on the corresponding command(s).

Motivation

It can be mandatory for a task to provide its own environment (overrides). Instead of setting it as a class attribute it should have access to self.arguments to allow conversion/mapping of job arguments to environment variables.

Potential Implementation

class FoobarTask(Task):

    argument_defaults = {"some_arg": "foobar"}


    def cmd(self):
        return ["/bin/echo", "$SOME_ARG"]

    def env(self)
        return {"SOME_ARG": self.arguments.foobar.processed}

Goal

Port the internally implemented hierarchy pruning to jobtronaut.

Motivation

To avoid needlessly long chains of tasks that basically to nothing, we want to prune all those tasks and just keep the absolute minimum number of tasks to define the hierarchy.

Goal

Have a possibility to inject different additional commandflags for resolved commands.

Motivation

Applying additional commandflags for commands is pretty useful, especially in a production environment using an application bootstrapping system. But not always you want to inject the same flags for all commands in a task dependency. We need a more fine-grain option to define other flags or exclude flags for specific commands entirely.

Implementation

The argument that defines the additional commands should be able to consume a dict where the key represents a regex that would be checked against the resolved executable and the value that gets injected in case it matches.

Goal

For the task expansion via Task.__EXPAND__ we need a way to apply different argument dictionaries to required tasks of the same.

Motivation

Right now the Task.__EXPAND__ works great if you want to use it with different required task names. When having multiple required tasks of the same name, the given arguments dictionary will be applied to all of them.

Potential Implementation/Solution

Example 1 current situation
This applies one argument dictionary to all Foobar task references.

from jobtronaut.author import Task

class ExpandTask(Task):
    elements_id = "some_var"

    def cmd(self):
        return ["python", "-c"]

    def script(self):

        self.__EXPAND__(
            "RootTask",
            {
                "FooTask": {"some_var": "hello all!"}
            }
        )

class RootTask(Task):
    required_tasks = (
        "FooTask",
        ("FooTask", ["FooTask", "FooTask"])
    )

class FooTask(Task):
    elements_id = "some_var"

    def cmd(self):
        return ["/bin/echo", "\"{}\"".format(self.arguments.some_var.processed)]`

Example 2 new possibility
Wwe can use a list/tuple and provide different argument dictionaries. These will be applied in inspection order and the number of given argument dictionaries have to match the amount of required tasks.

class ExpandTask(Task):
    elements_id = "some_var"

    def cmd(self):
        return ["python", "-c"]

    def script(self):

        self.__EXPAND__(
            "RootTask",
            {
                "FooTask": [
                    {"some_var": "hello!"}, 
                    {"some_var": "hello foo"}, 
                    {"some_var": "goodbye"}, 
                    {"some_var": "ciao"}
                ]
            }
        )

class RootTask(Task):
    required_tasks = (
        "FooTask",  # -> hello!
        (
            # -> hello foo
            "FooTask", [
                "FooTask", # -> goodbye
                "FooTask"  # -> ciao
            ]
        )
    )

class FooTask(Task):
    elements_id = "some_var"

    def cmd(self):
        return ["/bin/echo", "\"{}\"".format(self.arguments.some_var.processed)]

Given a task like that

class Foo(Task):
   required_tasks = "Bar"

results in the problem that the plugin discovery looks up for B, a and r plugins as it doesn't check if the given value for required_tasks is a supported list or tuple. We can handle that more carefully.

Problem Statement

If a task defines an argument as elements_id which datatype is a dict then we only iterate over the keys via Task.Flags.PER_ELEMENT instead of the key value pair.

Goal

Allow modify_cmds and modify_tasks to use predicates as value.

Motivation

At the moment each modify_cmds and modify_tasks call can except a single value to set. This works fine for many cases, but is insufficient if different commands would need different tags, especially if the decision can only be made via argv inspection.

Potential Implementation

current usage

job.modif_cmds(
    predicate=decide_something,  # returns the True/False state only
    attribute="tags",
    value=["foo", "bar"]
)

possible additional usage

job.modif_cmds(
    predicate=lambda x: decide_something()[0],  # return a tuple (True/False state, tags)
    attribute="tags",
    value=lambda x: decide_something()[1]
)

Goal

A dispatcher that converts a missioncontrol nodegraph into static python code containing Hierarchy Task classes.

Motivation

The resulting python module can then be inspected by the Plugin system. The resulting hierarchies can then be submitted to tractor via jobtronaut.

Goal

Enable the option to create jobs that can run exclusively on the spoolhost.

Motivation

The ability of binding jobs or task dependencies to the spoolhost is quite useful when other hosts have no access to the same environment or are simply occupied constantly.

Implementation

Creation of a job that would only produce commands of type local.

from jobtronaut.author import Job

job = Job("RootTask", {"my_argument": "foobar"}, local=Tue)

As an expansion would create new job instances too, we also need the option to define the ability of further binding new commands to the spoolhost via the dedicated expansion method. A default of local=None would automatically inherit the state of the current command assuming it would be called from within a running command on Tractor.

class Task(Task):
   ...
    @staticmethod
    def __EXPAND__(root_task_name, arguments_mapping, local=None):
       ...

Goal

Conditionally stop further task expansion based on a condition that's evaluated at hierarchy creation time.

Motivation

We currently have to duplicate a lot of Hierachy Tasks with different combinations of required_tasks and TaskWithOverrides to make conditional adjustments to our hierarchy at submission time.
It would be beneficial if the hierarchy creation could be interrupted at any point based on a flexible condition that's evaluated during hierarchy creation. This way it will have access to the full task definition as well as all the currently available arguments to make a decision.

Implementation

Add a method to the Task class called stop_traversal.
The implementation in the Base class will return False to not alter any existing functionality. This is also the expected behaviour if we won't override the method in the inheriting class.
This method will have access to the task object itself via self and will be handed all the arguments.
The skip method should return a boolean value or a value that can be automatically coerced into a boolean.
If the return value is True, we will not continue generating the task hierarchy in that specific branch.
If the return value is False we will continue the hierarchy creation normally.

Goal

Allow inspection of required arguments for command tasks. This information can be listed via jobtronaut info command.

Motivation

Picking up a command task to reuse in another task dependency often leads to the situation, that I need to know what arguments this task expects. Currently you can only figure this out by jumping to the source of the code and manually check or simply try and error.

The current implementation of jobtronaut.query.initialize_engine() lets tractor deal with the engine resolution instead of picking it properly from the jobtronaut configuration if defined.

Goal

Introduce a contextmanager that simplifies the report of a progress via TR_PROGRESS.

Motivation

Handling of progress reports can be simplified.

Potential Implementation

Example usage

class DoSomethingTask(Task):

    def cmd(self):
        return ["/some/python-executable", "-c"]

    def script(self):

        with self.report_progress(max=10) as progress:
            for i in range(0, 9): 
                progress += 1

Goal

Categorize sitestatusfilter into categories BladeFilter and CommandFilter. Allow to configure a callable that will return the information what *Filter plugin to delegate to.

Motivation

Right now the sitestatusfilters are static and will be picked up if

1. the module is in the SiteModulesPaths for a specific blade profile
2. the module is name TractorSiteStatusFilter.py
3. the filter class itself itself is named TractorSiteStatusFilter

Potential Implementation

Introduce the following mechanism within the jobtronaut configuration and the global TractorSiteStatusFilter.py delegate.

BLADEFILTER_CONDUCTOR = lambda stateDict: "MyCustomBladeFilter"  # --> returns the name of the sitestatusfilter plugin and has access to the passed stateDict
COMMANDFILTER_CONDUCTOR = lambda cmd: "MyCustomCommandFilter" # --> returns the name of the sitestatusfilter plugin and has access to the passed CmdTracker instance

The delegate function just calls BLADEFILTER_CONDUCTER and COMMANDFILTER_CONDUCTOR directly based on the passed arguments and can dynamically decide what sitestatusfilter plugin to delegate to.

Goal

Extend the plugins discovery to find and load site status filters and implement a global site status filter that delegates to the plugin implementation.

Motivation

Tractors Site status will only be updated when reloading the blade configuration. Our plugin discovery can help in that regard.

Potential Implementation

For now a site status filter plugin will look identical to the TrStatusFilter class interface. A global TractorSiteStatusFilter class will directly delegate each function call to the plugin that we can implement.

Goal

Make it possible to batch submit multiple jobs in the jobtronaut commandline interface.

Motivation

The current jobtronaut cli is missing the option that we have in the API to submit multiple jobs at once. Although it is possible via a shell script to create a simple batch submission, merging multiple jobs sharing the same root task with different arguments sets can only be done via python yet.
Especially in combination with the --expandchunk flag a new option for handling this would be very powerful and make porting new subhierchies to jobtronaut even more attractive.

Difficulties

In Python we can express the task/job dependencies via a list vs. tuple syntax. How can we do this in the cli though?

Goal

Redesign the plugins discovery to be more generic and not only consider tasks.py and processors.py modules.

Motivation

The number of Task and Processor implementations can grow rapidly. It is useful to place plugins in different modules without having these modules in individual folders that needs to be added individually via PLUGIN_PATH.