django-orchestra/orchestra/apps/orchestration/manager.py

78 lines
2.4 KiB
Python
Raw Normal View History

import logging
2014-05-08 16:59:35 +00:00
import threading
from django import db
2014-05-13 13:46:40 +00:00
from orchestra.utils.python import import_class
2014-05-08 16:59:35 +00:00
from . import settings
from .helpers import send_report
logger = logging.getLogger(__name__)
2014-05-08 16:59:35 +00:00
def as_task(execute):
def wrapper(*args, **kwargs):
2014-07-11 21:09:17 +00:00
with db.transaction.commit_manually():
log = execute(*args, **kwargs)
db.transaction.commit()
2014-05-08 16:59:35 +00:00
if log.state != log.SUCCESS:
send_report(execute, args, log)
return log
return wrapper
def close_connection(execute):
""" Threads have their own connection pool, closing it when finishing """
# TODO rewrite as context manager
def wrapper(*args, **kwargs):
log = execute(*args, **kwargs)
db.connection.close()
# Using the wrapper function as threader messenger for the execute output
wrapper.log = log
return wrapper
def execute(operations):
""" generates and executes the operations on the servers """
2014-05-13 13:46:40 +00:00
router = import_class(settings.ORCHESTRATION_ROUTER)
2014-05-08 16:59:35 +00:00
# Generate scripts per server+backend
scripts = {}
2014-07-17 16:09:24 +00:00
cache = {}
2014-05-08 16:59:35 +00:00
for operation in operations:
logger.info("Queued %s" % str(operation))
2014-07-17 16:09:24 +00:00
servers = router.get_servers(operation, cache=cache)
2014-05-08 16:59:35 +00:00
for server in servers:
key = (server, operation.backend)
if key not in scripts:
scripts[key] = (operation.backend(), [operation])
2014-07-25 15:17:50 +00:00
scripts[key][0].prepare()
2014-05-08 16:59:35 +00:00
else:
scripts[key][1].append(operation)
method = getattr(scripts[key][0], operation.action)
method(operation.instance)
# Execute scripts on each server
threads = []
executions = []
for key, value in scripts.iteritems():
server, __ = key
backend, operations = value
backend.commit()
execute = as_task(backend.execute)
execute = close_connection(execute)
thread = threading.Thread(target=execute, args=(server,))
thread.start()
threads.append(thread)
executions.append((execute, operations))
[ thread.join() for thread in threads ]
logs = []
for execution, operations in executions:
for operation in operations:
logger.info("Executed %s" % str(operation))
2014-05-08 16:59:35 +00:00
operation.log = execution.log
operation.save()
logger.info(execution.log.stderr)
2014-05-08 16:59:35 +00:00
logs.append(execution.log)
return logs