Author: tmckay Date: 2012-03-30 20:27:05 +0000 (Fri, 30 Mar 2012) New Revision: 5280
Added: branches/play_plumage/cumin/bin/cumin-report branches/play_plumage/cumin/model/admin/ branches/play_plumage/cumin/model/admin/cumin.xml branches/play_plumage/cumin/model/admin/rosemary.xml branches/play_plumage/cumin/model/plumage/ branches/play_plumage/cumin/model/plumage/plumage.xml branches/play_plumage/cumin/model/plumage/rosemary.xml branches/play_plumage/mint/python/mint/plumage/ branches/play_plumage/mint/python/mint/plumage/__init__.py branches/play_plumage/mint/python/mint/plumage/main.py branches/play_plumage/mint/python/mint/plumage/session.py branches/play_plumage/mint/python/mint/plumage/update.py Removed: branches/play_plumage/cumin/model/plumage.xml Modified: branches/play_plumage/cumin/bin/cumin branches/play_plumage/cumin/bin/cumin-admin branches/play_plumage/cumin/bin/cumin-data branches/play_plumage/cumin/bin/cumin-web branches/play_plumage/cumin/etc/module.profile branches/play_plumage/cumin/model/cumin.xml branches/play_plumage/cumin/model/rosemary.xml branches/play_plumage/cumin/python/cumin/config.py branches/play_plumage/cumin/python/cumin/main.py branches/play_plumage/cumin/python/cumin/messaging/brokergroup.py branches/play_plumage/cumin/python/cumin/model.py branches/play_plumage/cumin/python/cumin/parameters.py branches/play_plumage/mint/python/mint/model.py branches/play_plumage/rosemary/python/rosemary/model.py Log: Initial framework for cumin-report. Fair amout of refactoring around config.py, rosemary/model
Modified: branches/play_plumage/cumin/bin/cumin =================================================================== --- branches/play_plumage/cumin/bin/cumin 2012-03-30 17:17:00 UTC (rev 5279) +++ branches/play_plumage/cumin/bin/cumin 2012-03-30 20:27:05 UTC (rev 5280) @@ -80,18 +80,28 @@ help="Configuration section names for cumin-data instances."\ "\nEach value implies a separate cumin-data instance.")
+ parser.add_option("--reports", dest="reports", default=values.reports, + help="Configuration section names for cumin-report instances."\ + "\nEach value implies a separate cumin-report instance.") + parser.add_option("--console", dest="console", action="store_true", default=False, help="Log to stderr rather than master.log, no IO redirection for children.")
+ parser.add_option("--web-options", dest="web_options", default="", type=str, + help="Additional options string to pass to web instances."\ + "\nEnclose in quotes, options must be --option form, splits on spaces."\ + '\nExample: web_options="--debug --port=12345"') + parser.add_option("--data-options", dest="data_options", default="", type=str, help="Additional options string to pass to data instances."\ "\nEnclose in quotes, options must be --option form, splits on spaces."\ '\nExample: data_options="--print-events=5 --print-stats"')
- parser.add_option("--web-options", dest="web_options", default="", type=str, - help="Additional options string to pass to web instances."\ - "\nEnclose in quotes, options must be --option form, splits on spaces."\ - '\nExample: web_options="--debug --port=12345"') + parser.add_option("--report-options", dest="report_options", default="", type=str, + help="Additional options string to pass to report instances."\ + "\nEnclose in quotes, options must be --option form, splits on spaces."\ + '\nExample: report_options="--print-events=5 --print-stats"') + parser.add_option("--syslog", dest="syslog", action="store_true", default=False, help="Log general error notfications to syslog. Intended for systemd")
@@ -139,6 +149,11 @@ for instance in options.datas.split(','): args, prog_string = get_args("cumin-data", instance, options.init_only, console, options.data_options) apps.append([None, args, prog_string]) + + if len(options.reports) > 0: + for instance in options.reports.split(','): + args, prog_string = get_args("cumin-report", instance, options.init_only, console, options.report_options) + apps.append([None, args, prog_string])
# Launch and babysit, do not restart if options.init_only is set complete = 0
Modified: branches/play_plumage/cumin/bin/cumin-admin =================================================================== --- branches/play_plumage/cumin/bin/cumin-admin 2012-03-30 17:17:00 UTC (rev 5279) +++ branches/play_plumage/cumin/bin/cumin-admin 2012-03-30 20:27:05 UTC (rev 5280) @@ -65,10 +65,9 @@ print "Command '%s' is unknown" % name sys.exit(1)
- broker_uris = [x.strip() for x in opts.brokers.split(",")] authmech = [x.strip() for x in values.common.auth.split(";")]
- app = Cumin(config.get_home(), broker_uris, opts.database, authmech=authmech) + app = Cumin(config.get_home(), [], opts.database, authmech=authmech)
app.check()
Modified: branches/play_plumage/cumin/bin/cumin-data =================================================================== --- branches/play_plumage/cumin/bin/cumin-data 2012-03-30 17:17:00 UTC (rev 5279) +++ branches/play_plumage/cumin/bin/cumin-data 2012-03-30 20:27:05 UTC (rev 5280) @@ -137,7 +137,7 @@ mint = None setup_initial_logging()
- parser = CuminOptionParser() + parser = BrokeredOptionParser()
# Add additional parameters for data parser.add_option("--print-stats", action="store_true", default=False) @@ -169,7 +169,8 @@ log.error("Extra arguments:" + "".join([" "+arg for arg in args])) raise ArgError
- model_dir = os.path.join(config.home, "model") + + model_dir = [os.path.join(config.home, x) for x in ("model", "model/admin")]
broker_uris = [x.strip() for x in opts.brokers.split(",")]
Added: branches/play_plumage/cumin/bin/cumin-report =================================================================== --- branches/play_plumage/cumin/bin/cumin-report (rev 0) +++ branches/play_plumage/cumin/bin/cumin-report 2012-03-30 20:27:05 UTC (rev 5280) @@ -0,0 +1,243 @@ +#!/usr/bin/python + +import os +import sys +from traceback import print_exc + +home = os.environ.get("CUMIN_HOME", os.path.normpath("/usr/share/cumin")) +sys.path.append(os.path.join(home, "python")) + +from cumin.config import * +from cumin.util import * +from mint.plumage.main import Plumage +from parsley.loggingex import PipeLogThread +from psycopg2 import OperationalError +from cumin.admin import SchemaVersion, SchemaMissing +from cumin.errors import CuminErrors + +def restore_IO(): + sys.stderr = sys.__stderr__ + sys.stdout = sys.__stdout__ + +def process_classes(mint, values, section_name, on_empty=None): + + return_code = CuminErrors.NO_ERROR + pkgs = set() + if values and len(values) > 0: + + for cls_str in values.split(","): + pair = cls_str.strip().split(":") + if len(pair) == 2: + pname = pair[0] + cname = pair[1] + else: + log.error("Configuration section '%s',"\ + " class name '%s' is badly formed"\ + % (section_name, cls_str.strip())) + return_code = CuminErrors.PARSE_ERROR + break + + try: + pkg = mint.model._packages_by_name[pname] + if cname == "*": + pkgs = pkgs.union(set(pkg._classes)) + else: + try: + cls = pkg._classes_by_name[cname] + pkgs.add(cls) + except KeyError: + log.warning("Configuration section '%s',"\ + " class '%s' is not contained in package"\ + " '%s'" % (section_name, cname, pname)) + except KeyError: + log.warning("Configuration section '%s',"\ + " package '%s' not found" % (section_name, pname)) + elif on_empty == "all": + for pkg in mint.model._packages: + pkgs = pkgs.union(set(pkg._classes)) + + return return_code, pkgs + +def adjust_return(passed_init, ret): + # Shift non-zer0 return codes left 1 bit + # and OR in whether or not init passed + if ret != 0: + ret = ret << 1 | passed_init + return ret + +def main(): + passed_init = 1 + return_code = CuminErrors.NO_ERROR + + # Do our own simple option check so we can redirect IO early + # without worrying about other options or the behavior of optParse + opts = check_for_options(["--section", "--daemon"], sys.argv[1:]) + if type(opts["--section"]) is str: + section_name = opts["--section"] + else: + section_name = "report" + + # If the --daemon option has been set, redirect stderr and stdio through + # pipes. Launch a thread to read those pipes and direct the content to + # files with rollover control. + pipeThread = None + if opts["--daemon"]: + try: + cumin_home = os.environ.get("CUMIN_HOME", os.path.normpath("/usr/share/cumin")) + err_file = os.path.join(cumin_home, "log", section_name+".stderr") + out_file = os.path.join(cumin_home, "log", section_name+".stdout") + + # Open pipes and redirect + err_r, err_w = os.pipe() + out_r, out_w = os.pipe() + pipeThread = PipeLogThread(descriptors=[err_r, out_r], + paths=[err_file, out_file], + call_on_fail=restore_IO) + pipeThread.start() + sys.stderr = os.fdopen(err_w, "w", 0) + sys.stdout = os.fdopen(out_w, "w", 0) + except: + print_exc() + pipeThread = None + + class ArgError(Exception): + pass + + try: + mint = None + setup_initial_logging() + + parser = PlumageOptionParser() + + # Add additional parameters for report + parser.add_option("--print-stats", action="store_true", default=False) + parser.add_option("--print-events", type="int", default=0, metavar="LEVEL") + parser.add_option("--section", default="report") + parser.add_option("--daemon", action="store_true", default=False) + parser.add_option("--no-vacuum", dest="vacuum_enabled", action="store_false") + parser.add_option("--no-expire", dest="expire_enabled", action="store_false") + + # Get options + opts, args = parser.parse_args() + + # --section controls which section is read from the config file + # If a section other than "report" is specified, require it to exist + config = CuminReportConfig(opts.section, strict_section = opts.section != "report") + + # There might be other sections returned from config.parse() but + # currently we don't care about them... + values = getattr(config.parse(), opts.section) + + # Use the config values as defaults for unspecified options + apply_defaults(values, opts) + + setup_operational_logging(opts, + values.log_max_mb, + values.log_max_archives) + + if len(args) != 0: + log.error("Extra arguments:" + "".join([" "+arg for arg in args])) + raise ArgError + + model_dir = [os.path.join(config.home, x) for x in ("model/admin", "model/plumage")] + + broker_uris = [] +# broker_uris = [x.strip() for x in opts.brokers.split(",")] + + mint = Plumage(model_dir, broker_uris, opts.database) + + mint.print_event_level = opts.print_events + + mint.expire_enabled = opts.expire_enabled + mint.expire_thread.interval = values.expire_interval + mint.expire_thread.threshold = values.expire_threshold + + mint.vacuum_enabled = opts.vacuum_enabled + mint.vacuum_thread.interval = values.vacuum_interval + + mint.check() + mint.init() + + # Handle QMF class binding options + if values.include_classes or values.exclude_classes: + return_code, includes = process_classes(mint, + values.include_classes, + opts.section, on_empty="all") + + return_code, excludes = process_classes(mint, + values.exclude_classes, + opts.section) + + mint.qmf_classes = includes.difference(excludes) + + + # If init_only was set or we failed init, don't proceed... + if not opts.init_only and not return_code: + passed_init = 0 + + mint.start() + stats = mint.update_thread.stats + count = 0 + + if opts.print_stats: + print "[Starred columns are the number of events per second]" + + while True: + if count % 20 == 0: + stats.print_headings() + + count += 1 + + stats.print_values() + + sleep(5) + else: + while True: + sleep(86400) + + except KeyboardInterrupt: + log.info("Received shutdown signal") + + except SystemExit: + if "--help" not in sys.argv: + log.error("Error in options") + return_code = CuminErrors.PARSE_ERROR + + except ArgError: + return_code = CuminErrors.PARSE_ERROR + + except OperationalError: + # Failed to talk to the database on check() + log.info("Run 'cumin-database check' as root for more information.") + return_code = CuminErrors.DATABASE_ERROR + + except SchemaMissing: + log.info("Run 'cumin-admin create-schema' as root") + return_code = CuminErrors.SCHEMA_ERROR + + except SchemaVersion: + log.info("Run 'cumin-admin upgrade-schema' as root") + return_code = CuminErrors.SCHEMA_VER_ERROR + + except: + print_exc() + return_code = CuminErrors.UNHANDLED_ERROR + + if mint: + mint.stop() + if pipeThread: + pipeThread.stop() + log.info("about to call logging shutdown") + logging.shutdown() + return adjust_return(passed_init, return_code) + +def make_ctrl_c(sig, frame): + raise KeyboardInterrupt + +if __name__ == "__main__": + try: + import signal + signal.signal(signal.SIGTERM, make_ctrl_c) + sys.exit(main()) + except KeyboardInterrupt: + sys.exit(0)
Property changes on: branches/play_plumage/cumin/bin/cumin-report ___________________________________________________________________ Added: svn:executable + *
Modified: branches/play_plumage/cumin/bin/cumin-web =================================================================== --- branches/play_plumage/cumin/bin/cumin-web 2012-03-30 17:17:00 UTC (rev 5279) +++ branches/play_plumage/cumin/bin/cumin-web 2012-03-30 20:27:05 UTC (rev 5280) @@ -95,7 +95,7 @@
setup_initial_logging()
- parser = CuminOptionParser() + parser = BrokeredOptionParser()
# Add additional parameters for web parser.add_option("--host")
Modified: branches/play_plumage/cumin/etc/module.profile =================================================================== --- branches/play_plumage/cumin/etc/module.profile 2012-03-30 17:17:00 UTC (rev 5279) +++ branches/play_plumage/cumin/etc/module.profile 2012-03-30 20:27:05 UTC (rev 5280) @@ -1,3 +1,3 @@ export PYTHONPATH="${PWD}/python:${PYTHONPATH}" -export PATH="${PWD}/bin:${PATH}" +export PATH="${PWD}/bin:${PWD}/bin/test:${PATH}" export CUMIN_HOME="${PWD}/instance"
Added: branches/play_plumage/cumin/model/admin/cumin.xml =================================================================== --- branches/play_plumage/cumin/model/admin/cumin.xml (rev 0) +++ branches/play_plumage/cumin/model/admin/cumin.xml 2012-03-30 20:27:05 UTC (rev 5280) @@ -0,0 +1,21 @@ +<model> + <package name="com.redhat.cumin"> + <class name="Info"> + <property name="schema_version" type="sstr"/> + </class> + + <class name="User"> + <property name="name" type="sstr" index="y"/> + <property name="password" type="sstr"/> + </class> + + <class name="Role"> + <property name="name" type="sstr" index="y"/> + </class> + + <class name="UserRoleMapping"> + <property name="user" references="User" index="y"/> + <property name="role" references="Role" index="y"/> + </class> + </package> +</model>
Added: branches/play_plumage/cumin/model/admin/rosemary.xml =================================================================== --- branches/play_plumage/cumin/model/admin/rosemary.xml (rev 0) +++ branches/play_plumage/cumin/model/admin/rosemary.xml 2012-03-30 20:27:05 UTC (rev 5280) @@ -0,0 +1,7 @@ +<model> + <package name="com.redhat.cumin" persistent="y"> + <class name="User"> + <property name="name" unique="y"/> + </class> + </package> +</model>
Modified: branches/play_plumage/cumin/model/cumin.xml =================================================================== --- branches/play_plumage/cumin/model/cumin.xml 2012-03-30 17:17:00 UTC (rev 5279) +++ branches/play_plumage/cumin/model/cumin.xml 2012-03-30 20:27:05 UTC (rev 5280) @@ -1,5 +1,5 @@ <model> - <package name="com.redhat.cumin"> + <package name="com.redhat.cumin.messaging"> <class name="BrokerGroup"> <property name="name" type="sstr"/> <property name="description" type="lstr" optional="y"/> @@ -9,24 +9,6 @@ <property name="broker" type="objId" references="org.apache.qpid.broker:Broker"/> <property name="group" type="objId" references="BrokerGroup"/> </class> - - <class name="Info"> - <property name="schema_version" type="sstr"/> - </class> - - <class name="User"> - <property name="name" type="sstr" index="y"/> - <property name="password" type="sstr"/> - </class> - - <class name="Role"> - <property name="name" type="sstr" index="y"/> - </class> - - <class name="UserRoleMapping"> - <property name="user" references="User" index="y"/> - <property name="role" references="Role" index="y"/> - </class> </package>
<package name="com.redhat.cumin.grid">
Added: branches/play_plumage/cumin/model/plumage/plumage.xml =================================================================== --- branches/play_plumage/cumin/model/plumage/plumage.xml (rev 0) +++ branches/play_plumage/cumin/model/plumage/plumage.xml 2012-03-30 20:27:05 UTC (rev 5280) @@ -0,0 +1,11 @@ +<?xml version="1.0" encoding="utf-8"?> +<schema package="com.redhat.grid.plumage"> + <class name="OSUtil" qmf_headers="False"> + <property name="uuid" index="y" type="uuid" access="RC" desc="UUID of System Image"/> + + <property name="osName" type="sstr" access="RO" desc="Operating System Name" /> + + <statistic name="avail" type="uint32" unit="uint32" /> + <statistic name="used" type="uint32" unit="uint32" /> + </class> +</schema>
Added: branches/play_plumage/cumin/model/plumage/rosemary.xml =================================================================== --- branches/play_plumage/cumin/model/plumage/rosemary.xml (rev 0) +++ branches/play_plumage/cumin/model/plumage/rosemary.xml 2012-03-30 20:27:05 UTC (rev 5280) @@ -0,0 +1,18 @@ +<model> + <package name="com.redhat.grid.plumage"> + <class name="OSUtil"> + <property name="osName"> + <title>OS</title> + </property> + + <statistic name="used"> + <title>Used</title> + </statistic> + + <statistic name="avail"> + <title>Available</title> + </statistic> + </class> + </package> + +</model>
Deleted: branches/play_plumage/cumin/model/plumage.xml =================================================================== --- branches/play_plumage/cumin/model/plumage.xml 2012-03-30 17:17:00 UTC (rev 5279) +++ branches/play_plumage/cumin/model/plumage.xml 2012-03-30 20:27:05 UTC (rev 5280) @@ -1,11 +0,0 @@ -<?xml version="1.0" encoding="utf-8"?> -<schema package="com.redhat.grid.plumage"> - <class name="OSUtil" qmf_headers="False"> - <property name="uuid" index="y" type="uuid" access="RC" desc="UUID of System Image"/> - - <property name="osName" type="sstr" access="RO" desc="Operating System Name" /> - - <statistic name="avail" type="uint32" unit="uint32" /> - <statistic name="used" type="uint32" unit="uint32" /> - </class> -</schema>
Modified: branches/play_plumage/cumin/model/rosemary.xml =================================================================== --- branches/play_plumage/cumin/model/rosemary.xml 2012-03-30 17:17:00 UTC (rev 5279) +++ branches/play_plumage/cumin/model/rosemary.xml 2012-03-30 20:27:05 UTC (rev 5280) @@ -1,10 +1,4 @@ <model> - <package name="com.redhat.cumin" persistent="y"> - <class name="User"> - <property name="name" unique="y"/> - </class> - </package> - <package name="com.redhat.cumin.grid"> <class name="JobSummary"> <property name="GlobalJobId"> @@ -204,7 +198,7 @@ </class> </package>
- <package name="com.redhat.cumin"> + <package name="com.redhat.cumin.messaging"> <class name="BrokerGroup"> <title>Broker group</title>
@@ -547,20 +541,4 @@ </class> </package>
- <package name="com.redhat.grid.plumage"> - <class name="OSUtil"> - <property name="osName"> - <title>OS</title> - </property> - - <statistic name="used"> - <title>Used</title> - </statistic> - - <statistic name="avail"> - <title>Available</title> - </statistic> - </class> - </package> - </model>
Modified: branches/play_plumage/cumin/python/cumin/config.py =================================================================== --- branches/play_plumage/cumin/python/cumin/config.py 2012-03-30 17:17:00 UTC (rev 5279) +++ branches/play_plumage/cumin/python/cumin/config.py 2012-03-30 20:27:05 UTC (rev 5280) @@ -16,90 +16,7 @@
if not os.path.isdir(self.home): raise Exception("Home path '%s' is not a directory") - - test = CuminConfigSection(self, "test") - test.log_file.default = os.path.join(self.home, "log", "test.log") - - def create_web_section(self, name, strict_section): - web = CuminConfigSection(self, name, strict_section) - web.log_file.default = os.path.join(self.home, "log", name + ".log") - - param = ConfigParameter(web, "update-interval", int) - param.default = 10 - - param = ConfigParameter(web, "max-qmf-table-sort", int) - param.default = 1000 - - param = ConfigParameter(web, "host", str) - param.default = "localhost" - - param = ConfigParameter(web, "port", int) - param.default = 45672 - - param = ConfigParameter(web, "operator-email", str) - - param = ConfigParameter(web, "user", str) - - param = ConfigParameter(web, "request-memory", int) - param.default = 512 # MB - param = ConfigParameter(web, "request-memory-vm", int) - param.default = 512 # MB - param = ConfigParameter(web, "request-disk", int) - param.default = 1024 #MB - param = ConfigParameter(web, "request-disk-vm", int) - param.default = 5 * 1024 #MB - - param = ConfigParameter(web, "persona", str) - param.default = "grid"
- param = ConfigParameter(web, "fast-view-attributes", str) - param.default = "JobStatus,Cmd,Args,ExitStatus,JobStartDate,LastRemoteHost,LastJobStatus,Owner" - - param = ConfigParameter(web, "notification-timeout", int) - param.default = 180 - - # Hidden parameter used to force html doctype rather than xhtml - # This is hopefully a temporary workaround so that selenium can - # be used to do some automated testing against cumin - param = ConfigParameter(web, "force-html-doctype", bool) - param.default = False - - def create_data_section(self, name, strict_section): - data = CuminConfigSection(self, name, strict_section) - data.log_file.default = os.path.join(self.home, "log", name + ".log") - - param = ConfigParameter(data, "include-classes", str) - param = ConfigParameter(data, "exclude-classes", str) - param = ConfigParameter(data, "agents", str) - - param = ConfigParameter(data, "expire-enabled", bool) - param.default = True - - param = ConfigParameter(data, "expire-interval", int) - param.default = 60 * 60 # 1 hour - - param = ConfigParameter(data, "expire-threshold", int) - param.default = 24 * 60 * 60 # 1 day - - param = ConfigParameter(data, "vacuum-enabled", bool) - param.default = True - - param = ConfigParameter(data, "vacuum-interval", int) - param.default = 60 * 60 # 1 hour - - def create_master_section(self, name, strict_section): - ''' - Creates a blank master section. - CuminConfigSection values are not interesting to the master - ''' - - master = ConfigSection(self, name, strict_section) - param = ConfigParameter(master, "datas", str) - param.default = "data" - - param = ConfigParameter(master, "webs", str) - param.default = "web" - def get_home(self): return self.home
@@ -130,20 +47,59 @@ def __init__(self, section_name="web", strict_section="False"): super(CuminWebConfig, self).__init__()
- self.create_web_section(section_name, strict_section) + web = CuminWebConfigSection(self, section_name, strict_section) + web.log_file.default = os.path.join(self.home, "log", + section_name + ".log") + +def _common_data_section(self, data, name): + data.log_file.default = os.path.join(self.home, "log", name + ".log") + + param = ConfigParameter(data, "include-classes", str) + param = ConfigParameter(data, "exclude-classes", str) + + param = ConfigParameter(data, "expire-enabled", bool) + param.default = True + + param = ConfigParameter(data, "expire-interval", int) + param.default = 60 * 60 # 1 hour + + param = ConfigParameter(data, "expire-threshold", int) + param.default = 24 * 60 * 60 # 1 day + + param = ConfigParameter(data, "vacuum-enabled", bool) + param.default = True + + param = ConfigParameter(data, "vacuum-interval", int) + param.default = 60 * 60 # 1 hour
class CuminDataConfig(CuminConfig): def __init__(self, section_name="data", strict_section=False): super(CuminDataConfig, self).__init__()
- self.create_data_section(section_name, strict_section) + data = CuminDataConfigSection(self, section_name, strict_section) + _common_data_section(self, data, section_name)
+class CuminReportConfig(CuminConfig): + def __init__(self, section_name="report", strict_section=False): + super(CuminReportConfig, self).__init__() + + data = CuminDataConfigSection(self, section_name, strict_section) + _common_data_section(self, data, section_name) + class CuminMasterConfig(CuminConfig): def __init__(self, section_name="master", strict_section=False): super(CuminMasterConfig, self).__init__()
- self.create_master_section(section_name, strict_section) + master = ConfigSection(self, section_name, strict_section) + param = ConfigParameter(master, "datas", str) + param.default = "data"
+ param = ConfigParameter(master, "webs", str) + param.default = "web" + + param = ConfigParameter(master, "reports", str) + param.default = "" + class CuminConfigSection(ConfigSection): def __init__(self, config, name, strict_section=False): super(CuminConfigSection, self).__init__(config, name, strict_section) @@ -151,16 +107,42 @@ param = ConfigParameter(self, "database", str) param.default = "dbname=cumin user=cumin host=localhost"
+ # Put this here, because authentication is something that + # might need to be done commonly + param = ConfigParameter(self, "auth", str) + param.default = 'internal' + + self.log_file = ConfigParameter(self, "log-file", str) + + param = ConfigParameter(self, "log-level", str) + param.default = "info" + + param = ConfigParameter(self, "log-max-mb", float) + param.default = 10 + + param = ConfigParameter(self, "log-max-archives", int) + param.default = 1 + + param = ConfigParameter(self, "debug", bool) + param.default = False + +class BrokeredConfigSection(CuminConfigSection): + def __init__(self, config, name, strict_section=False): + super(BrokeredConfigSection, self).__init__(config, name, + strict_section) + param = ConfigParameter(self, "brokers", str) param.default = "amqp://localhost"
- param = ConfigParameter(self, "auth", str) - param.default = 'internal' - # Leave default set to None, which is equivalent to # previous behavior param = ConfigParameter(self, "sasl-mech-list", str)
+class CuminWebConfigSection(BrokeredConfigSection): + def __init__(self, config, name, strict_section=False): + super(CuminWebConfigSection, self).__init__(config, name, + strict_section) + param = ConfigParameter(self, "wallaby-broker", str) param.default = ""
@@ -190,20 +172,60 @@ param = ConfigParameter(self, "aviary-suds-logs", bool) param.default = False
- self.log_file = ConfigParameter(self, "log-file", str) + param = ConfigParameter(self, "update-interval", int) + param.default = 10
- param = ConfigParameter(self, "log-level", str) - param.default = "info" + param = ConfigParameter(self, "max-qmf-table-sort", int) + param.default = 1000
- param = ConfigParameter(self, "log-max-mb", float) - param.default = 10 + param = ConfigParameter(self, "host", str) + param.default = "localhost"
- param = ConfigParameter(self, "log-max-archives", int) - param.default = 1 + param = ConfigParameter(self, "port", int) + param.default = 45672
- param = ConfigParameter(self, "debug", bool) + param = ConfigParameter(self, "operator-email", str) + + param = ConfigParameter(self, "user", str) + + param = ConfigParameter(self, "request-memory", int) + param.default = 512 # MB + param = ConfigParameter(self, "request-memory-vm", int) + param.default = 512 # MB + param = ConfigParameter(self, "request-disk", int) + param.default = 1024 #MB + param = ConfigParameter(self, "request-disk-vm", int) + param.default = 5 * 1024 #MB + + param = ConfigParameter(self, "persona", str) + param.default = "grid" + + param = ConfigParameter(self, "fast-view-attributes", str) + param.default = "JobStatus,Cmd,Args,ExitStatus,JobStartDate,LastRemoteHost,LastJobStatus,Owner" + + param = ConfigParameter(self, "notification-timeout", int) + param.default = 180 + + # Hidden parameter used to force html doctype rather than xhtml + # This is hopefully a temporary workaround so that selenium can + # be used to do some automated testing against cumin + param = ConfigParameter(self, "force-html-doctype", bool) param.default = False
+class CuminDataConfigSection(BrokeredConfigSection): + def __init__(self, config, name, strict_section=False): + super(CuminDataConfigSection, self).__init__(config, name, + strict_section) + param = ConfigParameter(self, "agents", str) + +class CuminReportConfigSection(CuminConfigSection): + def __init__(self, config, name, strict_section=False): + super(CuminReportConfigSection, self).__init__(config, name, + strict_section) + + param = ConfigParameter(self, "plumage_server", str) + param.default = "localhost" + class CuminOptionParser(OptionParser,object): def __init__(self): OptionParser.__init__(self) @@ -214,10 +236,25 @@ # section named in --section, via apply_defaults() self.add_option("--debug", action="store_true") self.add_option("--database") - self.add_option("--brokers") self.add_option("--log-file") self.add_option("--log-level")
+class BrokeredOptionParser(CuminOptionParser): + def __init__(self): + CuminOptionParser.__init__(self) + + # Defaults for these come later from the config + # section named in --section, via apply_defaults() + self.add_option("--brokers") + +class PlumageOptionParser(CuminOptionParser): + def __init__(self): + CuminOptionParser.__init__(self) + + # Defaults for these come later from the config + # section named in --section, via apply_defaults() + self.add_option("--server") + def apply_defaults(values, opts): for k, v in opts.__dict__.iteritems(): if v == None and k in values:
Modified: branches/play_plumage/cumin/python/cumin/main.py =================================================================== --- branches/play_plumage/cumin/python/cumin/main.py 2012-03-30 17:17:00 UTC (rev 5279) +++ branches/play_plumage/cumin/python/cumin/main.py 2012-03-30 20:27:05 UTC (rev 5280) @@ -39,7 +39,7 @@ self.home = home self.authmech = authmech
- model_dir = os.path.join(self.home, "model") + model_dir = [os.path.join(self.home, x) for x in ("model/admin", "model")]
self.model = CuminModel(self, model_dir) self.session = CuminSession(self, broker_uris)
Modified: branches/play_plumage/cumin/python/cumin/messaging/brokergroup.py =================================================================== --- branches/play_plumage/cumin/python/cumin/messaging/brokergroup.py 2012-03-30 17:17:00 UTC (rev 5279) +++ branches/play_plumage/cumin/python/cumin/messaging/brokergroup.py 2012-03-30 20:27:05 UTC (rev 5280) @@ -18,7 +18,7 @@
class BrokerGroupSelector(ObjectSelector): def __init__(self, app, name): - cls = app.model.com_redhat_cumin.BrokerGroup + cls = app.model.com_redhat_cumin_messaging.BrokerGroup
super(BrokerGroupSelector, self).__init__(app, name, cls)
@@ -54,7 +54,7 @@
class BrokerGroupFrame(ObjectFrame): def __init__(self, app, name): - cls = app.model.com_redhat_cumin.BrokerGroup + cls = app.model.com_redhat_cumin_messaging.BrokerGroup
super(BrokerGroupFrame, self).__init__(app, name, cls)
@@ -106,7 +106,7 @@ try: cursor = conn.cursor()
- cls = self.app.model.com_redhat_cumin.BrokerGroup + cls = self.app.model.com_redhat_cumin_messaging.BrokerGroup group = cls.create_object(cursor)
group.name = name
Modified: branches/play_plumage/cumin/python/cumin/model.py =================================================================== --- branches/play_plumage/cumin/python/cumin/model.py 2012-03-30 17:17:00 UTC (rev 5279) +++ branches/play_plumage/cumin/python/cumin/model.py 2012-03-30 20:27:05 UTC (rev 5280) @@ -40,9 +40,12 @@ def check(self): log.info("Checking %s", self)
- assert os.path.isdir(self.model_dir) + if not type(self.model_dir) in (list, tuple): + self.model_dir = [self.model_dir]
- log.debug("Model dir exists at '%s'", self.model_dir) + for dirs in self.model_dir: + assert os.path.isdir(dirs) + log.debug("Model dir exists at '%s'", dirs)
def init(self): log.info("Initializing %s", self)
Modified: branches/play_plumage/cumin/python/cumin/parameters.py =================================================================== --- branches/play_plumage/cumin/python/cumin/parameters.py 2012-03-30 17:17:00 UTC (rev 5279) +++ branches/play_plumage/cumin/python/cumin/parameters.py 2012-03-30 20:27:05 UTC (rev 5280) @@ -82,7 +82,7 @@ class NewBrokerGroupParameter(Parameter): def do_unmarshal(self, string): id = int(string) - cls = self.app.model.com_redhat_cumin.BrokerGroup + cls = self.app.model.com_redhat_cumin_messaging.BrokerGroup
cursor = self.app.database.get_read_cursor()
Modified: branches/play_plumage/mint/python/mint/model.py =================================================================== --- branches/play_plumage/mint/python/mint/model.py 2012-03-30 17:17:00 UTC (rev 5279) +++ branches/play_plumage/mint/python/mint/model.py 2012-03-30 20:27:05 UTC (rev 5280) @@ -20,11 +20,14 @@
def check(self): log.info("Checking %s", self) + + if not type(self.model_dir) in (tuple, list): + self.model_dir = [self.model_dir]
- assert os.path.isdir(self.model_dir) + for dirs in self.model_dir: + assert os.path.isdir(dirs) + log.debug("Model dir exists at '%s'", dirs)
- log.debug("Model dir exists at '%s'", self.model_dir) - def init(self): log.info("Initializing %s", self)
Added: branches/play_plumage/mint/python/mint/plumage/__init__.py =================================================================== Added: branches/play_plumage/mint/python/mint/plumage/main.py =================================================================== --- branches/play_plumage/mint/python/mint/plumage/main.py (rev 0) +++ branches/play_plumage/mint/python/mint/plumage/main.py 2012-03-30 20:27:05 UTC (rev 5280) @@ -0,0 +1,85 @@ +from mint.database import MintDatabase +from mint.expire import ExpireThread +from mint.model import MintModel +from mint.vacuum import VacuumThread +from session import PlumageSession +from update import UpdateThread +from cumin.admin import CuminAdmin +#from util import * + +import logging + +log = logging.getLogger("mint.plumage") + +class Plumage(object): + def __init__(self, model_dir, server_url, database_dsn): + + # Can we just use a different model dir here? + # That would control visible packages, vacuuming, etc. + self.model = MintModel(self, model_dir) + self.model.sql_logging_enabled = False + + self.database = MintDatabase(self, database_dsn) + self.admin = CuminAdmin(self) + self.update_thread = UpdateThread(self) + self.session = PlumageSession(self, server_url) + + self.expire_enabled = False + self.expire_thread = ExpireThread(self) + + self.vacuum_enabled = False + self.vacuum_thread = VacuumThread(self) + + self.print_event_level = 0 + + self.packages = [] + + def check(self): + log.info("Checking %s", self) + + self.model.check() + self.database.check() + + def init(self): + log.info("Initializing %s", self) + + def state(cond): + return cond and "enabled" or "disabled" + + log.info("Expiration is %s", state(self.expire_enabled)) + log.info("Vacuum is %s", state(self.vacuum_enabled)) + + self.model.init() + self.database.init() + + self.update_thread.init() + self.session.init() + #self.expire_thread.init() + #self.vacuum_thread.init() + + def start(self): + log.info("Starting %s", self) + + self.update_thread.start() + self.session.start() + + #if self.expire_enabled: + # self.expire_thread.start() + + #if self.vacuum_enabled: + # self.vacuum_thread.start() + + def stop(self): + log.info("Stopping %s", self) + + self.update_thread.stop() + self.session.stop() + + #if self.expire_enabled: + # self.expire_thread.stop() + + #if self.vacuum_enabled: + # self.vacuum_thread.stop() + + def __repr__(self): + return self.__class__.__name__
Added: branches/play_plumage/mint/python/mint/plumage/session.py =================================================================== --- branches/play_plumage/mint/python/mint/plumage/session.py (rev 0) +++ branches/play_plumage/mint/python/mint/plumage/session.py 2012-03-30 20:27:05 UTC (rev 5280) @@ -0,0 +1,89 @@ +import logging + +from mint.util import MintDaemonThread +from time import sleep + +log = logging.getLogger("mint.plumage.session") + +class PlumageSession(object): + def __init__(self, app, server_url): + self.app = app + self.server_url = server_url + + def check(self): + log.info("Checking %s", self) + + def init(self): + log.info("Initializing %s", self) + + def start(self): + log.info("Starting %s", self) + + def stop(self): + log.info("Stopping %s", self) + + def __repr__(self): + return "%s(%s)" % (self.__class__.__name__, self.server_url) + +class PlumageSessionThread(MintDaemonThread): + def __init__(self, app): + super(PumageSessionThread, self).__init__(app) + + def run(self): + while True: + if self.stop_requested: + break + + # We create objects here + sleep(5) + +''' +class MintConsole(Console): + def __init__(self, model): + self.model = model + + def brokerConnected(self, qmf_broker): + message = "Broker %s:%i is connected" + self.model.print_event(1, message, qmf_broker.host, qmf_broker.port) + + def brokerInfo(self, qmf_broker): + message = "Broker info from %s:%i" + self.model.print_event(1, message, qmf_broker.host, qmf_broker.port) + + def brokerDisconnected(self, qmf_broker): + message = "Broker %s:%i is disconnected" + self.model.print_event(1, message, qmf_broker.host, qmf_broker.port) + + def newAgent(self, qmf_agent): + self.model.print_event(3, "Creating %s", qmf_agent) + + up = AgentUpdate(self.model, qmf_agent) + self.model.app.update_thread.enqueue(up) + + def delAgent(self, qmf_agent): + self.model.print_event(3, "Deleting %s", qmf_agent) + + up = AgentDelete(self.model, qmf_agent) + self.model.app.update_thread.enqueue(up) + + def heartbeat(self, qmf_agent, timestamp): + message = "Heartbeat from %s at %s" + self.model.print_event(5, message, qmf_agent, timestamp) + + up = AgentUpdate(self.model, qmf_agent) + self.model.app.update_thread.enqueue(up) + + def newPackage(self, name): + self.model.print_event(2, "New package %s", name) + + def newClass(self, kind, classKey): + self.model.print_event(2, "New class %s", classKey) + + def objectProps(self, broker, qmf_object): + up = ObjectUpdate(self.model, qmf_object) + self.model.app.update_thread.enqueue(up) + + def objectStats(self, broker, qmf_object): + up = ObjectUpdate(self.model, qmf_object) + self.model.app.update_thread.enqueue(up) +'''
Added: branches/play_plumage/mint/python/mint/plumage/update.py =================================================================== --- branches/play_plumage/mint/python/mint/plumage/update.py (rev 0) +++ branches/play_plumage/mint/python/mint/plumage/update.py 2012-03-30 20:27:05 UTC (rev 5280) @@ -0,0 +1,363 @@ +import copy +import resource +import pickle + +from psycopg2 import IntegrityError, TimestampFromTicks +from psycopg2.extensions import cursor as Cursor +from rosemary.model import * + +from mint.model import * +from mint.util import * + +log = logging.getLogger("mint.plumage.update") + +sample_window_min = 60 +sample_window_max = 60 * 5 + +class UpdateThread(MintDaemonThread): + def __init__(self, app): + super(UpdateThread, self).__init__(app) + + self.updates = ConcurrentQueue(maxsize=1000) + self.stats = UpdateStats(self.app) + + self.conn = None + self.cursor = None + + self.halt_on_error = False + + def _delete_all_objects(self): + def loop_body(cls): + if cls._storage != "none": + if cls.check_persistent(): + log.debug("Skipping persistent class " + str(cls)) + else: + cls.delete_selection(self.cursor) + self.cursor.connection.commit() + + log.debug("Delete all objects by bound packages " + str(self.app.packages)) + for pkg in self.app.packages: + for cls in pkg._classes: + loop_body(cls) + + def init(self): + self.conn = self.app.database.get_connection() + + self.cursor = self.conn.cursor(cursor_factory=UpdateCursor) + self.cursor.stats = self.stats + + def enqueue(self, update): + self.updates.put(update) + + self.stats.enqueued += 1 + + def run(self): + while True: + if self.stop_requested: + break + + try: + update = self.updates.get(True, 1) + except Empty: + continue + + self.stats.dequeued += 1 + + update.process(self) + +class UpdateStats(object): + + def __init__(self, app): + self.enqueued = 0 + self.dequeued = 0 + self.dropped = 0 + + self.objects_created = 0 + self.objects_updated = 0 + self.objects_deleted = 0 + + self.objects_created_by_class = defaultdict(int) + self.objects_updated_by_class = defaultdict(int) + self.objects_deleted_by_class = defaultdict(int) + + self.sql_ops = 0 + self.errors = 0 + + self.time = None + self.cpu = 0 + self.memory = 0 + +class UpdateCursor(Cursor): + def execute(self, sql, args=None): + super(UpdateCursor, self).execute(sql, args) + + self.stats.sql_ops += 1 + +class Update(object): + def __init__(self, model): + self.model = model + + def process(self, thread): + log.debug("Processing %s", self) + try: + self.do_process(thread.cursor, thread.stats, thread.app.packages) + thread.conn.commit() + except UpdateDropped: + log.debug("Update dropped") + + thread.conn.rollback() + + thread.stats.dropped += 1 + except: + log.debug("Update failed", exc_info=True) + + thread.conn.rollback() + + thread.stats.errors += 1 + + if thread.halt_on_error: + raise + + def do_process(self, cursor, stats, packages): + raise Exception("Not implemented") + + def __repr__(self): + return self.__class__.__name__ + +class ObjectUpdate(Update): + def __init__(self, model, plumage_object): + super(ObjectUpdate, self).__init__(model) + + self.plumage_object = plumage_object + + def do_process(self, cursor, stats, bound_classes, bound_packages): + cls = self.get_class() + object_id = self.get_object_id() + + # Okay, we want to do a lookup by object id here. + # If we can't find it, we create it. Agent is + # a mint Agent class. + # We need a lookup by field on machine. + + try: + obj = agent.get_object( cursor, cls, object_id) + except RosemaryNotFound: + obj = self.create_object(cursor, stats, cls) + return + + self.update_object(cursor, stats, obj) + + def maybe_drop_sample(self, obj): + properties = self.qmf_object.getProperties() + statistics = self.qmf_object.getStatistics() + + if not properties and statistics: + # Just stats; do we want it? + # if stats.enqueued - stats.dequeued > 500: + + now = time.time() + update = self.qmf_object.getTimestamps()[0] / 1000000000 + sample = obj._sample_time + + if update < now - sample_window_max: + # The sample is too old + raise UpdateDropped() + + if sample and sample > now - sample_window_min: + # The samples are too fidelitous + raise UpdateDropped() + + def create_object(self, cursor, stats, cls): + obj = cls.create_object(cursor) + + object_columns = list() + sample_columns = list() + + table = cls.sql_table + + object_columns.append(table._id) + + self.process_properties(obj, object_columns, cursor) + self.process_statistics(obj, object_columns, sample_columns) + + statements = list() + + sql = cls.sql_insert_object.emit(object_columns) + statements.append(sql) + + if sample_columns: + sql = cls.sql_samples_insert.emit(sample_columns) + statements.append(sql) + obj._sample_time = time.time() + + sql = "; ".join(statements) + self.execute_sql(cursor, sql, obj.__dict__) + obj._save_time = datetime.now() + self.model.print_event(3, "Created %s", obj) + + stats.objects_created += 1 + #stats.objects_created_by_class[cls] += 1 + + return obj + + def update_object(self, cursor, stats, obj): + object_columns = list() + sample_columns = list() + + self.process_properties(obj, object_columns, cursor) + self.process_statistics(obj, object_columns, sample_columns) + + statements = list() + cls = obj._class + + # force a write if it's been too long, even if the values match + if object_columns: + sql = cls.sql_update_object.emit(object_columns) + statements.append(sql) + + if sample_columns: + sql = cls.sql_samples_insert.emit(sample_columns) + statements.append(sql) + + obj._sample_time = time.time() + + if not statements: + raise UpdateDropped() + + sql = "; ".join(statements) + self.execute_sql(cursor, sql, obj.__dict__) + + self.model.print_event(4, "Updated %s", obj) + + stats.objects_updated += 1 + #stats.objects_updated_by_class[cls] += 1 + + def process_properties(self, obj, columns, cursor): + cls = obj._class + + # getProperties comes from console.py + # list of tuples, first is an instance + # which can be spoofed, second is a value for the property + + for prop, value in self.qmf_object.getProperties(): + try: + col, nvalue = self.process_value(cls, prop, value) + except MappingException, e: + log.debug(e) + continue + + # XXX This optimization will be obsolete when QMF does it + # instead + if nvalue == getattr(obj, col.name): + continue + + setattr(obj, col.name, nvalue) + columns.append(col) + + def process_value(self, cls, prop, value): + try: + col = cls._properties_by_name[prop.name].sql_column + except KeyError: + raise MappingException("Property %s is unknown" % prop) + + if value is not None: + value = transform_value(prop, value) + + return col, value + + def process_statistics(self, obj, update_columns, insert_columns): + build_columns = list() + saw_change = False + + + # getStatistics comes from console.py + # list of tuples, first is an instance + # which can be spoofed, second is a value for the property + + for stat, value in self.qmf_object.getStatistics(): + try: + col = obj._class._statistics_by_name[stat.name].sql_column + except KeyError: + log.debug("Statistic %s is unknown", stat) + + continue + + if value is not None: + value = transform_value(stat, value) + + # Don't write unchanged values + # + # XXX This optimization will be obsolete when QMF does it + # instead + if value != getattr(obj, col.name): + setattr(obj, col.name, value) + update_columns.append(col) + saw_change = True + + # If we do end up seeing a value change, we will + # need to insert an entire row so build it up + build_columns.append(col) + + if saw_change: + insert_columns.extend(build_columns) + + def execute_sql(self, cursor, text, args): + try: + cursor.execute(text, args) + except: + log.debug("%s failed sql execute", self, exc_info=True) + log.error("Sql execute failed") + log.error("Sql text: %s", text) + log.error("Sql values:") + + for item in sorted(args.items()): + log.error(" %-34s %r", *item) + + log.error("Sql row count: %i", cursor.rowcount) + + log.error("object properties:") + + # see process_properties + for item in sorted(self.qmf_object.getProperties()): + log.error(" %-34s %r", *item) + + log.error("object statistics:") + + # see process statistics + for item in sorted(self.qmf_object.getStatistics()): + log.error(" %-34s %r", *item) + + raise + + def __repr__(self): + name = self.__class__.__name__ + return name + + #agent_id = self.get_agent_id() + #cls = self.qmf_object.getClassKey().getClassName() + #obj_id = self.get_object_id() + + #return "%s(%s,%s,%s)" % (name, agent_id, cls, obj_id) + +class MappingException(Exception): + pass + +def transform_default(value): + return value + +def transform_timestamp(value): + if value != 0: + return datetime.fromtimestamp(value / 1000000000) + +def transform_pickle(value): + return pickle.dumps(x) + +transformers = list([transform_default for x in range(32)]) + +transformers[8] = transform_timestamp +transformers[10] = str +transformers[14] = str +transformers[15] = transform_pickle + +def transform_value(attr, value): + return transformers[attr.type](value)
Modified: branches/play_plumage/rosemary/python/rosemary/model.py =================================================================== --- branches/play_plumage/rosemary/python/rosemary/model.py 2012-03-30 17:17:00 UTC (rev 5279) +++ branches/play_plumage/rosemary/python/rosemary/model.py 2012-03-30 20:27:05 UTC (rev 5280) @@ -16,25 +16,29 @@
self.sql_logging_enabled = False
- def load_model_dir(self, path): - assert os.path.isdir(path) + def load_model_dir(self, paths): + # Let this work across multiple directories + if not type(paths) in (list, tuple): + paths = [paths]
- extensions = os.path.join(path, "rosemary.xml") + for path in paths: + assert os.path.isdir(path) + extensions = os.path.join(path, "rosemary.xml")
- for name in os.listdir(path): - file_path = os.path.join(path, name) + for name in os.listdir(path): + file_path = os.path.join(path, name)
- if not os.path.isfile(file_path): - continue + if not os.path.isfile(file_path): + continue
- if file_path == extensions: - continue + if file_path == extensions: + continue
- if file_path.endswith(".xml"): - self.load_model_file(file_path) + if file_path.endswith(".xml"): + self.load_model_file(file_path)
- if os.path.isfile(extensions): - self.load_extensions(extensions) + if os.path.isfile(extensions): + self.load_extensions(extensions)
def load_model_file(self, path): tree = ElementTree()
cumin-developers@lists.fedorahosted.org