Author: justi9 Date: 2007-12-13 14:24:38 -0500 (Thu, 13 Dec 2007) New Revision: 1480 Modified: mgmt/cumin/python/cumin/broker.py mgmt/cumin/python/cumin/brokergroup.py mgmt/cumin/python/cumin/client.py mgmt/cumin/python/cumin/client.strings mgmt/cumin/python/cumin/exchange.py mgmt/cumin/python/cumin/formats.py mgmt/cumin/python/cumin/queue.py mgmt/mint/python/mint/__init__.py mgmt/mint/python/mint/schema.py mgmt/mint/python/mint/schema.sql mgmt/notes/justin-todo.txt Log: Fixes group filtering in the broker browser. Handles duplicate group-broker mappings correctly. Updates schema. Raises the max length before fmt_shorten shortens strings. Fixes the pending actions placeholder in client.strings. Modified: mgmt/cumin/python/cumin/broker.py =================================================================== --- mgmt/cumin/python/cumin/broker.py 2007-12-13 18:09:05 UTC (rev 1479) +++ mgmt/cumin/python/cumin/broker.py 2007-12-13 19:24:38 UTC (rev 1480) @@ -2,6 +2,7 @@ from wooly import * from wooly.widgets import * from random import random +from psycopg2 import IntegrityError from configproperty import * from virtualhost import * @@ -15,7 +16,7 @@ strings = StringCatalog(__file__) -class BrokerSetForm(SQLObjectItemSet, Form): +class BrokerSetForm(PaginatedItemSet, Form): def __init__(self, app, name): super(BrokerSetForm, self).__init__(app, name) @@ -37,9 +38,6 @@ self.submit = BooleanParameter(app, "submit") self.add_parameter(self.submit) self.add_form_parameter(self.submit) - - self.paginator = Paginator(app, "page") - self.add_child(self.paginator) def get_title(self, session, model): return "Brokers %s" % fmt_count(self.get_item_count(session, model)) @@ -48,10 +46,12 @@ return BrokerRegistration.select().count() def do_get_items(self, session, model): - start, end = self.paginator.get_bounds(session) + start, end = self.get_bounds(session) return BrokerRegistration.select(orderBy="name")[start:end] def do_process(self, session, model): + super(BrokerSetForm, self).do_process(session, model) + if self.submit.get(session): self.submit.set(session, False) @@ -69,12 +69,12 @@ if group: for broker in brokers: - broker.addBrokerGroup(group) + try: + broker.addBrokerGroup(group) + except IntegrityError: + pass self.page().set_redirect_url(session, session.marshal()) - else: - count = self.get_item_count(session, model) - self.paginator.set_count(session, count) def render_action_param_name(self, session, broker): return self.action.path() @@ -464,11 +464,23 @@ profile = self.parent().profile.get(session) cluster = self.parent().cluster.get(session) - brokers = BrokerRegistration.selectBy \ - (profile = profile, cluster = cluster) + brokers = BrokerRegistration.select() - # XXX add group filtering + if profile: + brokers = brokers.filter \ + (BrokerRegistration.q.profileID == profile.id) + if cluster: + brokers = brokers.filter \ + (BrokerRegistration.q.clusterID == cluster.id) + + if group: + brokers = brokers.filter \ + (BrokerGroupMapping.q.brokerRegistrationID \ + == BrokerRegistration.q.id) + brokers = brokers.filter \ + (BrokerGroupMapping.q.brokerGroupID == group.id) + return brokers def render_none(self, session, model): Modified: mgmt/cumin/python/cumin/brokergroup.py =================================================================== --- mgmt/cumin/python/cumin/brokergroup.py 2007-12-13 18:09:05 UTC (rev 1479) +++ mgmt/cumin/python/cumin/brokergroup.py 2007-12-13 19:24:38 UTC (rev 1480) @@ -28,12 +28,12 @@ return fmt_olink(branch, group) def render_item_config(self, session, group): - return len(group.brokers) + return group.brokers.count() def render_item_status(self, session, group): writer = Writer() - for broker in sorted_by(group.brokers): + for broker in group.brokers: writer.write(fmt_ostatus(broker)) return writer.to_string() @@ -87,8 +87,12 @@ class GroupBrokerTab(BrokerSetForm): def get_title(self, session, group): - return "Brokers %s" % fmt_count(len(group.brokers)) + return "Brokers %s" % \ + fmt_count(self.get_item_count(session, group)) + def get_item_count(self, session, group): + return group.brokers.count() + def do_get_items(self, session, group): return group.brokers Modified: mgmt/cumin/python/cumin/client.py =================================================================== --- mgmt/cumin/python/cumin/client.py 2007-12-13 18:09:05 UTC (rev 1479) +++ mgmt/cumin/python/cumin/client.py 2007-12-13 19:24:38 UTC (rev 1480) @@ -35,12 +35,12 @@ if vhost: start, end = self.paginator.get_bounds(session) return Client.select(Client.q.vhostID == vhost.id, - orderBy="ipAddr")[start:end] + orderBy="address")[start:end] def render_item_link(self, session, client): branch = session.branch() self.page().show_client(branch, client).show_view(branch) - return fmt_olink(branch, client, name=client.ipAddr) + return fmt_olink(branch, client, name=client.address) def render_item_sessions(self, session, client): branch = session.branch() @@ -76,7 +76,7 @@ self.set_view_mode(self.view) def get_title(self, session, client): - return "Client %s" % client.ipAddr + return "Client %s" % client.address class ClientStatus(CuminStatus): def render_messages_produced(self, session, client): @@ -114,13 +114,13 @@ return self.tabs.show_mode(session, self.sessions) def get_title(self, session, client): - return "Client '%s'" % client.ipAddr + return "Client '%s'" % client.address def render_data_url(self, session, client): return "client.xml?id=%i" % client.id def render_address(self, session, client): - return client.ipAddr + return client.address def render_auth_id(self, session, client): return "e50e7dcaa8d6a039a" #XXX get rid of this Modified: mgmt/cumin/python/cumin/client.strings =================================================================== --- mgmt/cumin/python/cumin/client.strings 2007-12-13 18:09:05 UTC (rev 1479) +++ mgmt/cumin/python/cumin/client.strings 2007-12-13 19:24:38 UTC (rev 1480) @@ -54,7 +54,7 @@ <!-- <div>{status_info}</div> --> - <div>{pending_actions}</div> + <div>{actions_pending}</div> <table> <tr> Modified: mgmt/cumin/python/cumin/exchange.py =================================================================== --- mgmt/cumin/python/cumin/exchange.py 2007-12-13 18:09:05 UTC (rev 1479) +++ mgmt/cumin/python/cumin/exchange.py 2007-12-13 19:24:38 UTC (rev 1480) @@ -370,12 +370,15 @@ % queue.id class ExchangeProducerSet(ItemSet): - def get_title(self, session, queue): - return "Producers %s" % fmt_count(len(queue.producers)) + def get_title(self, session, exchange): + return "Producers %s" % fmt_count(len(exchange.producers)) - def do_get_items(self, session, queue): - return sorted_by(queue.producers) + def get_item_count(self, session, exchange): + return Producer.select(Producer.q.exchangeID == exchange.id).count() + def do_get_items(self, session, exchange): + return Producer.select(Producer.q.exchangeID == exchange.id) + def render_item_name(self, session, producer): return producer.name Modified: mgmt/cumin/python/cumin/formats.py =================================================================== --- mgmt/cumin/python/cumin/formats.py 2007-12-13 18:09:05 UTC (rev 1479) +++ mgmt/cumin/python/cumin/formats.py 2007-12-13 19:24:38 UTC (rev 1480) @@ -101,8 +101,8 @@ return "<span class=\"none\">None</span>" def fmt_shorten(string): - if len(string) > 15: - return string[:10] + "..." + string[-5:] + if len(string) > 20: + return string[:13] + "..." + string[-6:] else: return string Modified: mgmt/cumin/python/cumin/queue.py =================================================================== --- mgmt/cumin/python/cumin/queue.py 2007-12-13 18:09:05 UTC (rev 1479) +++ mgmt/cumin/python/cumin/queue.py 2007-12-13 19:24:38 UTC (rev 1480) @@ -479,32 +479,39 @@ return "queue.png?id=%i;s=enqueueTxnCount;s=dequeueTxnCount" \ % queue.id -class QueueConsumerSet(ItemSet): +class QueueConsumerSet(PaginatedItemSet): def get_title(self, session, queue): - return "Consumers %s" % fmt_count(len(queue.consumers)) + return "Consumers %s" % fmt_count(self.get_item_count(session, queue)) + def get_item_count(self, session, queue): + return Consumer.select(Consumer.q.queueID == queue.id).count() + def do_get_items(self, session, queue): - return queue.consumers + start, end = self.get_bounds(session) + return Consumer.select(Consumer.q.queueID == queue.id)[start:end] def render_item_name(self, session, consumer): return consumer.name def render_item_messages_consumed(self, session, consumer): - return consumer.mintConsumerStats.msgsConsumed + stat = self.app.model.consumer.get_stat("msgsConsumed") + return stat.value(consumer) def render_item_messages_consumed_rate(self, session, consumer): - value = consumer.mintConsumerStats.msgsConsumed - return fmt_rate(value, "msg", "sec") + stat = self.app.model.consumer.get_stat("msgsConsumed") + return fmt_rate(stat.rate(consumer), "msg", "sec") def render_item_bytes_consumed(self, session, consumer): - return consumer.mintConsumerStats.bytesConsumed + stat = self.app.model.consumer.get_stat("bytesConsumed") + return stat.value(consumer) def render_item_bytes_consumed_rate(self, session, consumer): - value = consumer.mintConsumerStats.bytesConsumed - return fmt_rate(value, "byte", "sec") + stat = self.app.model.consumer.get_stat("bytesConsumed") + return fmt_rate(stat.rate(consumer), "byte", "sec") def render_item_unacked_messages(self, session, consumer): - return consumer.mintConsumerStats.unackedMessages + stat = self.app.model.consumer.get_stat("unackedMessages") + return stat.value(consumer) class QueueXmlPage(CuminXmlPage): def __init__(self, app, name): Modified: mgmt/mint/python/mint/__init__.py =================================================================== --- mgmt/mint/python/mint/__init__.py 2007-12-13 18:09:05 UTC (rev 1479) +++ mgmt/mint/python/mint/__init__.py 2007-12-13 19:24:38 UTC (rev 1480) @@ -25,14 +25,24 @@ host = StringCol(length=1000, default=None) port = SmallIntCol(default=None) broker = ForeignKey("Broker", cascade="null", default=None) - groups = RelatedJoin("BrokerGroup") + groups = SQLRelatedJoin("BrokerGroup", + intermediateTable="broker_group_mapping", + createRelatedTable=False) cluster = ForeignKey("BrokerCluster", cascade="null", default=None) profile = ForeignKey("BrokerProfile", cascade="null", default=None) class BrokerGroup(SQLObject): name = StringCol(length=1000, default=None) - brokers = RelatedJoin("BrokerRegistration") + brokers = SQLRelatedJoin("BrokerRegistration", + intermediateTable="broker_group_mapping", + createRelatedTable=False) +class BrokerGroupMapping(SQLObject): + brokerRegistration = ForeignKey("BrokerRegistration", notNull=True, + cascade=True) + brokerGroup = ForeignKey("BrokerGroup", notNull=True, cascade=True) + unique = index.DatabaseIndex(brokerRegistration, brokerGroup, unique=True) + class BrokerCluster(SQLObject): name = StringCol(length=1000, default=None) brokers = MultipleJoin("BrokerRegistration", joinColumn="cluster_id") Modified: mgmt/mint/python/mint/schema.py =================================================================== --- mgmt/mint/python/mint/schema.py 2007-12-13 18:09:05 UTC (rev 1479) +++ mgmt/mint/python/mint/schema.py 2007-12-13 19:24:38 UTC (rev 1480) @@ -62,13 +62,6 @@ model.getConnectedBroker(managedBrokerLabel).method(methodId, self.idOriginal, \ classToSchemaNameMap[self.__class__.__name__], "echo", args=actualArgs, packageName="qpid") - def crash(self, model, managedBrokerLabel, callbackMethod): - """Temporary test method to crash the broker""" - actualArgs = dict() - methodId = model.registerCallback(callbackMethod) - model.getConnectedBroker(managedBrokerLabel).method(methodId, self.idOriginal, \ - classToSchemaNameMap[self.__class__.__name__], "crash", args=actualArgs, packageName="qpid") - System.sqlmeta.addJoin(MultipleJoin('Broker', joinMethodName='brokers')) @@ -115,7 +108,6 @@ durable = BoolCol(default=None) autoDelete = BoolCol(default=None) exclusive = BoolCol(default=None) - pageMemoryLimit = IntCol(default=None) def purge(self, model, managedBrokerLabel, callbackMethod): """Discard all messages on queue""" @@ -124,14 +116,6 @@ model.getConnectedBroker(managedBrokerLabel).method(methodId, self.idOriginal, \ classToSchemaNameMap[self.__class__.__name__], "purge", args=actualArgs, packageName="qpid") - def increaseJournalSize(self, model, managedBrokerLabel, callbackMethod, pages): - """Increase number of disk pages allocated for this queue""" - actualArgs = dict() - actualArgs["pages"] = pages - methodId = model.registerCallback(callbackMethod) - model.getConnectedBroker(managedBrokerLabel).method(methodId, self.idOriginal, \ - classToSchemaNameMap[self.__class__.__name__], "increaseJournalSize", args=actualArgs, packageName="qpid") - Vhost.sqlmeta.addJoin(MultipleJoin('Queue', joinMethodName='queues')) @@ -139,34 +123,6 @@ idOriginal = BigIntCol(default=None) recTime = TimestampCol(default=None) queue = ForeignKey('Queue', cascade='null', default=None) - journalLocation = StringCol(length=1000, default=None) - journalBaseFileName = StringCol(length=1000, default=None) - journalInitialFileCount = IntCol(default=None) - journalCurrentFileCount = IntCol(default=None) - journalDataFileSize = IntCol(default=None) - journalFreeFileCount = IntCol(default=None) - journalFreeFileCountLow = IntCol(default=None) - journalFreeFileCountHigh = IntCol(default=None) - journalAvailableFileCount = IntCol(default=None) - journalAvailableFileCountLow = IntCol(default=None) - journalAvailableFileCountHigh = IntCol(default=None) - journalRecordDepth = IntCol(default=None) - journalRecordDepthLow = IntCol(default=None) - journalRecordDepthHigh = IntCol(default=None) - journalRecordEnqueues = BigIntCol(default=None) - journalRecordDequeues = BigIntCol(default=None) - journalWriteWaitFailures = BigIntCol(default=None) - journalWriteBusyFailures = BigIntCol(default=None) - journalReadRecordCount = BigIntCol(default=None) - journalReadBusyFailures = BigIntCol(default=None) - journalWritePageCacheDepth = IntCol(default=None) - journalWritePageCacheDepthLow = IntCol(default=None) - journalWritePageCacheDepthHigh = IntCol(default=None) - journalWritePageSize = IntCol(default=None) - journalReadPageCacheDepth = IntCol(default=None) - journalReadPageCacheDepthLow = IntCol(default=None) - journalReadPageCacheDepthHigh = IntCol(default=None) - journalReadPageSize = IntCol(default=None) msgTotalEnqueues = BigIntCol(default=None) msgTotalDequeues = BigIntCol(default=None) msgTxnEnqueues = BigIntCol(default=None) @@ -206,6 +162,9 @@ unackedMessages = IntCol(default=None) unackedMessagesLow = IntCol(default=None) unackedMessagesHigh = IntCol(default=None) + messageLatencyMin = BigIntCol(default=None) + messageLatencyMax = BigIntCol(default=None) + messageLatencyAvg = BigIntCol(default=None) Queue.sqlmeta.addJoin(MultipleJoin('QueueStats', joinMethodName='stats')) @@ -253,15 +212,15 @@ managedBroker = StringCol(length=1000, default=None) statsCurr = ForeignKey('BindingStats', cascade='null', default=None) statsPrev = ForeignKey('BindingStats', cascade='null', default=None) - queue = ForeignKey('Queue', cascade='null', default=None) exchange = ForeignKey('Exchange', cascade='null', default=None) + queue = ForeignKey('Queue', cascade='null', default=None) bindingKey = StringCol(length=1000, default=None) -Queue.sqlmeta.addJoin(MultipleJoin('Binding', joinMethodName='bindings')) - Exchange.sqlmeta.addJoin(MultipleJoin('Binding', joinMethodName='bindings')) +Queue.sqlmeta.addJoin(MultipleJoin('Binding', joinMethodName='bindings')) + class BindingStats(SQLObject): idOriginal = BigIntCol(default=None) recTime = TimestampCol(default=None) @@ -280,8 +239,7 @@ statsCurr = ForeignKey('ClientStats', cascade='null', default=None) statsPrev = ForeignKey('ClientStats', cascade='null', default=None) vhost = ForeignKey('Vhost', cascade='null', default=None) - ipAddr = IntCol(default=None) - port = SmallIntCol(default=None) + address = StringCol(length=1000, default=None) def close(self, model, managedBrokerLabel, callbackMethod): actualArgs = dict() @@ -303,9 +261,9 @@ recTime = TimestampCol(default=None) client = ForeignKey('Client', cascade='null', default=None) authIdentity = StringCol(length=1000, default=None) - msgsProduced = BigIntCol(default=None) + framesFromClient = BigIntCol(default=None) msgsConsumed = BigIntCol(default=None) - bytesProduced = BigIntCol(default=None) + bytesFromClient = BigIntCol(default=None) bytesConsumed = BigIntCol(default=None) Client.sqlmeta.addJoin(MultipleJoin('ClientStats', joinMethodName='stats')) Modified: mgmt/mint/python/mint/schema.sql =================================================================== --- mgmt/mint/python/mint/schema.sql 2007-12-13 18:09:05 UTC (rev 1479) +++ mgmt/mint/python/mint/schema.sql 2007-12-13 19:24:38 UTC (rev 1480) @@ -7,10 +7,13 @@ id SERIAL PRIMARY KEY, name VARCHAR(1000) ); -CREATE TABLE broker_group_broker_registration ( -broker_group_id INT NOT NULL, -broker_registration_id INT NOT NULL + +CREATE TABLE broker_group_mapping ( + id SERIAL PRIMARY KEY, + broker_registration_id INT NOT NULL, + broker_group_id INT NOT NULL ); +CREATE UNIQUE INDEX broker_group_mapping_unique ON broker_group_mapping (broker_registration_id, broker_group_id); CREATE TABLE broker_profile ( id SERIAL PRIMARY KEY, @@ -31,8 +34,7 @@ id SERIAL PRIMARY KEY, name VARCHAR(1000), value VARCHAR(1000), - type VARCHAR(1), - profile_id INT + type VARCHAR(1) ); CREATE TABLE binding ( @@ -44,8 +46,8 @@ managed_broker VARCHAR(1000), stats_curr_id INT, stats_prev_id INT, - queue_id INT, exchange_id INT, + queue_id INT, binding_key VARCHAR(1000) ); @@ -99,8 +101,7 @@ stats_curr_id INT, stats_prev_id INT, vhost_id INT, - ip_addr INT, - port SMALLINT + address VARCHAR(1000) ); CREATE TABLE client_stats ( @@ -109,9 +110,9 @@ rec_time TIMESTAMP, client_id INT, auth_identity VARCHAR(1000), - msgs_produced BIGINT, + frames_from_client BIGINT, msgs_consumed BIGINT, - bytes_produced BIGINT, + bytes_from_client BIGINT, bytes_consumed BIGINT ); @@ -233,8 +234,7 @@ name VARCHAR(1000), durable BOOL, auto_delete BOOL, - exclusive BOOL, - page_memory_limit INT + exclusive BOOL ); CREATE TABLE queue_stats ( @@ -242,34 +242,6 @@ id_original BIGINT, rec_time TIMESTAMP, queue_id INT, - journal_location VARCHAR(1000), - journal_base_file_name VARCHAR(1000), - journal_initial_file_count INT, - journal_current_file_count INT, - journal_data_file_size INT, - journal_free_file_count INT, - journal_free_file_count_low INT, - journal_free_file_count_high INT, - journal_available_file_count INT, - journal_available_file_count_low INT, - journal_available_file_count_high INT, - journal_record_depth INT, - journal_record_depth_low INT, - journal_record_depth_high INT, - journal_record_enqueues BIGINT, - journal_record_dequeues BIGINT, - journal_write_wait_failures BIGINT, - journal_write_busy_failures BIGINT, - journal_read_record_count BIGINT, - journal_read_busy_failures BIGINT, - journal_write_page_cache_depth INT, - journal_write_page_cache_depth_low INT, - journal_write_page_cache_depth_high INT, - journal_write_page_size INT, - journal_read_page_cache_depth INT, - journal_read_page_cache_depth_low INT, - journal_read_page_cache_depth_high INT, - journal_read_page_size INT, msg_total_enqueues BIGINT, msg_total_dequeues BIGINT, msg_txn_enqueues BIGINT, @@ -308,7 +280,10 @@ bindings_high INT, unacked_messages INT, unacked_messages_low INT, - unacked_messages_high INT + unacked_messages_high INT, + message_latency_min BIGINT, + message_latency_max BIGINT, + message_latency_avg BIGINT ); CREATE TABLE session ( @@ -375,22 +350,24 @@ vhost_id INT ); +ALTER TABLE broker_group_mapping ADD CONSTRAINT broker_registration_id_exists FOREIGN KEY (broker_registration_id) REFERENCES broker_registration (id) ON DELETE CASCADE; + +ALTER TABLE broker_group_mapping ADD CONSTRAINT broker_group_id_exists FOREIGN KEY (broker_group_id) REFERENCES broker_group (id) ON DELETE CASCADE; + ALTER TABLE broker_registration ADD CONSTRAINT broker_id_exists FOREIGN KEY (broker_id) REFERENCES broker (id) ON DELETE SET NULL; ALTER TABLE broker_registration ADD CONSTRAINT cluster_id_exists FOREIGN KEY (cluster_id) REFERENCES broker_cluster (id) ON DELETE SET NULL; ALTER TABLE broker_registration ADD CONSTRAINT profile_id_exists FOREIGN KEY (profile_id) REFERENCES broker_profile (id) ON DELETE SET NULL; -ALTER TABLE config_property ADD CONSTRAINT profile_id_exists FOREIGN KEY (profile_id) REFERENCES broker_profile (id) ON DELETE SET NULL; - ALTER TABLE binding ADD CONSTRAINT stats_curr_id_exists FOREIGN KEY (stats_curr_id) REFERENCES binding_stats (id) ON DELETE SET NULL; ALTER TABLE binding ADD CONSTRAINT stats_prev_id_exists FOREIGN KEY (stats_prev_id) REFERENCES binding_stats (id) ON DELETE SET NULL; -ALTER TABLE binding ADD CONSTRAINT queue_id_exists FOREIGN KEY (queue_id) REFERENCES queue (id) ON DELETE SET NULL; - ALTER TABLE binding ADD CONSTRAINT exchange_id_exists FOREIGN KEY (exchange_id) REFERENCES exchange (id) ON DELETE SET NULL; +ALTER TABLE binding ADD CONSTRAINT queue_id_exists FOREIGN KEY (queue_id) REFERENCES queue (id) ON DELETE SET NULL; + ALTER TABLE binding_stats ADD CONSTRAINT binding_id_exists FOREIGN KEY (binding_id) REFERENCES binding (id) ON DELETE SET NULL; ALTER TABLE broker ADD CONSTRAINT stats_curr_id_exists FOREIGN KEY (stats_curr_id) REFERENCES broker_stats (id) ON DELETE SET NULL; Modified: mgmt/notes/justin-todo.txt =================================================================== --- mgmt/notes/justin-todo.txt 2007-12-13 18:09:05 UTC (rev 1479) +++ mgmt/notes/justin-todo.txt 2007-12-13 19:24:38 UTC (rev 1480) @@ -4,8 +4,6 @@ * Add legends to charts - * Make group slider in broker browser work - * Add a groups column to the browser broker list * Sort in tables @@ -26,8 +24,6 @@ * Paginate sessions - * Prevent users from mapping a broker to a group more than once - * Deal with problem of calling method on broker that is not there * Only put something in pending actions if the call succeeds @@ -39,10 +35,6 @@ * Need to handle exceptions in broker connect thread, so it doesn't stop trying - * Ask tross to take some prints out of ManagedBroker.start - - * Get rid of CuminClass.mint_stats_class - * Add inactive state to some status lights * Fix session and client naming @@ -52,6 +44,10 @@ Deferred + * Ask tross to take some prints out of ManagedBroker.start + + * Get rid of CuminClass.mint_stats_class + * Add a do_get_item_count, and cache result for use by get_item_count * Add a ~3 second (or use broker update interval, if we can get that)