Index: trunk/ippm/FilesMonitoring/Supervision/cluster_status.py =================================================================== --- trunk/ippm/FilesMonitoring/Supervision/cluster_status.py (revision 35984) +++ (revision ) @@ -1,343 +1,0 @@ -#!/usr/bin/env python - -import sys -import MySQLdb -import subprocess -import _mysql_exceptions -import re - -toGigs = 1024.**3 -toGigsFromK = 1024.**2 - -bin_directory = "%s/%s/bin" % (os.environ['PSCONFDIR'], os.environ['PSCONFIG']) -share_directory = "%s/%s/share/filemon/" % (os.environ['PSCONFDIR'], os.environ['PSCONFIG']) - -def usage(): - sys.stderr.write('Usage: cluster_status.py [-sql] \n') - sys.stderr.write('Usage: cluster_status.py [-h|-help|--help]\n') - sys.stderr.write('\n') - sys.stderr.write('\t Nebulous disk usage about \n') - sys.stderr.write('\t is the usual node name, e.g. ipp008, ..., ipp064\n') - sys.stderr.write('\t can be:\n') - sys.stderr.write('\t\t * mhpcc: all storage nodes at mhpcc\n') - sys.stderr.write('\t\t * set0: ipp004, ipp005, ipp006, ipp007, ipp008, ipp009, ipp060, ipp061, ipp062, ipp063, ipp064, ipp065, ipp066.0\n') - sys.stderr.write('\t\t * set1: ipp01x\n') - sys.stderr.write('\t\t * set2: ipp02x\n') - sys.stderr.write('\t\t * set3: ipp03x\n') - sys.stderr.write('\t\t * set4: ipp04x\n') - sys.stderr.write('\t\t * set5: ipp05x\n') - sys.stderr.write('\t\t * atrc: ippb0x (all 3 partitions for each node)\n') - sys.stderr.write('\t\t * stcsi: stcsi0x (all 3 partitions for each node)\n') - sys.stderr.write('\t\t * all: all nodes\n') - sys.stderr.write('\n') - sys.stderr.write('\t\t [-sql] Check if the mysql server is running on the nodes\n') - -def process_arguments(argv): - sql_check = False - if len(argv) == 1: - usage() - sys.exit(1) - elif argv[1] == '-h': - usage() - sys.exit(1) - elif argv[1] == '-help': - usage() - sys.exit(1) - elif argv[1] == '--help': - usage() - sys.exit(1) - elif argv[1] == '-sql': - sql_check = True - elif argv[1] == 'mhpcc': - nodesFilename = '%s/ippMhpccStorageNodes' % share_directory - nodesFile = open(nodesFilename) - nodes = [] - for line in nodesFile: - nodes.append(line[:-1] + ".0") - elif argv[1] == 'set0': - nodes = [ "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", ] - elif argv[1] == 'set1': - nodes = [ "ipp010.0", "ipp011.0", "ipp012.0", "ipp013.0", "ipp014.0", - "ipp015.0", "ipp016.0", "ipp017.0", "ipp018.0", "ipp019.0", ] - elif argv[1] == 'set2': - nodes = [ "ipp020.0", "ipp021.0", "ipp023.0", "ipp024.0", - "ipp025.0", "ipp026.0", "ipp027.0", "ipp028.0", "ipp029.0", ] - elif argv[1] == 'set3': - nodes = [ "ipp030.0", "ipp031.0", "ipp032.0", "ipp033.0", "ipp034.0", - "ipp035.0", "ipp036.0", "ipp037.0", "ipp038.0", "ipp039.0", ] - elif argv[1] == 'set4': - nodes = [ "ipp040.0", "ipp041.0", "ipp042.0", "ipp043.0", "ipp044.0", - "ipp045.0", "ipp046.0", "ipp047.0", "ipp048.0", "ipp049.0", ] - elif argv[1] == 'set5': - nodes = [ "ipp050.0", "ipp051.0", "ipp052.0", "ipp053.0", "ipp054.0", - "ipp055.0", "ipp056.0", "ipp057.0", "ipp058.0", "ipp059.0", ] - elif argv[1] == 'ippb': - nodes = [ "ippb00.0", "ippb00.1", "ippb00.2", - "ippb01.0", "ippb01.1", "ippb01.2", - "ippb02.0", "ippb02.1", "ippb02.2", - "ippb03.0", "ippb03.1", "ippb03.2", ] - elif argv[1] == 'stsci': - nodes = [ "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", ] - elif argv[1] == 'all': - nodesFilename = '%s/ippMhpccStorageNodes' % share_directory - nodesFile = open(nodesFilename) - nodes = [] - for line in nodesFile: - nodes.append(line[:-1] + ".0") - 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", ]) - 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", ] - else: - nodes = [ ] - for node in argv[1:]: - nodes.append((node + ".0")) - return (sql_check, nodes) - -def get_nebdf_output(): - nebdf_output = dict() - p = subprocess.Popen( ['neb-df', '--names', ], - stderr = subprocess.PIPE, - stdout = subprocess.PIPE) - if p.wait()!=0: - logger.error('Error while running neb-df') - for line in p.stderr: - logger.error(line) - sys.exit(3) - for line in p.stdout: - if not line.startswith('Filesystem') and not line.startswith('/summmary') and not line.startswith(' '): - line = re.sub('\s+', ':', line[:-1]) - (node, available, used, free, percentage, mountpoint) = line.split(':') - nebdf_output[node] = (float(available)/toGigsFromK, float(used)/toGigsFromK, float(free)/toGigsFromK) - return nebdf_output - -def is_mysql_server_running(node): - try: - MySQLdb.connect(node, 'ippuser', 'ipp', 'DisksMonitoring') - return "mysql server: Running" - except _mysql_exceptions.OperationalError: - return "mysql server: Stopped" - -def is_update_running(node): - p = subprocess.Popen( ['%s/node_status.py' % bin_directory, node], - stderr = subprocess.PIPE, - stdout = subprocess.PIPE) - if p.wait()!=0: - return "Can't tell if node's running update/refresh (ssh error for %s)" % node - for line in p.stdout: - return " ".join(line[:-1].split(' ')[1:]) - -if __name__ == '__main__': - (sql_check, nodes) = process_arguments(sys.argv) - nebdf = get_nebdf_output() - dms_db = MySQLdb.connect('ippc11', 'ippuser', 'ipp', - 'DiskMonitoringSummary') - dms_cursor = dms_db.cursor() - aggregate_by_role = dict() - aggregate_by_stage = dict() - nodes_sizes = dict() - aggregate = dict() - raw_aggregate = dict() - aggregate_percentage = 0. - for node in nodes: - try: - query = "SELECT SUM(counts) FROM CurrentDiskUsage WHERE node='%s' AND status!='deleted'" % node - dms_cursor.execute(query) - total = dms_cursor.fetchall()[0][0] - #print total, query - query = "SELECT SUM(counts) FROM CurrentDiskUsage WHERE node='%s' AND status!='undefined' AND status!='deleted'" % node - dms_cursor.execute(query) - defined = dms_cursor.fetchall()[0][0] - #print defined, query - percentage_done = float(defined) / float(total) - # - try: - query = "SELECT SUM(sizes) FROM CurrentDiskUsage WHERE node='%s' AND status!='undefined' AND status!='deleted'" % node - dms_cursor.execute(query) - nodes_sizes[node] = float(dms_cursor.fetchall()[0][0]) / percentage_done / toGigs - except TypeError, e: - print e - node_name = node.split('.')[0] - if sql_check: - print 'Node %s: %5.1f%% done | %20s | %40s' % (node, percentage_done * 100., - is_mysql_server_running(node_name), - is_update_running(node_name)) - else: - print 'Node %s: %5.1f%% done' % (node, percentage_done * 100.) - aggregate_percentage += percentage_done - query = "SELECT role, SUM(sizes) FROM CurrentDiskUsage WHERE node='%s' AND status!='undefined' AND status!='deleted' GROUP BY role" % node - dms_cursor.execute(query) - for row in dms_cursor.fetchall(): - if row[0] != 'undefined': - try: - aggregate_by_role[row[0]] += (float(row[1])/percentage_done/toGigs) - except KeyError: - aggregate_by_role[row[0]] = (float(row[1])/percentage_done/toGigs) - query = "SELECT stage, SUM(sizes) FROM CurrentDiskUsage WHERE node='%s' AND status!='undefined' AND status!='deleted' GROUP BY stage" % node - dms_cursor.execute(query) - for row in dms_cursor.fetchall(): - if row[0] != 'undefined': - try: - aggregate_by_stage[row[0]] += (float(row[1])/percentage_done/toGigs) - except KeyError: - aggregate_by_stage[row[0]] = (float(row[1])/percentage_done/toGigs) - query = "SELECT stage, role, SUM(sizes) FROM CurrentDiskUsage WHERE node='%s' AND status!='undefined' AND status!='deleted' GROUP BY stage, role" % node - dms_cursor.execute(query) - for row in dms_cursor.fetchall(): - if row[0] != 'undefined': - try: - aggregate[row[0]][row[1]] += (float(row[2])/percentage_done/toGigs) - raw_aggregate[row[0]][row[1]] += float(row[2])/toGigs - except KeyError: - try: - aggregate[row[0]][row[1]] = (float(row[2])/percentage_done/toGigs) - raw_aggregate[row[0]][row[1]] = float(row[2])/toGigs - except KeyError: - aggregate[row[0]] = dict() - raw_aggregate[row[0]] = dict() - aggregate[row[0]][row[1]] = (float(row[2])/percentage_done/toGigs) - raw_aggregate[row[0]][row[1]] = float(row[2])/toGigs - except TypeError, e: - print 'Node %s: being refreshed' % (node) - print e - aggregate_percentage /= len(nodes) - - print - print "### By role (in gigs / relative % / in teras)" - total = 0. - for size in aggregate_by_role.itervalues(): - total += size - for role, size in aggregate_by_role.iteritems(): - print "%10s: %10.1f %10.1f%% %10.1f" % (role, size, size/total*100., size/1024.) - print "%10s: %10.1f %11s %10.1f" % ("Total", total, "", total/1024.) - - print - print "### By stage (in gigs / relative % / in teras)" - total = 0. - for size in aggregate_by_stage.itervalues(): - total += size - for stage, size in aggregate_by_stage.iteritems(): - print "%10s: %10.1f %10.1f%% %10.1f" % (stage, size, size/total*100., size/1024.) - print "%10s: %10.1f %11s %10.1f" % ("Total", total, "", total/1024.) - print - - if aggregate_percentage == 1.: - print - print "### Interpolated Sizes (in gigs), if all nodes were all 100% done" - header = '%10s' % '' - roles = dict() - for stage in aggregate.iterkeys(): - header = '%s%10s' % (header, stage) - for role in aggregate[stage].iterkeys(): - roles[role] = '' - print header - for role in roles.iterkeys(): - line = '%10s' % role - for stage in aggregate.iterkeys(): - try: - line = '%10s%10.1f' % (line, aggregate[stage][role]) - except KeyError: - line = '%10s%10s' % (line, '-') - print line - - print - print "### Sizes (in gigs)" - header = '%10s' % '' - roles = dict() - for stage in aggregate.iterkeys(): - header = '%s%10s' % (header, stage) - for role in aggregate[stage].iterkeys(): - roles[role] = '' - print header - for role in roles.iterkeys(): - line = '%10s' % role - for stage in aggregate.iterkeys(): - try: - line = '%10s%10.1f' % (line, raw_aggregate[stage][role]) - except KeyError: - line = '%10s%10s' % (line, '-') - print line - - print - print "### By node: Mine \t neb-df \tDifference (compared to disk size)\t\"Non-nebulous\" space" - neb_total = 0. - nebdf_total = 0. - nonneb_total = 0. - available = 0. - for node in nodes: - try: - try: - neb_total += nodes_sizes[node] - except KeyError, e: - print 'No nodes_sizes[node]' - raise e - try: - nebdf_total += nebdf[node][1] - except KeyError, e: - print 'No nodes_nebdf[node]' - raise e - nonneb_total += (nebdf[node][1]-nodes_sizes[node]) - available += nebdf[node][0] - print '%10s: %10.2f %10.2f %10.2f%% %10.2f' % ( node, nodes_sizes[node], nebdf[node][1], - (nodes_sizes[node]-nebdf[node][1])/nebdf[node][0]*100., - nebdf[node][1]-nodes_sizes[node] - ) - except KeyError: - print '%10s: Incomplete data' % node - print "---------------------------------------------------" - print '%10s: %10.2f %10.2f %10.2f%% %10.2f' % ( "Total", neb_total, nebdf_total, - (neb_total-nebdf_total)/available*100., - nonneb_total ) - exclude_nodes = ['ipp004.0', 'ipp005.0', 'ipp064.0', ] - for node in exclude_nodes: - if node in nodes_sizes: - neb_total -= nodes_sizes[node] - nebdf_total -= nebdf[node][1] - nonneb_total -= (nebdf[node][1]-nodes_sizes[node]) - available -= nebdf[node][0] - print '%10s: %10.2f %10.2f %10.2f%% %10.2f' % ( "Total*", neb_total, nebdf_total, - (neb_total-nebdf_total)/available*100., - nonneb_total ) - print "Total* = Total excluding nodes:", exclude_nodes - Index: trunk/ippm/FilesMonitoring/Supervision/node_status.py =================================================================== --- trunk/ippm/FilesMonitoring/Supervision/node_status.py (revision 35984) +++ (revision ) @@ -1,122 +1,0 @@ -#!/usr/bin/env python - -import sys -import warnings -warnings.simplefilter("ignore") # Importing paramiko displays the following warning: -#/home/panstarrs/schastel/local/lib/python2.5/site-packages/pycrypto-2.5-py2.5-linux-x86_64.egg/Crypto/Util/number.py:57: PowmInsecureWarning: Not using mpz_powm_sec. You should rebuild using libgmp >= 5 to avoid timing attack vulnerability. -# _warn("Not using mpz_powm_sec. You should rebuild using libgmp >= 5 to avoid timing attack vulnerability.", PowmInsecureWarning) -import paramiko -warnings.simplefilter("always") -import datetime - -bin_directory = "%s/%s/bin" % (os.environ['PSCONFDIR'], os.environ['PSCONFIG']) -share_directory = "%s/%s/share/filemon/" % (os.environ['PSCONFDIR'], os.environ['PSCONFIG']) -node_directory = '/export/%s/ipp_filemon/' - -def process_arguments(argv): - if len(argv) == 1: - usage() - sys.exit(1) - elif argv[1] == 'all': - nodesFilename = '%s/ippMhpccStorageNodes' % share_directory - nodesFile = open(nodesFilename) - nodes = [] - for line in nodesFile: - nodes.append(line[:-1] + ".0") - elif argv[1] == 'set0': - nodes = [ "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", ] - elif argv[1] == 'set1': - nodes = [ "ipp010.0", "ipp011.0", "ipp012.0", "ipp013.0", "ipp014.0", - "ipp015.0", "ipp016.0", "ipp017.0", "ipp018.0", "ipp019.0", ] - elif argv[1] == 'set2': - nodes = [ "ipp020.0", "ipp021.0", "ipp023.0", "ipp024.0", - "ipp025.0", "ipp026.0", "ipp027.0", "ipp028.0", "ipp029.0", ] - elif argv[1] == 'set3': - nodes = [ "ipp030.0", "ipp031.0", "ipp032.0", "ipp033.0", "ipp034.0", - "ipp035.0", "ipp036.0", "ipp037.0", "ipp038.0", "ipp039.0", ] - elif argv[1] == 'set4': - nodes = [ "ipp040.0", "ipp041.0", "ipp042.0", "ipp043.0", "ipp044.0", - "ipp045.0", "ipp046.0", "ipp047.0", "ipp048.0", "ipp049.0", ] - elif argv[1] == 'set5': - nodes = [ "ipp050.0", "ipp051.0", "ipp052.0", "ipp053.0", "ipp054.0", - "ipp055.0", "ipp056.0", "ipp057.0", "ipp058.0", "ipp059.0", ] - else: - nodes = [] - for node in argv[1:]: - nodes.append(node + ".0") - return nodes - -def get_seconds_since_last_message(ssh, node): - logfilename = ('%s/%s' % (node_directory, '/logs/%s_refresh.log') % (node, node)) - stdin, stdout, stderr = ssh.exec_command('tail -n 1 %s' % (logfilename)) - for line in stdout: - try: - last_message_date = datetime.datetime.strptime(line[:-1].split(' ')[0], '%Y-%m-%dT%H:%M:%S') - except ValueError: - return 999999 - now = datetime.datetime.now() - try: - td = now - last_message_date - except UnboundLocalError: #local variable 'last_message_date' migh tbe referenced before assignment - return 999999 - return td.seconds + td.days * 24 * 3600 - -def get_crontab_status(ssh): - stdin, stdout, stderr = ssh.exec_command('crontab -l | grep refresh_host.sh') - crontab_installed = False - next_run = None - for line in stdout: - crontab_installed = True - if 'nice' not in line: - print 'In crontab but NOT nice' - schedule = line.split(' ') - next_run = schedule[1] + ':' + schedule[0] - return (crontab_installed, next_run) - -def guess_if_update_or_refresh_is_running(ssh): - stdin, stdout, stderr = ssh.exec_command('ps waux | grep update_host_undefined.sh | grep -v grep') - update_is_running = False - for line in stdout: - update_is_running = True - stdin, stdout, stderr = ssh.exec_command('ps waux | grep refresh_table.py | grep -v grep') - refresh_is_running = False - for line in stdout: - refresh_is_running = True - if update_is_running: - if refresh_is_running: - message = "Both Update and Refresh are running" - else: - message = "Update is running" - else: - if refresh_is_running: - message = "Refresh is running" - else: - message = "Nothing is running" - return message - -if __name__ == '__main__': - nodes = process_arguments(sys.argv) - show_crontab = False - for node in nodes: - node_name = node.split('.')[0] - ssh = paramiko.SSHClient() - ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) - ssh.connect(node_name) - seconds_since_last_message = get_seconds_since_last_message(ssh, node) - (is_crontab_installed, next_run) = get_crontab_status(ssh) - print '%s: %s' % (node, guess_if_update_or_refresh_is_running(ssh)) - # if seconds_since_last_message > 1000 and not is_crontab_installed: - # print 'Node %s: Not installed in crontab and last run was %d seconds ago' % (node, seconds_since_last_message) - # else: - # if is_crontab_installed: - # if seconds_since_last_message < 1000: - # msg = "OK" - # else: - # msg = "Stopped or crashed? (next run at %s)" % next_run - # else: - # msg= "Running (but crontab is not installed)" - # show_crontab = True - # print 'Node %s: %s' % (node, msg) - # if show_crontab: - # print '58 13 * * * /usr/bin/nice -n 19 /home/panstarrs/schastel/dev/FilesMonitoring/refresh_host.sh'