101 lines
4.1 KiB
Python
101 lines
4.1 KiB
Python
import configparser
|
|
import importlib
|
|
import logging
|
|
import sys
|
|
import os
|
|
import random
|
|
from pyexecplugins.pyexecplugins import *
|
|
from conductor.ConductorWorker import ConductorWorker
|
|
|
|
class PyExec():
|
|
errors = { "info" : logging.INFO, "error" : logging.ERROR, "debug" : logging.DEBUG}
|
|
|
|
def __init__(self, pluginlist):
|
|
self.init(pluginlist)
|
|
|
|
def computeDomain(self, plg):
|
|
domain = self.getDomain(plg)
|
|
commondomain = self.getDomain("common")
|
|
if domain is not None and commondomain is not None:
|
|
domain = domain + "," + commondomain
|
|
elif commondomain is not None:
|
|
domain = commondomain
|
|
return domain
|
|
|
|
def getDomain(self, plg):
|
|
if self.cfg.has_section(plg):
|
|
return self.cfg[plg].get("domain")
|
|
else:
|
|
return None
|
|
|
|
def init(self, pluginlist):
|
|
self.cfg = configparser.ConfigParser()
|
|
self.cfg.read("config.cfg")
|
|
self.workerid = self.cfg["common"].get("workerid", "pythonworker-" + str(random.randint(1000,10000)))
|
|
|
|
logging.basicConfig(level=self.errors[self.cfg["common"]["loglevel"]])
|
|
logging.info("Initializing PyExec worker %s with config.cfg", self.workerid)
|
|
|
|
self.threads = self.cfg["common"].getint("threads", 3)
|
|
self.pollrate = self.cfg["common"].getfloat("pollrate", .1)
|
|
self.server = os.environ.get('CONDUCTOR_SERVER', self.cfg["common"].get("server", "http://localhost:8080/api"))
|
|
for plg in pluginlist:
|
|
path = "pyexecplugins." + plg
|
|
try:
|
|
importlib.import_module(path)
|
|
except Exception as err:
|
|
logging.warning("Skipping plugin %s %s", path, err)
|
|
continue
|
|
|
|
PyExecPlugin.registerTaskDefinitions(self.server)
|
|
|
|
def start(self):
|
|
logging.info("Started PyExec with plugins: %s", PyExecPlugin.getPluginNames())
|
|
cc = ConductorWorker(self.server, self.threads, self.pollrate, self.workerid)
|
|
|
|
# start workers for all included plugins that are available also as standalone tasks ...
|
|
for plg in PyExecPlugin.getPlugins():
|
|
plugin = PyExecPlugin.get(plg)
|
|
if plugin.supportsStandalone(plugin):
|
|
domain = self.computeDomain(plugin.taskdef["name"])
|
|
cc.start(plugin.taskdef["name"], self.pyexec, False, domain)
|
|
|
|
# ... plus fallback pyexec for managing non standalone calls, Nop, Identity and Sequence
|
|
domain = self.computeDomain("pyexec")
|
|
cc.start('pyexec', self.pyexec, True, domain)
|
|
|
|
def pyexec(self, task):
|
|
try:
|
|
logging.info("Executing task {} of type {}[{}] from wkf {}[{}]".format(
|
|
task["workflowTask"]["taskReferenceName"],
|
|
task["taskDefName"],
|
|
task["taskId"],
|
|
task["workflowType"],
|
|
task["workflowInstanceId"]))
|
|
|
|
# Get operation from input or task type (in case of standalone tasks) fallback to Nop
|
|
operation = task["inputData"].get("operation") or task["taskDefName"] or "Nop"
|
|
logging.debug("Operation is %s", operation)
|
|
|
|
# Get plugin by checking either name or alias (in case of standalone tasks the task type is aliased to the plugin name)
|
|
p = PyExecPlugin.get(operation) or PyExecPlugin.getAlias(operation)
|
|
logging.debug("Plugin is %s", p)
|
|
|
|
if p != None:
|
|
pi = p(task["inputData"])
|
|
ret = pi.execute()
|
|
|
|
else: raise Exception("Operation {} not found.".format(operation))
|
|
return { "status" : "COMPLETED", "output" : ret, "logs" : ["one","two"]}
|
|
|
|
except Exception as exc:
|
|
exc_type, exc_obj, exc_tb = sys.exc_info()
|
|
fname = os.path.split(exc_tb.tb_frame.f_code.co_filename)[1]
|
|
logging.debug(exc_type, fname, exc_tb.tb_lineno)
|
|
logging.error(str(exc))
|
|
return { "status" : "FAILED", "output" : { "message" : str(exc) }, "logs" : ["one","two"]}
|
|
|
|
if __name__ == '__main__':
|
|
pyexec = PyExec(sys.argv[1:])
|
|
pyexec.start()
|