2:15 p.m.
Author: tmckay
Date: 2010-12-21 20:15:51 +0000 (Tue, 21 Dec 2010)
New Revision: 4443
Modified:
branches/scale_testing/cumin/bin/cumin-data
branches/scale_testing/cumin/python/cumin/config.py
branches/scale_testing/mint/python/mint/main.py
branches/scale_testing/mint/python/mint/session.py
branches/scale_testing/mint/python/mint/update.py
Log:
Changes to collect stats in CSV files for scale testing.
Modified: branches/scale_testing/cumin/bin/cumin-data
===================================================================
--- branches/scale_testing/cumin/bin/cumin-data 2010-12-20 14:12:38 UTC (rev 4442)
+++ branches/scale_testing/cumin/bin/cumin-data 2010-12-21 20:15:51 UTC (rev 4443)
@@ -16,6 +16,8 @@
parser = CuminOptionParser(values)
parser.add_option("--print-stats", action="store_true")
+ parser.add_option("--scale-stats", action="store_true")
+ parser.add_option("--object-stats", action="store_true")
parser.add_option("--print-events", type="int", default=0,
metavar="LEVEL")
opts, args = parser.parse_args()
@@ -26,7 +28,7 @@
broker_uris = [x.strip() for x in opts.brokers.split(",")]
- mint = Mint(model_dir, broker_uris, opts.database)
+ mint = Mint(model_dir, broker_uris, opts.database, values.queue_size)
mint.print_event_level = opts.print_events
@@ -65,19 +67,40 @@
if opts.print_stats:
print "[Starred columns are the number of events per second]"
+ # set up for scale stats here
+ if opts.scale_stats:
+ stats.init_stats_path(mint, config.home)
+
+ if opts.print_stats or \
+ opts.scale_stats or \
+ opts.object_stats:
while True:
- if count % 20 == 0:
- stats.print_headings()
+ stats.capture()
+ if opts.print_stats:
+ if count % 20 == 0:
+ stats.print_headings()
+ count += 1
+ stats.print_values()
+
+ if opts.object_stats:
+ stats.print_values_by_class(mint)
- count += 1
+ if opts.scale_stats:
+ stats.write_values_by_class(mint, time.localtime())
+ stats.write_values(time.localtime())
- stats.print_values()
-
sleep(5)
else:
while True:
sleep(86400)
+ except Exception as inst:
+ print inst
+
finally:
+ # close the scale stats CSV file here:
+ if opts.scale_stats:
+ pass
+
mint.stop()
if __name__ == "__main__":
Modified: branches/scale_testing/cumin/python/cumin/config.py
===================================================================
--- branches/scale_testing/cumin/python/cumin/config.py 2010-12-20 14:12:38 UTC (rev
4442)
+++ branches/scale_testing/cumin/python/cumin/config.py 2010-12-21 20:15:51 UTC (rev
4443)
@@ -56,6 +56,9 @@
param = ConfigParameter(data, "vacuum-interval", int)
param.default = 60 * 60 # 1 hour
+ param = ConfigParameter(data, "queue-size", int)
+ param.default = 1000
+
def parse(self):
paths = list()
Modified: branches/scale_testing/mint/python/mint/main.py
===================================================================
--- branches/scale_testing/mint/python/mint/main.py 2010-12-20 14:12:38 UTC (rev 4442)
+++ branches/scale_testing/mint/python/mint/main.py 2010-12-21 20:15:51 UTC (rev 4443)
@@ -10,14 +10,14 @@
log = logging.getLogger("mint.main")
class Mint(object):
- def __init__(self, model_dir, broker_uris, database_dsn):
+ def __init__(self, model_dir, broker_uris, database_dsn, queue_max=1000):
self.model = MintModel(self, model_dir)
self.model.sql_logging_enabled = False
self.session = MintSession(self, broker_uris)
self.database = MintDatabase(self, database_dsn)
- self.update_thread = UpdateThread(self)
+ self.update_thread = UpdateThread(self,queue_max)
self.expire_enabled = True
self.expire_thread = ExpireThread(self)
Modified: branches/scale_testing/mint/python/mint/session.py
===================================================================
--- branches/scale_testing/mint/python/mint/session.py 2010-12-20 14:12:38 UTC (rev 4442)
+++ branches/scale_testing/mint/python/mint/session.py 2010-12-21 20:15:51 UTC (rev 4443)
@@ -64,14 +64,17 @@
def brokerConnected(self, qmf_broker):
message = "Broker %s:%i is connected"
self.model.print_event(1, message, qmf_broker.host, qmf_broker.port)
+ self.model.app.update_thread.ignore()
def brokerInfo(self, qmf_broker):
message = "Broker info from %s:%i"
self.model.print_event(1, message, qmf_broker.host, qmf_broker.port)
+ self.model.app.update_thread.ignore()
def brokerDisconnected(self, qmf_broker):
message = "Broker %s:%i is disconnected"
self.model.print_event(1, message, qmf_broker.host, qmf_broker.port)
+ self.model.app.update_thread.ignore()
def newAgent(self, qmf_agent):
self.model.print_event(3, "Creating %s", qmf_agent)
@@ -94,9 +97,11 @@
def newPackage(self, name):
self.model.print_event(2, "New package %s", name)
+ self.model.app.update_thread.ignore()
def newClass(self, kind, classKey):
self.model.print_event(2, "New class %s", classKey)
+ self.model.app.update_thread.ignore()
def objectProps(self, broker, qmf_object):
up = ObjectUpdate(self.model, qmf_object)
@@ -108,6 +113,7 @@
def event(self, broker, event):
self.model.print_event(4, "New event %s from %s", broker, event)
+ self.model.app.update_thread.ignore()
def methodResponse(self, broker, seq, response):
message = "Method response for request %i received from %s"
@@ -121,3 +127,4 @@
callback(response.text, response.outArgs)
finally:
self.model.lock.release()
+ self.model.app.update_thread.method_response()
Modified: branches/scale_testing/mint/python/mint/update.py
===================================================================
--- branches/scale_testing/mint/python/mint/update.py 2010-12-20 14:12:38 UTC (rev 4442)
+++ branches/scale_testing/mint/python/mint/update.py 2010-12-21 20:15:51 UTC (rev 4443)
@@ -1,6 +1,8 @@
import copy
import resource
import pickle
+import os.path
+import csv
from psycopg2 import IntegrityError, TimestampFromTicks
from psycopg2.extensions import cursor as Cursor
@@ -15,10 +17,10 @@
sample_window_max = 60 * 5
class UpdateThread(MintDaemonThread):
- def __init__(self, app):
+ def __init__(self, app, queue_max=1000):
super(UpdateThread, self).__init__(app)
- self.updates = ConcurrentQueue(maxsize=1000)
+ self.updates = ConcurrentQueue(maxsize=queue_max)
self.stats = UpdateStats(self.app)
self.conn = None
@@ -33,24 +35,37 @@
self.cursor.stats = self.stats
def enqueue(self, update):
+ update.enqueue_time = time.time()
+
self.updates.put(update)
-
+
self.stats.enqueued += 1
+ def ignore(self):
+ self.stats.ignore += 1
+
+ def method_response(self):
+ self.stats.method_response +=1
+
def run(self):
- while True:
- if self.stop_requested:
- break
+ try:
+ while True:
+ if self.stop_requested:
+ break
- try:
- update = self.updates.get(True, 1)
- except Empty:
- continue
+ try:
+ update = self.updates.get(True) #, 1)
+ except Empty:
+ continue
+ update.dequeue_time = time.time()
+ self.stats.dequeued += 1
- self.stats.dequeued += 1
+ update.process(self)
+ self.conn.commit()
+
+ except:
+ print "uh oh"
- update.process(self)
-
class UpdateStats(object):
group_names = ("Updates", "Agents", "Objects")
groups = "%43s | %32s | %32s |" % group_names
@@ -60,6 +75,7 @@
"*Created", "*Updated", "*Deleted",
"*Created", "*Updated", "*Deleted",
"*Sql Ops", "Errors", "Cpu (%)", "Mem
(M)")
+
headings_fmt = \
"%10s %10s %10s %10s | " + \
"%10s %10s %10s | " + \
@@ -67,7 +83,6 @@
"%10s %10s %10s %10s"
headings = headings_fmt % heading_names
-
values_fmt = \
"%10i %10.1f %10.1f %10.1f | " + \
"%10.1f %10.1f %10.1f | " + \
@@ -81,6 +96,8 @@
self.enqueued = 0
self.dequeued = 0
self.dropped = 0
+ self.ignore = 0
+ self.method_response = 0
self.agents_created = 0
self.agents_updated = 0
@@ -93,6 +110,7 @@
self.objects_created_by_class = defaultdict(int)
self.objects_updated_by_class = defaultdict(int)
self.objects_deleted_by_class = defaultdict(int)
+ self.objects_dropped_by_class = defaultdict(int)
self.sql_ops = 0
self.errors = 0
@@ -101,11 +119,41 @@
self.cpu = 0
self.memory = 0
+ self.init_update_times()
+
+ def init_update_times(self):
+ # Track average time of update from arrival to
+ # processing complete. Refreshed per interval.
+ self.update_queue_duration = 0
+ self.update_process_duration = 0
+ self.queue_time = 0
+ self.initial_process_time = 0
+ self.commit_time = 0
+ self.record_samples = 0
+ self.init_update_flag = False
+
+ def record_update_times(self,update):
+ try:
+ if self.init_update_flag:
+ self.init_update_times()
+ self.record_samples += 1
+ self.update_queue_duration += \
+ update.start_process - update.create_time
+ self.update_process_duration += \
+ update.finish_process - update.start_process
+ self.queue_time += \
+ update.dequeue_time - update.enqueue_time
+ self.initial_process_time += \
+ update.after_process - update.start_process
+ self.commit_time += \
+ update.after_commit - update.after_process
+ except:
+ print "record update exception"
+
def capture(self):
now = copy.copy(self)
now.time = time.time()
-
rusage = resource.getrusage(resource.RUSAGE_SELF)
now.cpu = rusage[0] + rusage[1]
@@ -113,6 +161,7 @@
UpdateStats.then = UpdateStats.now
UpdateStats.now = now
+ self.init_update_flag = True
def get_resident_pages(self):
try:
@@ -126,12 +175,7 @@
print self.groups
print self.headings
- def print_values(self):
- self.capture()
-
- if not self.then:
- return
-
+ def _gather_values(self):
values = [self.now.enqueued - self.then.enqueued,
self.now.dequeued - self.then.dequeued,
self.now.dropped - self.then.dropped,
@@ -153,23 +197,118 @@
values.append(self.errors)
values.append(int((self.now.cpu - self.then.cpu) / secs * 100))
values.append(self.now.memory / 1000000.0)
+ return values
- print self.values_fmt % tuple(values)
+ def print_values(self):
+# moved capture call to cumin-data so that write_values can
+# be run as well
+# self.capture()
+ if not self.then:
+ return
- def print_values_by_class(self):
- names = ("Class", "Created", "Updated",
"Deleted")
- print "%20s %10s %10s %10s" % names
+ print self.values_fmt % tuple(self._gather_values())
+ def print_values_by_class(self,mint):
+ names = ("Class", "Created", "Updated",
"Deleted", "Dropped")
+ print "%20s %10s %10s %10s %10s" % names
+
for pkg in mint.model._packages:
for cls in pkg._classes:
- created = stats.created_by_class[cls]
- updated = stats.updated_by_class[cls]
- deleted = stats.deleted_by_class[cls]
+ created = self.objects_created_by_class[cls]
+ updated = self.objects_updated_by_class[cls]
+ deleted = self.objects_deleted_by_class[cls]
+ dropped = self.objects_dropped_by_class[cls]
+ if created or updated or deleted or dropped:
+ args = (cls._name, created, updated, deleted, dropped)
+ print "%-20s %10i %10i %10i %10i" % args
+
+ def init_stats_path(self, mint, path):
+ p = os.path.join(path, \
+ "stats_"+
time.strftime("%m%d%y_%H%M%S",time.localtime()))
+
+ if not os.path.exists(p):
+ os.makedirs(p)
+
+ # init the path for general stats
+ self.stats_path = os.path.join(p, "stats.csv")
+
+ # init the paths for stats by class
+ self.class_stats_paths = {}
+ for pkg in mint.model._packages:
+ for cls in pkg._classes:
+ self.class_stats_paths[cls._name] = \
+ os.path.join(p, cls._name+".csv")
+
+ def write_values_by_class(self, mint, localtime_):
+ names = ("Time", "Created", "Updated",
"Deleted", "Dropped")
+
+ for pkg in mint.model._packages:
+ for cls in pkg._classes:
+ created = self.objects_created_by_class[cls]
+ updated = self.objects_updated_by_class[cls]
+ deleted = self.objects_deleted_by_class[cls]
+ dropped = self.objects_dropped_by_class[cls]
+
if created or updated or deleted:
- args = (cls._name, created, updated, deleted)
- print "%-20s %10i %10i %10i" % args
-
+ fname = self.class_stats_paths[cls._name]
+ write_header = not os.path.isfile(fname)
+ db = csv.writer(open(fname, "a+"))
+ if write_header:
+ db.writerow(names)
+ db.writerow((time.strftime("%H:%M:%S",localtime_),
+ created, updated, deleted, dropped))
+
+ def get_update_durations(self):
+ total = self.now.record_samples
+ if total == 0:
+ create_dur = proc_dur = que_dur = init_proc_time = commit_time = 0
+ else:
+ create_dur = self.now.update_queue_duration / total
+ proc_dur = self.now.update_process_duration / total
+ que_dur = self.now.queue_time / total
+ init_proc_time = self.now.initial_process_time / total
+ commit_time = self.now.commit_time / total
+ return (create_dur, proc_dur, que_dur, init_proc_time, commit_time)
+
+
+ def write_values(self, localtime_):
+ stats_headings = \
+ ("Time", "Total Msgs",
+ "Depth", "*Avg Que", "*Avg Create", "*Avg
Proc",
+ "*Init Proc", "*Commit",
+ "*Enqueued", "*Dequeued", "*Dropped",
+ "*Create Agent", "*Update Agent", "*Delete
Agent",
+ "*Create", "*Update", "*Delete",
+ "*Sql Ops", "Errors", "Cpu (%)", "Mem
(M)")
+
+ if not self.then:
+ return
+ try:
+ values = self._gather_values()
+ values.insert(0, time.strftime("%H:%M:%S",localtime_))
+ values.insert(1, self.now.enqueued)
+
+ # include object update times
+ avg_create_dur,avg_process_dur,avg_que_dur, \
+ avg_init_proc,avg_commit = self.get_update_durations()
+
+ values.insert(3, avg_que_dur)
+ values.insert(4, avg_create_dur)
+ values.insert(5, avg_process_dur)
+ values.insert(6, avg_init_proc)
+ values.insert(7, avg_commit)
+
+ write_header = not os.path.isfile(self.stats_path)
+ db = csv.writer(open(self.stats_path, "a+"))
+ if write_header:
+ db.writerow(stats_headings)
+ db.writerow(values)
+
+ except Exception as inst:
+ print "exception in write values"
+ print inst
+
class UpdateCursor(Cursor):
def execute(self, sql, args=None):
super(UpdateCursor, self).execute(sql, args)
@@ -179,35 +318,50 @@
class Update(object):
def __init__(self, model):
self.model = model
+ self.create_time = time.time()
def process(self, thread):
+ self.start_process = time.time()
log.debug("Processing %s", self)
try:
self.do_process(thread.cursor, thread.stats)
-
+ self.after_process = time.time()
thread.conn.commit()
+ self.after_commit = time.time()
except UpdateDropped:
log.debug("Update dropped")
thread.conn.rollback()
thread.stats.dropped += 1
+ try:
+ cls = self.get_class()
+ except:
+ cls = None
+ if cls != None:
+ thread.stats.objects_dropped_by_class[cls] += 1
+ self.after_commit = time.time()
+
except:
log.exception("Update failed")
thread.conn.rollback()
-
thread.stats.errors += 1
- #print_exc()
+ print_exc()
if thread.halt_on_error:
raise
+ self.finish_process = time.time()
+ thread.stats.record_update_times(self)
def do_process(self, cursor, stats):
raise Exception("Not implemented")
+ def get_class(self):
+ return None
+
def __repr__(self):
return self.__class__.__name__
@@ -346,7 +500,7 @@
self.model.print_event(3, "Created %s", obj)
stats.objects_created += 1
- #stats.objects_created_by_class[cls] += 1
+ stats.objects_created_by_class[cls] += 1
return obj
@@ -380,7 +534,9 @@
# force a write if it's been too long, even if the values match
if object_columns \
- or obj._save_time < obj._qmf_update_time - timedelta(hours=1):
+ or (obj._save_time != None and \
+ obj._qmf_update_time != None and \
+ obj._save_time < obj._qmf_update_time - timedelta(hours=1)):
object_columns.append(cls.sql_table._qmf_update_time)
sql = cls.sql_update_object.emit(object_columns)
@@ -405,7 +561,7 @@
self.model.print_event(4, "Updated %s", obj)
stats.objects_updated += 1
- #stats.objects_updated_by_class[cls] += 1
+ stats.objects_updated_by_class[cls] += 1
def delete_object(self, cursor, stats, obj):
obj.delete(cursor)
@@ -413,7 +569,7 @@
self.model.print_event(3, "Deleted %s", obj)
stats.objects_deleted += 1
- #stats.objects_deleted_by_class[obj._class] += 1
+ stats.objects_deleted_by_class[obj._class] += 1
def process_properties(self, obj, columns, cursor):
cls = obj._class
@@ -605,7 +761,7 @@
count = cls.delete_selection(cursor, _qmf_agent_id=agent.id)
stats.objects_deleted += count
- #stats.objects_deleted_by_class[cls] += count
+ stats.objects_deleted_by_class[cls] += count
cursor.connection.commit()