Index: trunk/ippm/FilesMonitoring/NodeRelated/deployment/generate_crontabs_and_deployment_script.py =================================================================== --- trunk/ippm/FilesMonitoring/NodeRelated/deployment/generate_crontabs_and_deployment_script.py (revision 35984) +++ (revision ) @@ -1,41 +1,0 @@ -#!/usr/bin/env python - -bin_directory = "%s/%s/bin" % (os.environ['PSCONFDIR'], os.environ['PSCONFIG']) -share_directory = "%s/%s/share/filemon/" % (os.environ['PSCONFDIR'], os.environ['PSCONFIG']) - -REFRESH_SCRIPT_LOCATION = '%s/refresh_table.py' % bin_directory -ROOT_DIRECTORY = '%s/deployment' % share_directory -PYTHONPATH = '%s/python' % share_directory -NODES_LIST = '%s/allStorageNodes' % share_directory -OUTPUT_DIRECTORY = '%s/scripts' % share_directory -DEPLOYMENT_SCRIPT = '%s/deploy.sh' % OUTPUT_DIRECTORY -NODE_SCRIPT_FORMAT = '%s/crontab.%%s' % OUTPUT_DIRECTORY - -if __name__ == '__main__': - nodes_file = open(NODES_LIST) - deployment_script_file = open(DEPLOYMENT_SCRIPT, 'w') - deployment_script_file.write('#!/bin/bash\n\n') - minutes = 1 - hours = 0 - for node in nodes_file: - node = node[:-1] - print '... Generating script for %s' % node - crontab_filename = NODE_SCRIPT_FORMAT % node - deployment_script_file.write("echo '%s'\n" % (node)) - deployment_script_file.write("ssh %s 'crontab %s'\n" % (node, crontab_filename)) - crontab_file = open(crontab_filename, 'w') - crontab_file.write("MAILTO=%s@ifa.hawaii.edu\n" % os.environ['USER'] ) - crontab_file.write("PYTHONPATH=%s\n" % PYTHONPATH) - crontab_file.write("%d %d * * * nice %s -disk 0\n" % (minutes, hours, REFRESH_SCRIPT_LOCATION)) - if not node.startswith('ipp'): #atrc or stsci nodes: add disks 1 and 2 - crontab_file.write("%d %d * * * nice %s -disk 1\n" % ((minutes+1)%60, hours, REFRESH_SCRIPT_LOCATION)) - crontab_file.write("%d %d * * * nice %s -disk 2\n" % ((minutes+2)%60, hours, REFRESH_SCRIPT_LOCATION)) - minutes += 7 - minutes %= 60 - hours += 1 - hours %= 24 - crontab_file.close() - nodes_file.close() - deployment_script_file.close() - print '\nNow run: %s\n' % DEPLOYMENT_SCRIPT - Index: trunk/ippm/FilesMonitoring/NodeRelated/refresh_table.py =================================================================== --- trunk/ippm/FilesMonitoring/NodeRelated/refresh_table.py (revision 35984) +++ (revision ) @@ -1,555 +1,0 @@ -#!/usr/bin/env python - -import sys -import os -import socket -import subprocess -import MySQLdb -import logging -import _mysql_exceptions -import datetime -import time - -################################### -# -class Constants: # Constants and/or default values - force = False # if we want to ignore the lock - limit_undefined = 10000 # the maximum number of undefined entries that we want to refresh - limit = 1000 # the maximum number of the oldest refreshed entries that we want to refresh - loggingLevel = logging.INFO # - skip_stages = [] -# Place we find all the installed IPP stuff - bin_directory = "%s/%s/bin" % (os.environ['PSCONFDIR'], os.environ['PSCONFIG']) - share_directory = "%s/%s/share/filemon/" % (os.environ['PSCONFDIR'], os.environ['PSCONFIG']) - -# Templates for things we'll either create or read - node_directory = '/export/%s.%s/ipp_filemon/' - mlocate_db_template = "%s/%s", (node_directory,'mlocate/%s.%s.%s.mlocate_db') - lock_directory = '/tmp/diskMonitoring/%s' # disk - lock_filename_template = '%s/lock.%s' - nebulous_directory_template = '/export/%s.%s/nebulous/%s' - -# Locations of the binaries we'll run - updatedb_binary = '%s/updatedb' % bin_directory - locate_binary = '%s/locate' % bin_directory - update_table_undefined_python_script = "%s/update_table_undefined.py" % bin_directory - update_table_python_script = '%s/update_table.py' % bin_directory - - disk = "0" # Most of the nodes have only one disk /export/.0 - next_directory = None - max_refresh_interval = 6*60*60 # Run the script if the last_modified value associated to the disk to refresh is older than that (6 hours) - # sleep = 60*10 # The time (in seconds) the script sleeps between two iterations - sleep = 1 # The time (in seconds) the script sleeps between two iterations - -def get_next_directory(current_directory): - """ - >>> get_next_directory("00") - '01' - >>> get_next_directory("ff") - '00' - >>> get_next_directory("ab") - 'ac' - >>> get_next_directory("cf") - 'd0' - """ - if Constants.next_directory is None: - Constants.next_directory = dict() - letters = '0123456789abcdef' - directories = [] - for letter1 in letters: - for letter2 in letters: - directories.append('%s%s'% (letter1, letter2)) - for index in range(len(directories)-1): - Constants.next_directory[directories[index]] = directories[index+1] - Constants.next_directory['ff'] = '00' - return Constants.next_directory[current_directory] - -################################### -# -class FilesDatabaseConnection: - def __init__(self, nodename, disk): - self.nodename = nodename - self.disk = disk - if nodename == 'neverland': - self.db = MySQLdb.connect('localhost', 'root', 'root', 'DisksMonitoring') - else: - self.db = MySQLdb.connect('localhost', 'root', '', 'DisksMonitoring') - self.count_inserts = 0 - - def close(self): - self.db.commit() - self.db.close() - - def find_instance(self, neb_ins_id, subdirname, filename): - query = 'SELECT neb_ins_id FROM neb_%s_%s WHERE neb_ins_id = %s' % (self.dirname, - self.disk, - neb_ins_id) - cursor = self.db.cursor() - cursor.execute(query) - rows = cursor.fetchall() - if len(rows) == 0: - print 'New entry' - elif len(rows) == 1: - print 'Existing entry' - else: - print 'Problem' - cursor.close() - - def get_instance_ids(self): - query = 'SELECT neb_ins_id, subdirname, filename FROM neb_%s_%s' % (self.disk, - self.dirname) - cursor = self.db.cursor() - cursor.execute(query) - ids = dict() - for row in cursor.fetchall(): - ids[row[0]] = (row[1], row[2]) - cursor.close() - return ids - - def add_new_file(self, filename, neb_ins_id, subdirname, debug = False): - query = """INSERT INTO neb_%s_%s ( - filename, - neb_ins_id, - subdirname, - nebkey, - creation_date) -VALUES ( - '%s', - %s, - '%s', - '%s', - NOW())""" % (self.disk, - self.dirname, - filename, - neb_ins_id, - subdirname, - generate_nebulous_key(filename)) - if debug: - logger.debug('%s;' % query) - return - cursor = self.db.cursor() - try: - cursor.execute(query) - except _mysql_exceptions.OperationalError, e: - logger.error(str(e)) - logger.info('Trying to replay query: %s' % query) - cursor.execute(query) - self.count_inserts += 1 - cursor.close() - if self.count_inserts >= 100: - self.db.commit() - self.count_inserts = 0 - - def mark_as_deleted(self, neb_ins_id): - table = "neb_%s_%s" % (self.disk, self.dirname) - logger.debug("Deleting entry %d from %s" % (neb_ins_id, table)) - cursor = self.db.cursor() - statement = "INSERT INTO %s_delete(filename, neb_ins_id, subdirname, nebkey, size, md5sum, role, stage, lifetime, status, creation_date, last_check) SELECT filename, neb_ins_id, subdirname, nebkey, size, md5sum, role, stage, lifetime, status, creation_date, last_check FROM %s WHERE neb_ins_id = %d" % (table, table, neb_ins_id) - logger.info('... Statement: (%s)' % (statement)) - try: - cursor.execute(statement) - except _mysql_exceptions.IntegrityError, e: - logger.error(e) - logger.error("Table is %s_delete" % (table)) - if 'Duplicate entry' in str(e): - pass - else: - raise e - except _mysql_exceptions.OperationalError, e: - logger.error(e) - if 'Lock wait timeout exceeded' in str(e): - pass - else: - raise e - statement = 'UPDATE %s_delete SET status=\'deleted\' WHERE neb_ins_id=%d' % (table, neb_ins_id) - logger.info('... Statement: (%s)' % (statement)) - cursor.execute(statement) - statement = 'DELETE FROM %s WHERE neb_ins_id=%d' % (table, neb_ins_id) - logger.info('... Statement: (%s)' % (statement)) - cursor.execute(statement) - cursor.close() - - def optimize_table(self): - cursor = self.db.cursor() - query = 'OPTIMIZE TABLE neb_%s_%s' % (self.disk, - self.dirname) - logger.debug('... Running statement "%s"' % (query)) - cursor.execute(query) - query = 'OPTIMIZE TABLE neb_%s_%s_delete' % (self.disk, - self.dirname) - logger.debug('... Running statement "%s"' % (query)) - cursor.execute(query) - cursor.close() - self.commit() - - def commit(self): - self.db.commit() - - def get_directory_to_refresh(self, force): - cursor = self.db.cursor() - cursor.execute('SELECT dirname, last_modified FROM refresh_status WHERE disk=%s' % self.disk) - for row in cursor.fetchall(): - (dirname, last_modified) = row - cursor.close() - then = last_modified - now = datetime.datetime.now() - td = now - then - delta_seconds = td.seconds + td.days * 24 * 3600 - if delta_seconds < Constants.max_refresh_interval and not force: - logger.warn("No need to restart the script") - sys.exit(0) - self.dirname = get_next_directory(dirname) - return self.dirname - - def update_refresh_status(self, dirname): - self.dirname = dirname - cursor = self.db.cursor() - cursor.execute("UPDATE refresh_status SET dirname = '%s' WHERE disk=%s" % (self.dirname, self.disk)) - self.commit() - cursor.close() - -# ################################### -# # -# class SummaryDatabaseConnection: -# def __init__(self): -# self.nodename = nodename -# self.dirname = dirname -# if nodename == 'neverland': -# self.db = MySQLdb.connect('localhost', 'root', 'root', 'DiskMonitoringSummary') -# else: -# self.db = MySQLdb.connect('localhost', 'ipp', 'ipp', 'DiskMonitoringSummary') - -################################### -# generate_nebulous_key() -def generate_nebulous_key(filename): - elements = filename.split('.') - neb_key = '.'.join(elements[1:]).replace(':', '/') - return neb_key - -################################### -# process_options() -def process_options(arguments): - # Default values: see usage() - force = Constants.force - limit = Constants.limit - skip_stages = Constants.skip_stages - index = 1 - disk = Constants.disk - while index] [-disk ] - %s -h - %s -unittest - - Refreshes files monitoring information about the - different /export/%s.0/nebulous/ nebulous storage - directories. is one of the 00..ff directories. - is obtained from the local database (DisksMonitoring/Status). - - The different stages are: - 0. Initializations. Exit if the information about the last refreshed - directory has been updated less than 24 hours ago. - 1. Get the directory to refresh - 2. Run updatedb on the current directory - 3. Remove deleted files from the FilesMonitoring database - 4. Add new files in the FilesMonitoring database - 5. Refresh the status of the oldest entries in the corresponding - table. - 6. Update the information about the directory. Goto 1 - - Arguments: - None: - : The nebulous directory to be refreshed 00..ff - - Options: - -f: - Force to ignore the lock. Beware of multiple instances of this - program running at the same time - -h, -help, --help: - Displays this help - -v: - Be verbose - -l/-limit : - Update up to new and then oldest files in the monitoring - database. By default, is %d - -disk : - Run for /export/./nebulous. - Default is /export/.%s/nebulous - -skip [,...]: - Debugging mode: Skip - You are supposed to know what you are doing when using this - option! - -unittest - Runs the unit tests associated to this program -""" - nodename = socket.gethostbyaddr(socket.gethostname())[0].split('.')[0] - sys.stderr.write(usage.__doc__ % (sys.argv[0], sys.argv[0], sys.argv[0], - nodename, Constants.limit, Constants.disk)) - -def refresh(dirname): - logger.info('Refreshing [%s]' % dirname) - ####################### - # 1. Stage 1: Run updatedb on directory /export/%s.%s/nebulous/ - ####################### - logger.info('Stage 1: updatedb') - if '1' not in skip_stages: - directory = Constants.nebulous_directory_template % (nodename, disk, dirname) - mlocate_db = Constants.mlocate_db_template % (nodename, - disk, - nodename, - disk, - dirname) - mlocate_db_directory = "/".join(mlocate_db.split('/')[:-1]) - if not os.path.exists(mlocate_db_directory): - os.makedirs(mlocate_db_directory) - p = subprocess.Popen( [Constants.updatedb_binary, - '-l', '0', - '-U', directory, - '-o', mlocate_db], - stderr = subprocess.PIPE, - stdout = subprocess.PIPE) - if p.wait()!=0: - logger.error('Error while running updatedb') - for line in p.stderr: - logger.error(line) - sys.exit(3) - logger.info('... Stage 1 complete') - else: - logger.info('... Stage 1 skipped') - - ####################### - # Stage 1.9: Collect all ins_id from the database and the mlocate file - ####################### - logger.info('Stage 1.9 (cannot be skipped): Get ins_id both from database and mlocate file') - if '2' in skip_stages and '3' in skip_stages: #Ignore if there is no database update - logger.info('... Stage 1.9 skipped since both 2 and 3 have to be skipped') - else: - # Collect ins_id from the database - ins_ids_in_db = filesMonitoring_database.get_instance_ids() - logger.debug('... Got %d entries from the database' % len(ins_ids_in_db)) - # Collect ins_id from the locate file - mlocate_db = Constants.mlocate_db_template % (nodename, - disk, - nodename, - disk, - dirname) - p = subprocess.Popen( [Constants.locate_binary, - '-d', mlocate_db, - '.'], - stderr = subprocess.PIPE, - stdout = subprocess.PIPE) - p.poll() - count = 0 - acceptable = 0 - root = Constants.nebulous_directory_template % (nodename, - disk, - dirname) - ins_ids_in_mlocate = dict() - for line in p.stdout: - long_filename = line[:-1].replace(root, - '') - if long_filename.startswith('/'): - long_filename = long_filename[1:] - try: - (subdirname, filename) = long_filename.split('/') - if logger.isEnabledFor(logging.DEBUG): - logger.debug('... -> %s' % line[:-1]) - logger.debug('...... [%s] split in (%s,%s)' % (long_filename, - subdirname, - filename)) - fields = filename.split('.') - if fields[0] != '': - ins_ids_in_mlocate[int(fields[0])] = (filename, subdirname) - else: - logger.warn('... Check file [%s]' % line[:-1]) - acceptable += 1 - except ValueError: - pass # These are the directories - count += 1 - logger.debug('... Looked at %d entries in updatedb database' % count) - logger.debug('... Found %d acceptable entries in updatedb database' % acceptable) - if acceptable+257!=count: - # We expect to have 256 subdirectories and the root dir - # excluded, that is exactly count-257 acceptable entries - # in the updatedb database. If not, it means that there is - # an unexpected file in the root directory - logger.debug('... !!! There might be a problem in updatedb database !!!') - logger.debug('... Got %d entries from the mlocate file' % len(ins_ids_in_mlocate)) - # Differences - logger.debug('... Checking differences') - deleted_entries = dict() - existing_entries = dict() - new_entries = dict() - for key in ins_ids_in_db.keys(): - if ins_ids_in_mlocate.has_key(key): - existing_entries[key] = ins_ids_in_mlocate[key] - else: - if ins_ids_in_db.has_key(key): - logger.debug('Adding [%s, %s] to files to delete' % (ins_ids_in_db[key])) - deleted_entries[key] = ins_ids_in_db[key] - else: - logger.debug('No entry in database for key [%s]' % key) - for key in ins_ids_in_mlocate.keys(): - if ins_ids_in_db.has_key(key): - existing_entries[key] = ins_ids_in_db[key] - else: - new_entries[key] = ins_ids_in_mlocate[key] - logger.info('... New files.............: %d' % len(new_entries)) - logger.info('... Deleted files.........: %d' % len(deleted_entries)) - logger.info('... Already existing files: %d' % (len(existing_entries))) - logger.info('... Stage 1.9 complete') - - ####################### - # 2. Stage 2: Remove deleted files from the FilesMonitoring database - ####################### - logger.info('Stage 2: remove deleted files from the FilesMonitoring database') - if '2' not in skip_stages: - logger.info('... %d entries to delete' % len(deleted_entries)) - for neb_ins_id in deleted_entries.keys(): - filesMonitoring_database.mark_as_deleted(neb_ins_id) - filesMonitoring_database.commit() - logger.info('... Stage 2 complete') - else: - logger.info('... Stage 2 skipped') - - ####################### - # 3. Add new files to the FilesMonitoring database - ####################### - logger.info('Stage 3: add new files to the FilesMonitoring database') - if '3' not in skip_stages: - logger.info('... %d entries to add' % len(new_entries)) - for neb_ins_id in new_entries.keys(): - filesMonitoring_database.add_new_file(new_entries[neb_ins_id][0], - neb_ins_id, - new_entries[neb_ins_id][1]) - filesMonitoring_database.commit() - logger.info('... Stage 3 complete') - else: - logger.info('... Stage 3 skipped') - - ####################### - # 3.9. OPTIMIZE TABLE - ####################### - logger.info('Stage 3.9 (cannot be skipped): optimize table') - if '2' in skip_stages and '3' in skip_stages: #Ignore if no db update - logger.info('... Stage 3.9 skipped since both 2 and 3 have to be skipped') - else: - filesMonitoring_database.optimize_table() - logger.info('... Stage 3.9 complete') - - ####################### - # 4. Refresh the status of the oldest entries in the - # corresponding table - ####################### - logger.info('Stage 4: refresh the status of the oldest entries (limit = %d)' % limit) - if '4' not in skip_stages: - p = subprocess.Popen( [Constants.update_table_undefined_python_script, - '-disk', disk, - '-l', str(Constants.limit_undefined), - dirname], - stderr = subprocess.PIPE, - stdout = subprocess.PIPE) - if p.wait()!=0: - logger.error('Error while running %s' % Constants.update_table_undefined_python_script) - for line in p.stderr: - logger.error(line) - sys.exit(3) - p = subprocess.Popen( [Constants.update_table_python_script, - '-disk', disk, - '-l', str(limit), - dirname], - stderr = subprocess.PIPE, - stdout = subprocess.PIPE) - if p.wait()!=0: - logger.error('Error while running %s' % Constants.update_table_python_script) - for line in p.stderr: - logger.error(line) - sys.exit(3) - logger.info('... Stage 4 complete') - else: - logger.info('... Stage 4 skipped') - - ################################################## - # 5. Stage 5.9: Update information in refresh_status table - ################################################## - filesMonitoring_database.update_refresh_status(dirname) - -################################### -# Main -if __name__ == '__main__': - ####################### - # 0. Initializations - ####################### - # Get nodename - nodename = socket.gethostbyaddr(socket.gethostname())[0].split('.')[0] - # 0.1. Setup logging - logger = logging.getLogger() - formatter = logging.Formatter('%(asctime)s | %(levelname)7s | %(message)s', - '%Y-%m-%dT%H:%M:%S') -# log_directory = '/data/%s.0/schastel/logs' % (nodename) - log_directory = ("%s/logs" % (node_directory)) % (nodename, disk) - log_filename = '%s/%s_refresh.log' % (log_directory, nodename) - try: - logging_output = logging.FileHandler(log_filename) - except IOError: - os.makedirs(log_directory) - logging_output = logging.FileHandler(log_filename) - logging_output.setFormatter(formatter) - logger.addHandler(logging_output) - logger.setLevel(Constants.loggingLevel) - # 0.2. Get program arguments - (force, limit, skip_stages, disk) = process_options(sys.argv) - logger.info('%s for [%s]/%s' % (sys.argv[0], nodename, disk)) - logger.info('Stage 0 (cannot be skipped): initializations') - # 0.3. Open a connection to the FilesMonitoring database - try: - filesMonitoring_database = FilesDatabaseConnection(nodename, disk) - except _mysql_exceptions.OperationalError, e: - logger.error('Caught exception while connecting to local MySQL server:\n\t%s' % str(e)) - sys.exit(1) - dirname = filesMonitoring_database.get_directory_to_refresh(force) - logger.info('Stage 0 complete') - while True: - refresh(dirname) - dirname = get_next_directory(dirname) - filesMonitoring_database.dirname = dirname # Dirty! - logger.info('Sleeping for a while (%d seconds)' % Constants.sleep) - time.sleep(Constants.sleep) Index: trunk/ippm/FilesMonitoring/NodeRelated/update_table.py =================================================================== --- trunk/ippm/FilesMonitoring/NodeRelated/update_table.py (revision 35984) +++ (revision ) @@ -1,204 +1,0 @@ -#!/usr/bin/env python - -import MySQLdb -import sys -import socket -import os -import datetime -import hashlib -from ipp.lifetime_templates import Template - -def md5sum(filename): - f = open(filename, "rb") - content = f.read() - f.close() - sum = hashlib.md5() - sum.update(content) - return sum.hexdigest() - -class DbEntry: - verbose = False - def __init__(self, db_key, full_filename): - self.db_key = db_key - self.full_filename = full_filename - - def __str__(self): - if not DbEntry.verbose and len(self.db_key) > 10: - db_key = '%s...' % self.db_key[:10] - full_filename = '...%s' % self.full_filename[-10:] - else: - db_key = self.db_key - full_filename = self.full_filename - if self.role != 'deleted': - return '[%s]: %s (%s/%s/%s), %d, %s, %s' % (db_key, - full_filename, - self.stage, - self.role, - self.lifetime, - self.size, - self.md5sum, - self.last_check) - else: - return '[%s]: %s (%s)' % (db_key, - full_filename, - self.role) - - def check(self): - try: - self.stage, self.role, self.lifetime = Template.classify(self.db_key) - stat_values = os.stat(self.full_filename) - self.size = stat_values.st_size - try: - self.md5sum = md5sum(self.full_filename) - self.status = 'ok' - except IOError: - self.status = 'error' - self.md5sum = None - except OSError: - self.status = 'deleted' - self.stage, self.role, self.lifetime = Template.classify(self.db_key) - self.last_check = datetime.datetime.now() - - def update(self, table): - if self.status == 'deleted': - self.stage, self.role, self.lifetime = Template.classify(self.db_key) - stmt = """UPDATE %s -SET status=\'%s\', - last_check=\'%s\', - role=\'%s\', - stage=\'%s\', - lifetime=\'%s\' WHERE filename=\'%s\';""" % ( table, - self.status, - self.last_check, - self.role, - self.stage, - self.lifetime, - self.db_key ) - return stmt - stmt = """ -UPDATE %s -SET - size=%d, - md5sum=\'%s\', - role=\'%s\', - lifetime=\'%s\', - last_check=\'%s\', - stage=\'%s\', - status=\'%s\' -WHERE filename=\'%s\'; -""" % ( table, - self.size, - self.md5sum, - self.role, - self.lifetime, - self.last_check, - self.stage, - self.status, - self.db_key ) - return stmt - -def usage(): - print - print 'Usage: %s [-v] [-l | -limit ] ' % sys.argv[0] - print ' Update the 100 last checked entries of FilesMonitoring _ table' - print ' where is the name of the host where this command is run and' - print ' is a nebulous directory name (range: 00 - ff)' - print - print ' * -h: display help' - print ' * -v: be verbose' - print ' * -l , -limit : process no more than entries (default 10)' - print - print ' e.g.: %s [-v] 00' % sys.argv[0] - print ' e.g.: %s [-v] 01' % sys.argv[0] - print ' e.g.: ..' - print ' e.g.: %s [-v] ff' % sys.argv[0] - print - -def process_options(argv): - limit = 10 - directory = None - argc = 1 - while argc < len(argv): - if sys.argv[argc] == '-h': - usage() - sys.exit(0) - elif sys.argv[argc] == '-v': - DbEntry.verbose = True - argc += 1 - elif sys.argv[argc] == '-limit' or sys.argv[argc] == '-l' : - #TODO: Make this POSIX - limit = int(sys.argv[argc+1]) - argc += 2 - elif sys.argv[argc] == '-disk': - disk = sys.argv[argc+1] - argc += 2 - else: - directory = sys.argv[argc] - argc += 1 - if directory is None: - usage() - sys.exit(1) - return limit, directory, disk - -def process_batch(filesMonitoring, disk, table_name, limit): - cursor = filesMonitoring.cursor() - query = 'SELECT filename, subdirname FROM %s ORDER BY last_check LIMIT %d' % ( table_name, - limit) - if DbEntry.verbose: - sys.stderr.write('%s\n' % query) - cursor.execute(query) - rows = cursor.fetchall() - cursor.close() - entries = [ ] - for row in rows: - filename = row[0] - subdirname = row[1] - full_filename = '/export/%s.%s/nebulous/%s/%s/%s' % ( hostname, - disk, - directory, - subdirname, - filename) - entries.append(DbEntry(filename, full_filename)) - for entry in entries: - entry.check() - count_updates = 0 - cursor = filesMonitoring.cursor() - for entry in entries: - query = entry.update(table_name) - count_updates += 1 - cursor.execute(query) - if DbEntry.verbose: - print query - cursor.close() - filesMonitoring.commit() - sys.stderr.write('\t# of updates: %d\n' % count_updates) - -####################################################################################### -if __name__ == '__main__': - if len(sys.argv) == 1: - usage() - sys.exit(1) - userlimit, directory, disk = process_options(sys.argv) - hostname = socket.gethostbyaddr(socket.gethostname())[0].split('.')[0] - table_name = 'neb_%s_%s' % (disk, directory) - sys.stderr.write('Updating: %s %s %s\n' % (hostname, directory, table_name)) - if hostname == 'neverland': - filesMonitoring = MySQLdb.connect( 'localhost', - 'root', 'root', - 'DisksMonitoring' ) - else: - filesMonitoring = MySQLdb.connect( 'localhost', - 'root', '', - 'DisksMonitoring' ) - max_batch_size = 2500 - if userlimit > max_batch_size: - sys.stderr.write('Batching by sets of %d\n' % max_batch_size) - count = 0 - while count < userlimit: - sys.stderr.write('Processing %d out of %d\n' % (count, userlimit) ) - process_batch(filesMonitoring, disk, table_name, max_batch_size) - count += max_batch_size - else: - process_batch(filesMonitoring, disk, table_name, userlimit) - - filesMonitoring.close() Index: trunk/ippm/FilesMonitoring/NodeRelated/update_table_undefined.py =================================================================== --- trunk/ippm/FilesMonitoring/NodeRelated/update_table_undefined.py (revision 35984) +++ (revision ) @@ -1,214 +1,0 @@ -#!/usr/bin/env python - -import MySQLdb -import sys -import socket -import os -import datetime -import hashlib -try: - from ipp.lifetime_templates import Template -except ImportError: - sys.path.append('%s/%s/share/python/' % os.environ['PSCONFDIR'], os.environ['PSCONFIG']) - sys.stderr.write('ImportError') - for folder in sys.path: - sys.stderr.write('\t%s\n' % folder) - from ipp.lifetime_templates import Template - -def md5sum(filename): - f = open(filename, "rb") - content = f.read() - f.close() - sum = hashlib.md5() - sum.update(content) - return sum.hexdigest() - -class DbEntry: - verbose = False - def __init__(self, db_key, full_filename): - self.db_key = db_key - self.full_filename = full_filename - - def __str__(self): - if not DbEntry.verbose and len(self.db_key) > 10: - db_key = '%s...' % self.db_key[:10] - full_filename = '...%s' % self.full_filename[-10:] - else: - db_key = self.db_key - full_filename = self.full_filename - if self.role != 'deleted': - return '[%s]: %s (%s/%s/%s), %d, %s, %s' % (db_key, - full_filename, - self.stage, - self.role, - self.lifetime, - self.size, - self.md5sum, - self.last_check) - else: - return '[%s]: %s (%s)' % (db_key, - full_filename, - self.role) - - def check(self): - try: - self.stage, self.role, self.lifetime = Template.classify(self.db_key) - stat_values = os.stat(self.full_filename) - self.size = stat_values.st_size - try: - self.md5sum = md5sum(self.full_filename) - self.status = 'ok' - except IOError: - self.status = 'error' - self.md5sum = None - except OSError: - self.status = 'deleted' - self.stage, self.role, self.lifetime = Template.classify(self.db_key) - self.last_check = datetime.datetime.now() - - def update(self, table): - if self.status == 'deleted': - self.stage, self.role, self.lifetime = Template.classify(self.db_key) - stmt = """UPDATE %s -SET status=\'%s\', - last_check=\'%s\', - role=\'%s\', - stage=\'%s\', - lifetime=\'%s\' WHERE filename=\'%s\';""" % ( table, - self.status, - self.last_check, - self.role, - self.stage, - self.lifetime, - self.db_key ) - return stmt - stmt = """ -UPDATE %s -SET - size=%d, - md5sum=\'%s\', - role=\'%s\', - lifetime=\'%s\', - last_check=\'%s\', - stage=\'%s\', - status=\'%s\' -WHERE filename=\'%s\'; -""" % ( table, - self.size, - self.md5sum, - self.role, - self.lifetime, - self.last_check, - self.stage, - self.status, - self.db_key ) - return stmt - -def usage(): - print - print 'Usage: %s [-v] [-l | -limit ] [-disk ] ' % sys.argv[0] - print ' Update the 10 last checked entries of FilesMonitoring _ table' - print ' where is the name of the host where this command is run and' - print ' is a nebulous directory name (range: 00 - ff)' - print - print ' * -h: display help' - print ' * -v: be verbose' - print ' * -l , -limit : process no more than entries (default 10)' - print ' * -disk : process disk /export/. (default: 0)' - print - print ' e.g.: %s [-v] 00' % sys.argv[0] - print ' e.g.: %s [-v] 01' % sys.argv[0] - print ' e.g.: ..' - print ' e.g.: %s [-v] ff' % sys.argv[0] - print - -def process_options(argv): - limit = 10 - directory = None - argc = 1 - disk = "0" - while argc < len(argv): - if sys.argv[argc] == '-h': - usage() - sys.exit(0) - elif sys.argv[argc] == '-v': - DbEntry.verbose = True - argc += 1 - elif sys.argv[argc] == '-limit' or sys.argv[argc] == '-l': - limit = int(sys.argv[argc+1]) - argc += 2 - elif sys.argv[argc] == '-disk': - disk = sys.argv[argc+1] - argc += 2 - else: - directory = sys.argv[argc] - argc += 1 - if directory is None: - usage() - sys.exit(1) - return limit, directory, disk - -def process_batch(filesMonitoring, disk, table_name, limit): - cursor = filesMonitoring.cursor() - query = 'SELECT filename, subdirname FROM %s WHERE status=\'undefined\' LIMIT %d' % ( table_name, - limit) - if DbEntry.verbose: - sys.stderr.write('%s\n' % query) - cursor.execute(query) - rows = cursor.fetchall() - cursor.close() - entries = [ ] - for row in rows: - filename = row[0] - subdirname = row[1] - full_filename = '/export/%s.%s/nebulous/%s/%s/%s' % ( hostname, - disk, - directory, - subdirname, - filename) - entries.append(DbEntry(filename, full_filename)) - for entry in entries: - entry.check() - count_updates = 0 - cursor = filesMonitoring.cursor() - for entry in entries: - query = entry.update(table_name) - count_updates += 1 - cursor.execute(query) - if DbEntry.verbose: - print query - cursor.close() - filesMonitoring.commit() - sys.stderr.write('\t# of updates: %d\n' % count_updates) - return count_updates - -####################################################################################### -if __name__ == '__main__': - if len(sys.argv) == 1: - usage() - sys.exit(1) - userlimit, directory, disk = process_options(sys.argv) - hostname = socket.gethostbyaddr(socket.gethostname())[0].split('.')[0] - table_name = 'neb_%s_%s' % (disk, directory) - sys.stderr.write('Updating: %s %s %s\n' % (hostname, directory, table_name)) - if hostname == 'neverland': - filesMonitoring = MySQLdb.connect( 'localhost', - 'root', 'root', - 'DisksMonitoring' ) - else: - filesMonitoring = MySQLdb.connect( 'localhost', - 'ipp', 'ipp', - 'DisksMonitoring' ) - max_batch_size = 2500 - if userlimit > max_batch_size: - sys.stderr.write('Batching by sets of %d\n' % max_batch_size) - count = 0 - while count < userlimit: - sys.stderr.write('\tProcessing: %d out of %d\n' % (count, userlimit) ) - if process_batch(filesMonitoring, disk, table_name, max_batch_size) == 0: - count = userlimit - count += max_batch_size - else: - process_batch(filesMonitoring, disk, table_name, userlimit) - - filesMonitoring.close()