Index: trunk/ippm/FilesMonitoring/Summary/snapshot.py =================================================================== --- trunk/ippm/FilesMonitoring/Summary/snapshot.py (revision 35983) +++ (revision ) @@ -1,281 +1,0 @@ -#!/usr/bin/env python - -import MySQLdb -import sys -import socket -import _mysql_exceptions -import datetime -import logging - -def process_arguments(argv): - if len(argv) == 1: - usage() - sys.exit(1) - index = 1 - nodes = [ ] - while index < len(argv): - if argv[index] == '-v': - logger.setLevel(logging.DEBUG) - index += 1 - elif argv[index] == 'mhpcc': - nodesFilename = '%s/%s/share/filemon/ippMhpccStorageNodes' % (os.environ['PSCONFDIR'], os.environ['PSCONFIG']) - nodesFile = open(nodesFilename) - nodes = [] - for line in nodesFile: - nodes.append((line[:-1]+".0")) - index += 1 - elif argv[index] == 'set0': - nodes.extend([ ("ipp004.0"), ("ipp005.0"), ("ipp006.0"), ("ipp007.0"), ("ipp008.0"), ("ipp009.0"), - ("ipp060.0"), ("ipp061.0"), ("ipp062.0"), ("ipp063.0"), ("ipp064.0"), ("ipp065.0"), - ("ipp066.0"), ] ) - index += 1 - elif argv[index] == 'set1': - nodes.extend([ ("ipp010.0"), ("ipp011.0"), ("ipp012.0"), ("ipp013.0"), ("ipp014.0"), - ("ipp015.0"), ("ipp016.0"), ("ipp017.0"), ("ipp018.0"), ("ipp019.0"), ] ) - index += 1 - elif argv[index] == 'set2': - nodes.extend( [ ("ipp020.0"), ("ipp021.0"), ("ipp023.0"), ("ipp024.0"), - ("ipp025.0"), ("ipp026.0"), ("ipp027.0"), ("ipp028.0"), ("ipp029.0"), ] ) - index += 1 - elif argv[index] == 'set3': - nodes.extend([ ("ipp030.0"), ("ipp031.0"), ("ipp032.0"), ("ipp033.0"), ("ipp034.0"), - ("ipp035.0"), ("ipp036.0"), ("ipp037.0"), ("ipp038.0"), ("ipp039.0"), ] ) - index += 1 - elif argv[index] == 'set4': - nodes.extend( [ ("ipp040.0"), ("ipp041.0"), ("ipp042.0"), ("ipp043.0"), ("ipp044.0"), - ("ipp045.0"), ("ipp046.0"), ("ipp047.0"), ("ipp048.0"), ("ipp049.0"), ] ) - index += 1 - elif argv[index] == 'set5': - nodes.extend( [ ("ipp050.0"), ("ipp051.0"), ("ipp052.0"), ("ipp053.0"), ("ipp054.0"), - ("ipp055.0"), ("ipp056.0"), ("ipp057.0"), ("ipp058.0"), ("ipp059.0"), ] ) - index += 1 - elif argv[index] == 'ippb': - nodes.extend( [ ("ippb00.0"), ("ippb00.1"), ("ippb00.2"), - ("ippb01.0"), ("ippb01.1"), ("ippb01.2"), - ("ippb02.0"), ("ippb02.1"), ("ippb02.2"), - ("ippb03.0"), ("ippb03.1"), ("ippb03.2"), ] ) - index += 1 - elif argv[index] == 'stsci': - nodes.extend( [ ("stsci00.0"), ("stsci00.1"), ("stsci00.2"), - ("stsci01.0"), ("stsci01.1"), ("stsci01.2"), - ("stsci02.0"), ("stsci02.1"), ("stsci02.2"), - ("stsci03.0"), ("stsci03.1"), ("stsci03.2"), - ("stsci04.0"), ("stsci04.1"), ("stsci04.2"), - ("stsci05.0"), ("stsci05.1"), ("stsci05.2"), - ("stsci06.0"), ("stsci06.1"), ("stsci06.2"), - ("stsci07.0"), ("stsci07.1"), ("stsci07.2"), - ("stsci08.0"), ("stsci08.1"), ("stsci08.2"), - ("stsci09.0"), ("stsci09.1"), ("stsci09.2"), - ("stsci10.0"), ("stsci10.1"), ("stsci10.2"), - ("stsci11.0"), ("stsci11.1"), ("stsci11.2"), - ("stsci12.0"), ("stsci12.1"), ("stsci12.2"), - ("stsci13.0"), ("stsci13.1"), ("stsci13.2"), - ("stsci14.0"), ("stsci14.1"), ("stsci14.2"), - ("stsci15.0"), ("stsci15.1"), ("stsci15.2"), - ("stsci16.0"), ("stsci16.1"), ("stsci16.2"), - ("stsci17.0"), ("stsci17.1"), ("stsci17.2"), - ("stsci18.0"), ("stsci18.1"), ("stsci18.2"), - ("stsci19.0"), ("stsci19.1"), ("stsci19.2"), ] ) - index += 1 - else: - node = argv[index] - if node.startswith('ippb'): - nodes.append((node+".0")) - nodes.append((node+".1")) - nodes.append((node+".2")) - elif node.startswith('stsci'): - nodes.append((node+".0")) - nodes.append((node+".1")) - nodes.append((node+".2")) - else: - nodes.append((node+".0")) - index += 1 - return nodes - -def get_snapshot(disk, dirname, fm_cursor, dms_cursor): - logger.debug("Getting snapshot of [%s]" % dirname) - select_query = 'SELECT role, stage, lifetime, SUM(size), COUNT(*), status FROM neb_%s_%s GROUP BY role, stage, lifetime, status' - insert_query = """INSERT INTO CurrentDiskUsage -SET node='%s', role='%s', stage='%s', dirname='%s', sizes=%d, counts=%d, lifetime='%s', last_update=NOW(), status='%s'""" - update_query = """UPDATE CurrentDiskUsage -SET sizes=%d, counts=%d, last_update=NOW() -WHERE node='%s' AND role='%s' AND stage='%s' AND dirname='%s' AND lifetime='%s' AND status='%s'""" - undefined_found = False - fm_cursor.execute(select_query % (disk, dirname)) - for (role, stage, lifetime, size, count, status) in fm_cursor.fetchall(): - if status == 'undefined': - undefined_found = True - if size < 0: - size=0 - try: - dms_cursor.execute(insert_query % (alias, - role, - stage, - dirname, - size, - count, - lifetime, - status) ) - except Exception, e: - dms_cursor.execute(update_query % (size, - count, - alias, - role, - stage, - dirname, - lifetime, - status) ) - if not undefined_found: - logger.debug('... No undefined found for %s' % dirname) - try: - logger.debug('... Trying to delete undefined entries') - dms_cursor.execute(delete_undefined_query % (alias) ) - except Exception, e: - logger.debug('... Trying to update') - dms_cursor.execute(update_query % (0, - 0, - alias, - 'undefined', - 'undefined', - dirname, - 'undefined', - 'undefined') ) - else: - logger.warn('... There are still undefined values for %s (dirname: %s)' % (alias, dirname)) - # Now delete all entries with status set as 'deleted' after the - # deleted entries have been measures - logger.debug('... Deleting deleted entries from %s (dirname: %s)' % (alias, dirname)) - stmt = "SELECT role, stage, lifetime, SUM(size), COUNT(*) FROM neb_%s_%s_delete GROUP BY role, stage, lifetime, status" - logger.debug(stmt % (disk, dirname)) - fm_cursor.execute(stmt % (disk, dirname)) - rows = fm_cursor.fetchall() - logger.debug("Got %d deleted entries" % len(rows)) - for (role, stage, lifetime, size, count) in rows: - try: - dms_cursor.execute(insert_query % (alias, - role, - stage, - dirname, - size, - count, - lifetime, - 'deleted') ) - except Exception, e: - dms_cursor.execute(update_query % (size, - count, - alias, - role, - stage, - dirname, - lifetime, - 'deleted') ) - try: - fm_cursor.execute("DELETE FROM neb_%s_%s_delete" % (disk, dirname)) - except _mysql_exceptions.OperationalError, e: - try: - # Try again?! - fm_cursor.execute("DELETE FROM neb_%s_%s_delete" % (disk, dirname)) - except _mysql_exceptions.OperationalError, e: - logger.warn("Couldn't delete 'deleted' entries from neb_%s_%s_delete on %s" % (dirname, disk, node)) - if not undefined_found: - logger.debug('... No undefined found for %s' % dirname) - delete_undefined_query = """DELETE FROM CurrentDiskUsage -WHERE node='%s' AND (status='undefined' OR role='undefined') -""" - try: - logger.debug('... Trying deleting undefined entries') - dms_cursor.execute(delete_undefined_query % (alias) ) - except Exception, e: - logger.debug('... Trying update') - dms_cursor.execute(update_query % (0, - 0, - alias, - 'undefined', - 'undefined', - dirname, - 'undefined', - 'undefined') ) - else: - logger.warn('... There are still undefined values for %s (dirname: %s)' % (alias, dirname)) - # Now move entries for this node which are older than the current date - logger.debug('... Moving old entries from CurrentDiskUsage for %s' % (alias)) - insert_select_statement = """INSERT IGNORE INTO DiskUsageTemporary(node, role, stage, sizes, counts, lifetime, dirname, status, last_update) -SELECT node, role, stage, sizes, counts, lifetime, dirname, status, last_update FROM CurrentDiskUsage -WHERE node='%s' AND last_update<'%s'""" - dms_cursor.execute(insert_select_statement % (node, snapshot_date)); - dms_cursor.execute("DELETE FROM CurrentDiskUsage WHERE node='%s' AND last_update<'%s'" % (alias, snapshot_date)); - -def update_daily_and_weekly(dms_cursor): - try: - stmt = 'CALL UpdateDailyByNode(\'%s\')' % alias - logger.info('... Calling UpdateDailyByNode: [%s]' % stmt) - dms_cursor.execute(stmt) - except _mysql_exceptions.OperationalError, e: - logger.warn('... MySQL Exception caught when calling UpdateDailyByNode(): ' + str(e) + '') - try: - logger.info('... Trying to guess if running UpdateWeeklyByNode makes sense') - today = datetime.datetime.today() - last_week = '%d-%d' % (today.year, today.isocalendar()[1]-1) - query = "SELECT COUNT(*), node, role, stage, lifetime, dirname, week, status FROM DiskUsageHistoryWeekly WHERE node = \'%s\' AND week = '%s' GROUP BY node, role, stage, lifetime, dirname, week, status" % (alias, last_week) - dms_cursor.execute(query) - row = dms_cursor.fetchone() - run_updateweekly = (row is None) - if run_updateweekly: - stmt = 'CALL UpdateWeeklyByNode(\'%s\')' % alias - logger.info('... Executing: %s' % stmt) - dms_cursor.execute(stmt) - else: - logger.info('... UpdateWeekly() call cancelled since last week entry already in database: %s' % last_week) - except _mysql_exceptions.OperationalError, e: - logger.warn('... MySQL Exception caught when calling UpdateWeekly(): ' + str(e) ) - -if __name__ == "__main__": - # Setup logging - logger = logging.getLogger() - formatter = logging.Formatter('%(asctime)s | %(levelname)7s | %(message)s', - '%Y-%m-%dT%H:%M:%S') - logging_output = logging.StreamHandler(sys.stderr) - logging_output.setFormatter(formatter) - logger.addHandler(logging_output) - logger.setLevel(logging.INFO) - - nodes = process_arguments(sys.argv) - letters = '0123456789abcdef' - - nodename = socket.gethostbyaddr(socket.gethostname())[0].split('.')[0] - if nodename != 'neverland' and nodename != 'ian' : - dms_db = MySQLdb.connect('ippc11', 'ippuser', 'ipp', - 'DiskMonitoringSummary') - else: # Test platform - logger.info("On [%s]" % nodename) - if nodename == 'ian': - root_passwd = '' - else: - root_passwd = 'root' - dms_db = MySQLdb.connect('localhost', 'root', root_passwd, - 'DiskMonitoringSummary') - dms_cursor = dms_db.cursor() - - for alias in nodes: - node, disk = alias.split('.') - logger.info('Getting snapshot of node: %s (%s)' % (node, alias)) - try: - if node != 'neverland' and node != 'ian' : - fm_db = MySQLdb.connect(node, 'ipp', 'ipp', 'DisksMonitoring') - else: - fm_db = MySQLdb.connect('localhost', 'root', root_passwd, - 'DisksMonitoring') - fm_cursor = fm_db.cursor() - snapshot_date = datetime.datetime.now() - for letter1 in letters: - logger.info('... Getting snapshot for nebulous subdirectories starting with %s (%s0, %s1, ...)' % (letter1, - letter1, - letter1)) - for letter2 in letters: - dirname = '%s%s' % (letter1, letter2) - get_snapshot(disk, dirname, fm_cursor, dms_cursor) - update_daily_and_weekly(dms_cursor) - fm_db.commit() - dms_db.commit() - except _mysql_exceptions.OperationalError, e: - logger.error(str(e))