Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Store queue times for pending jobs (raw data) #1

Open
wants to merge 4 commits into
base: master
Choose a base branch
from
Open

Store queue times for pending jobs (raw data) #1

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
65a5c18
Break out strftime() to retain 'now' var
imgrant Nov 15, 2019
9e2ac68
Add queue time for pending jobs (per job)
imgrant Nov 15, 2019
4c2e1ce
Write per-job queue times to a different retention policy
imgrant Nov 15, 2019
d173391
Fix: select from correct retention policy
imgrant Nov 19, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 22 additions & 0 deletions cq_examples.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,22 @@
CREATE RETENTION POLICY "1week" ON <database> DURATION 7d REPLICATION 1

CREATE CONTINUOUS QUERY "cq_time_pending_1day_partition" ON <database>
BEGIN
SELECT MAX("jobs_time_pending") INTO "partition_jobs_time_pending_daily" FROM "1week"."partition_jobs_time_pending" GROUP BY time(1d),*
END

CREATE CONTINUOUS QUERY "cq_time_pending_1day_group" ON <database>
BEGIN
SELECT MAX("jobs_time_pending") INTO "group_jobs_time_pending_daily" FROM "1week"."group_jobs_time_pending" GROUP BY time(1d),*
END


CREATE CONTINUOUS QUERY "cq_time_pending_1week_partition" ON <database>
BEGIN
SELECT MAX("jobs_time_pending") INTO "partition_jobs_time_pending_weekly" FROM "partition_jobs_time_pending_daily" GROUP BY time(7d),*
END

CREATE CONTINUOUS QUERY "cq_time_pending_1week_group" ON <database>
BEGIN
SELECT MAX("jobs_time_pending") INTO "group_jobs_time_pending_weekly" FROM "group_jobs_time_pending_daily" GROUP BY time(7d),*
END
24 changes: 21 additions & 3 deletions main.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,7 @@
metrics['partition']['jobs_pending'] = {}
metrics['partition']['queue_time'] = {}
metrics['partition']['queue_jobs'] = {}
metrics['partition']['jobs_time_pending'] = {}

metrics['user'] = {}
metrics['user']['cpu_usage'] = {}
Expand All @@ -82,6 +83,7 @@
metrics['group']['jobs_pending'] = {}
metrics['group']['queue_time'] = {}
metrics['group']['queue_jobs'] = {}
metrics['group']['jobs_time_pending'] = {}

if config['user_lookup']:
metrics['ldap_attrib'] = {}
Expand All @@ -97,7 +99,8 @@
user_groups = {}
user_ldap = {}

now = datetime.datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%SZ')
now = datetime.datetime.utcnow()
now_str = now.strftime('%Y-%m-%dT%H:%M:%SZ')

# Setup data structures, with stats set to 0
for part in partitions.keys() + ['ALL']:
Expand All @@ -122,6 +125,7 @@
metrics['partition']['jobs_pending'][part] = 0
metrics['partition']['queue_time'][part] = 0
metrics['partition']['queue_jobs'][part] = 0
metrics['partition']['jobs_time_pending'][part] = {}

for group in groups:
metrics['group']['cpu_usage'][group] = 0
Expand All @@ -131,6 +135,7 @@
metrics['group']['jobs_pending'][group] = 0
metrics['group']['queue_time'][group] = 0
metrics['group']['queue_jobs'][group] = 0
metrics['group']['jobs_time_pending'][group] = {}

members = grp.getgrnam(group)[3]
for user in members:
Expand Down Expand Up @@ -268,15 +273,19 @@
metrics['ldap_attrib']['queue_time'][user_ldap[user]] = (float(metrics['ldap_attrib']['queue_time'][user_ldap[user]] + queue_time)) / metrics['ldap_attrib']['queue_jobs'][user_ldap[user]]

elif job['job_state'] == 'PENDING':
time_pending = (now - datetime.datetime.utcfromtimestamp(job['submit_time'])).total_seconds()
metrics['partition']['jobs_pending']['ALL'] += 1
metrics['partition']['jobs_time_pending']['ALL'][job['job_id']] = time_pending # Not sure 'ALL' is really needed for this metric, but for consistency do it anyway
for partition in job['partition'].split(','):
metrics['partition']['jobs_pending'][partition] += 1
metrics['partition']['jobs_time_pending'][partition][job['job_id']] = time_pending

metrics['user']['jobs_pending'][user] += 1

if user in user_groups:
for group in user_groups[user]:
metrics['group']['jobs_pending'][group] += 1
metrics['group']['jobs_time_pending'][group][job['job_id']] = time_pending

if config['user_lookup']:
metrics['ldap_attrib']['jobs_pending'][user_ldap[user]] += 1
Expand All @@ -286,6 +295,15 @@
for reading in ['cpu_total', 'cpu_usage', 'cpu_usage_pc', 'gpu_total', 'gpu_usage', 'gpu_usage_pc', 'mem_total', 'mem_usage', 'mem_usage_pc', 'jobs_running', 'jobs_pending', 'queue_time']:
if reading in metrics[grouping] and len(metrics[grouping][reading]) > 0:
for key in metrics[grouping][reading].keys():
payload.append({'measurement': '%s_%s' % (grouping, reading), 'time': now, 'fields': {reading: float(metrics[grouping][reading][key])}, 'tags': {grouping: key}})

payload.append({'measurement': '%s_%s' % (grouping, reading), 'time': now_str, 'fields': {reading: float(metrics[grouping][reading][key])}, 'tags': {grouping: key}})
client.write_points(payload, database=config['influxdb_database'])

payload = []
for grouping in ['partition', 'group']:
for reading in ['jobs_time_pending']:
if reading in metrics[grouping] and len(metrics[grouping][reading]) > 0:
for key in metrics[grouping][reading].keys():
if len(metrics[grouping][reading][key]) > 0:
for jid in metrics[grouping][reading][key].keys():
payload.append({ 'measurement': '{grouping}_{reading}'.format(grouping=grouping,reading=reading), 'time': now_str, 'fields': {reading: float(metrics[grouping][reading][key][jid])}, 'tags': {'job_id': jid, grouping: key} })
client.write_points(payload, database=config['influxdb_database'], retention_policy='1week')