trixterfilm / jobtronaut Goto Github PK
View Code? Open in Web Editor NEWTractor API extension for authoring reusable task hierarchies.
Home Page: https://jobtronaut.readthedocs.io
License: BSD 3-Clause "New" or "Revised" License
Tractor API extension for authoring reusable task hierarchies.
Home Page: https://jobtronaut.readthedocs.io
License: BSD 3-Clause "New" or "Revised" License
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.
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.
Provide TRACTOR_ENGINE
in the jobtronaut python configuration.
Make it easier to spool jobs to another Tractor engine server.
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))
...
The get_arguments_objects() utillity function finds the arguments of a task even after that task "command" got 'neutralized'.
I'd like to inspect argument of 'neutralized' tasks for debugging purposes.
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.
Exend the jobtronaut
cli submit
command and provide --expandchunk
as bool flag.
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.
jobtronaut submit --task SomeTask --expandchunk
would not submit, but only print TR_EXPAND_CHUNK <temp_alf_file>
to STDOUT.
A simple commandline call to print arguments used in a particular task/command.
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.
Expample usage:
jobtronaut arguments <TR_JID>:<TR_TID>
Implement functionality to inherit, override and extend argument processors at TaskWithOverrides
.
'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.
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.
The jobtronaut commandline interface's submit
command is missing the maxactive option.
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).
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.
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}
Have a possibility to inject different additional commandflags for resolved commands.
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.
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.
For the task expansion via Task.__EXPAND__
we need a way to apply different argument dictionaries to required tasks of the same.
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.
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.
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.
Allow modify_cmds
and modify_tasks
to use predicates as value.
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.
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]
)
A dispatcher that converts a missioncontrol nodegraph into static python code containing Hierarchy Task classes.
The resulting python module can then be inspected by the Plugin system. The resulting hierarchies can then be submitted to tractor via jobtronaut.
Enable the option to create jobs that can run exclusively on the spoolhost.
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.
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):
...
Conditionally stop further task expansion based on a condition that's evaluated at hierarchy creation time.
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.
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.
Allow inspection of required arguments for command tasks. This information can be listed via jobtronaut info
command.
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.
Introduce a contextmanager that simplifies the report of a progress via TR_PROGRESS
.
Handling of progress reports can be simplified.
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
Categorize sitestatusfilter into categories BladeFilter
and CommandFilter
. Allow to configure a callable that will return the information what *Filter plugin to delegate to.
Right now the sitestatusfilters are static and will be picked up if
SiteModulesPaths
for a specific blade profileTractorSiteStatusFilter.py
TractorSiteStatusFilter
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.
Extend the plugins discovery to find and load site status filters and implement a global site status filter that delegates to the plugin implementation.
Tractors Site status will only be updated when reloading the blade configuration. Our plugin discovery can help in that regard.
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.
Make it possible to batch submit multiple jobs in the jobtronaut commandline interface.
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.
In Python we can express the task/job dependencies via a list vs. tuple syntax. How can we do this in the cli though?
Redesign the plugins discovery to be more generic and not only consider tasks.py
and processors.py
modules.
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
.
A declarative, efficient, and flexible JavaScript library for building user interfaces.
๐ Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
An Open Source Machine Learning Framework for Everyone
The Web framework for perfectionists with deadlines.
A PHP framework for web artisans
Bring data to life with SVG, Canvas and HTML. ๐๐๐
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
Some thing interesting about web. New door for the world.
A server is a program made to process requests and deliver data to clients.
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
Some thing interesting about visualization, use data art
Some thing interesting about game, make everyone happy.
We are working to build community through open source technology. NB: members must have two-factor auth.
Open source projects and samples from Microsoft.
Google โค๏ธ Open Source for everyone.
Alibaba Open Source for everyone
Data-Driven Documents codes.
China tencent open source team.