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

#!/usr/bin/env python3

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, node, value)"""
		idx_ts = 0; idx_node = 1; idx_value = 2
		nodes = list(set([line[idx_node] for line in data]))
		cls = Power(nodes)

		# for now ignore order to nodes
		values = {}
		for l in data:
			ts = l[idx_ts]
			if ts not in values:
				values[ts] = []
			# node = l[1]
			power = l[idx_value]
			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, header=""):
		"""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(header + 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:
	QUERY_STRING_HAWK = """
with _job as (
	select job_id, starttime, endtime, nodes from job where job_id='{jobid}.hawk-pbs5'
),
node_series as(
	select n.name, scmcavg.id as series_id from node 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, ns.name, a.value 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;
	"""
	
	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()

	@classmethod
	def build_query(cls, jobid, interval):
		"""Query taken from https://kb.hlrs.de/monitoring/index.php/TimescaleDB_-_Query_Guidelines#Power_by_Job_Query"""
		import sqlalchemy as db
		query_string = cls.QUERY_STRING_HAWK
		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)

	@staticmethod
	def warnings(config):
		warn = ""
		if config.interval < 5: 
			warn += '# Warning: interval<5 is very small and may lead to data gaps.'
		return warn
	
	def run_all(self):
		warnings = self.warnings(self.config)
		if warnings:
			print(warnings)

		header = f"# {config.datetime}: {config.cmd}\n"
		if warnings:
			header +=  f"{warnings}\n"
		header += "#\n"

		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, header)
			if fn:
				print('Created file {fn}'.format(fn=fn))


if __name__ == "__main__":
	import sys
	from datetime import datetime
	config = parse_arguments(sys.argv[1:])
	config.cmd = " ".join(sys.argv)
	config.datetime = f"{datetime.now()}"

	main = App(config)
	main.run_all()