hawk-utils-scripts/monitoring/db/scripts/get_detailed_power.py

import argparse
import numpy as np
from collections import OrderedDict
import os.path


def parse_arguments(args):
	parser = argparse.ArgumentParser(
				description='Produce detailed power usage data for a list of jobids.')
	parser.add_argument('-v', '--verbose', action='store_true',
				help='Show database querries, etc.')
	parser.add_argument('-t', '--interval', action='store', type=float, default=5.0,
				help="Interval between power values in seconds")
	parser.add_argument('jobid', type=parse_jobid,
			nargs='+', 
			help='Job ID such as "2260215" or "2260215.hawk-pbs5"')
    
	return parser.parse_args(args)

def parse_jobid(s):
	import re
	hawkpbs = r'.hawk-pbs5'
	jobid = re.sub(hawkpbs, '', s)
	if not jobid.isdigit():
		raise argparse.ArgumentTypeError(f'invalid job ID "{s}"')
	return jobid


class Power:
	def __init__(self, nodes, interval=-1):
		self.nodes = nodes
		self.epochs = OrderedDict()
		self.first_ts = None
		self.last_ts = None
		self.interval = interval

	@classmethod
	def from_list(cls, data):
		"""Assumes data is a list of tuples (timestamp, value, node)"""
		nodes = list(set([line[2] for line in data]))
		cls = Power(nodes)

		# for now ignore order to nodes
		values = {}
		for l in data:
			ts = l[0]
			if ts not in values:
				values[ts] = []
			# node = l[1]
			power = l[1]
			values[ts].append(power)

		epochs = sorted(values.keys())
		for epoch in epochs:
			cls.insert_epoch(epoch, values[epoch])
		
		return cls

	@classmethod
	def from_db(cls, db, jobid, interval):
		all_list = db.db_to_list(jobid, interval)
		if not all_list:
			raise RuntimeError

		power = cls.from_list(all_list)
		power.set_interval(interval*1000)    # milliseconds

		return power 
 
	def to_file(self, jobid):
		"""Dumps power data to file. Returns filename is succesfull and None if unsucessfull."""
		fname = self.filename(jobid)
		if os.path.exists(fname):
			print("Error: cowardly refusing to overwrite file ", fname)
			return None

		try:
			with open(fname, "w+") as f:
				f.write(self.header())
				f.write(self.body())
		except IOError:
			print("Error: could not write to file ", fname)
			fname = None

		return fname
	
	def set_interval(self, interval):
		self.interval = interval
	
	def insert_epoch(self, ts, values):
		self.epochs[ts] = values
		if not self.first_ts:
			self.first_ts = ts
		self.last_ts = ts

	def header(self):
		hd  = "# all timestamp have unit miliseconds since unix epoch\n"
		hd += "# all power values have unit Watt\n"
		hd += "timestamp,delta_t,head_node_power,avg_node_power,median_node_power,min_node_power,max_node_power,std_dev_node_power"
		# add node names here instead
		hd += ",NO_NODE_NAMES_YET\n"
		return hd

	def body(self):
		_body = ""
		for epoch in self.epochs.items():
			_body += self.pretty_print(self.summarize_epoch(epoch))
		return _body
	
	def summarize_time(self, ts):
		return ts, self.interval

	@staticmethod
	def summarize_values(val):
		values = np.asarray(val)
		head = values[0]
		min, max = values.min(), values.max()
		avg, stddev = values.mean(), values.std()
		median = np.median(values)
		return head, avg, median, min, max, stddev

	def summarize_epoch(self, epoch):
		ts, values = epoch
		return self.summarize_time(ts) \
				+ self.summarize_values(values)
				# + values
	
	@staticmethod
	def pretty_print(args):
		return ",".join(str(a) for a in args) + '\n'

	def filename(self, jobid):
		fname = "detailed_power_{jobid}.hawk-pbs5.{first}-{last}.csv".format(
			jobid=jobid, first=self.first_ts, last=self.last_ts
		)
		return fname
	

class MonitoringDB:
	def __init__(self, verbose):
		self.connection = self.init_db(verbose)

	@staticmethod
	def init_db(verbose):
		import sqlalchemy as db
		_verbose = False #True
		engine = db.create_engine('postgresql://hpc@hawk-monitor4:5432/coe_mon', echo=verbose)
		connection = engine.connect()
		return connection

	def close_db(self):
		return self.connection.close()

	@staticmethod
	def build_query(jobid, interval):
		import sqlalchemy as db
		query_string = """with job as (
		select job_id, starttime, endtime, nodes from jobs where job_id='{jobid}.hawk-pbs5'
	),
	node_series as(
		select n.name, scmcavg.id as series_id from nodes n 
			inner join (select * from label where key='node') l on n.id = l.value::int
			inner join series_cmc_power_racktraynodepoweravg scmcavg on l.id = scmcavg.labels[(
				select pos from  label_key_position 
					where metric_category= 'cmc_power' 
					and metric_name = 'RackTrayNodePowerAvg'  
					and key = 'node'
				)]
		where n.id = any((select nodes from job)::int[])
	)
	select a.time, a.value, ns.name from (
		select 
			time_bucket(extract ('epoch' from '{interval} seconds'::interval)::int*1000, cmcavg.ts) as time, 
			cmcavg.series_id::varchar,
			avg(cmcavg.val) AS value
		from cmc_power_racktraynodepoweravg cmcavg
	where
		ts  <= (select endtime from job) 
		and ts >= (select starttime from job)
		and series_id = Any(select series_id from node_series)
	group by time, cmcavg.series_id order by time desc) a
	inner join node_series ns on a.series_id::int = ns.series_id;
	"""
		return db.text(query_string.format(jobid=jobid, interval=interval))

	def db_to_list(self, jobid, interval):
		query = self.build_query(jobid, interval)
		return self.connection.execute(query).fetchall()

	def db_to_pf(self, jobid, inerval):
		query = self.build_query(jobid, interval)
		return pd.read_sql(query, con=self.connection)


class App:
	def __init__(self, config):
		self.config = config
		self.db = MonitoringDB(self.config.verbose)

	def run_all(self):
		for jobid in self.config.jobid:
			try:
				power = Power.from_db(self.db, jobid, self.config.interval)
			except RuntimeError:
				print('No data found for job ID "{}"'.format(jobid))
				continue

			fn = power.to_file(jobid)
			if fn:
				print('Created file {fn}'.format(fn=fn))


if __name__ == "__main__":
	import sys
	config = parse_arguments(sys.argv[1:])

	main = App(config)
	main.run_all()