Commit 5977108a authored by kaiyou's avatar kaiyou

Add a proper python-click based CLI and testing commands

parent 3be35a02
import celery
from amonit import config, scheduler
def schedule(sender, **kwargs):
sender.scheduler.schedule(sender)
def main(conf=None):
# Load the configuration
if not conf:
conf = config.load()
# Create and initialize the celery app
app = celery.Celery('amonit', broker=conf["general"]["broker"])
app.scheduler = scheduler.Scheduler(conf)
app.on_after_configure.connect(schedule)
return app
main()
import click
import threading
import json
from amonit import scheduler, config
@click.group()
def cli():
pass
@cli.group()
def test():
pass
@cli.command()
def worker():
app = scheduler.Scheduler.get_app()
app.Worker().start()
@cli.command()
def manager(worker=False):
app = scheduler.Scheduler.get_app()
app.scheduler.schedule(app)
app.Beat().run()
@cli.command()
def standalone():
app = scheduler.Scheduler.get_app()
app.scheduler.schedule(app)
worker = app.Worker()
thread = threading.Thread(target=worker.start)
thread.start()
app.Beat().run()
@test.command()
@click.option("--function", required=True, help="Which check to test")
@click.option("--context", default="{}", help="JSON string decoded as the context")
@click.option("--args", default="{}", help="JSON string decoded as the arguments")
def check(function, context, args):
conf = config.load()
conf["notifiers"] = {}
app = scheduler.Scheduler.get_app(conf=conf)
print(scheduler.check_run(
"dummy", function, json.loads(context), json.loads(args)
))
if __name__ == "__main__":
cli()
def dashboard(name, state):
pass
......@@ -2,7 +2,7 @@ import celery
import redis
import json
from amonit import util
from amonit import util, config
class State(object):
......@@ -29,6 +29,14 @@ class Scheduler(object):
self.config = config
self.state = State(config["general"]["storage"])
@classmethod
def get_app(cls, conf=None):
if not conf:
conf = config.load()
app = celery.Celery('amonit', broker=conf["general"]["broker"])
app.scheduler = cls(conf)
return app
def schedule(self, app):
""" Called by celery upon configuration so we can add beat tasks
"""
......@@ -79,6 +87,7 @@ def check_run(checkid, function, context, args):
stored.update(checkid=checkid, function=function)
result = util.resolve(function)(stored, **args)
celery.current_app.scheduler.update(checkid, stored, result)
return result
@celery.current_app.task
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment