Skip to content
This repository was archived by the owner on Dec 5, 2018. It is now read-only.

Commit 3278c7a

Browse files
author
Walter Shands
authored
Merge pull request #30 from BD2KGenomics/feature/primary-key-switch
General monitor changes
2 parents 2287962 + 5f1a773 commit 3278c7a

File tree

3 files changed

+75
-88
lines changed

3 files changed

+75
-88
lines changed

‎action.py‎

Lines changed: 5 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -18,9 +18,10 @@
1818
# import calendar
1919
# import click
2020
# TEST database call
21-
from sqlalchemy import create_engine, MetaData, String, Table, Float, Column, select
2221
import logging
2322
from database import db, login_db, login_manager, User
23+
from monitordb_lib import luigiDBInit
24+
from sqlalchemy import select, desc
2425

2526
actionbp = Blueprint('actionbp', 'actionbp')
2627

@@ -44,34 +45,9 @@ def login():
4445
@login_required
4546
@cross_origin()
4647
def get_action_service():
47-
db = create_engine(
48-
'postgresql://{}:{}@db/monitor'.format(os.getenv("POSTGRES_USER"), os.getenv("POSTGRES_PASSWORD")), echo=False)
49-
conn = db.connect()
50-
metadata = MetaData(db)
51-
luigi = Table('luigi', metadata,
52-
Column("luigi_job", String(100), primary_key=True),
53-
Column("status", String(20)),
54-
Column("submitter_specimen_id", String(100)),
55-
Column("specimen_uuid", String(100)),
56-
Column("workflow_name", String(100)),
57-
Column("center_name", String(100)),
58-
Column("submitter_donor_id", String(100)),
59-
Column("consonance_job_uuid", String(100)),
60-
Column("submitter_donor_primary_site", String(100)),
61-
Column("project", String(100)),
62-
Column("analysis_type", String(100)),
63-
Column("program", String(100)),
64-
Column("donor_uuid", String(100)),
65-
Column("submitter_sample_id", String(100)),
66-
Column("submitter_experimental_design", String(100)),
67-
Column("submitter_specimen_type", String(100)),
68-
Column("workflow_version", String(100)),
69-
Column("sample_uuid", String(100)),
70-
Column("start_time", String(100)),
71-
Column("last_updated", String(100))
72-
)
73-
select_query = select([luigi]).distinct("last_updated").order_by("last_updated")
74-
select_result = conn.execute(select_query)
48+
monitordb_connection, monitordb_table = luigiDBInit()
49+
select_query = select([monitordb_table]).order_by(desc("last_updated"))
50+
select_result = monitordb_connection.execute(select_query)
7551
result_list = [dict(row) for row in select_result]
7652
return jsonify(result_list)
7753

‎luigi-interface/monitor.py‎

Lines changed: 34 additions & 59 deletions
Original file line numberDiff line numberDiff line change
@@ -1,22 +1,23 @@
1-
# Alex Hancock, UCSC CGL
2-
#
3-
# Luigi Monitor
4-
51
import boto
62
import json
73
import os
4+
import re
85
import subprocess
96
import sys
107
import urllib2
118

12-
from sqlalchemy import create_engine, MetaData, Table, Column, String, select, and_
9+
from sqlalchemy import select, and_
1310
from datetime import datetime
1411

12+
# Add parent directory to get luigidb init
13+
sys.path.append( os.path.dirname( os.path.dirname( os.path.abspath(__file__) ) ) )
14+
from monitordb_lib import luigiDBInit
15+
1516

1617
def get_touchfile(bucket_name, touchfile_name):
1718
s3 = boto.connect_s3()
1819
bucket = s3.get_bucket(bucket_name, validate=False)
19-
print "GOT S3 BUCKET"
20+
print "GOT S3 BUCKET"
2021

2122
key = bucket.new_key(touchfile_name)
2223
print "CREATED NEW S3 KEY"
@@ -87,46 +88,20 @@ def get_consonance_status(consonance_uuid):
8788
return json.loads(status_text)
8889

8990

90-
# This was exported to a method to avoid duplication
91-
# for both the creation and update timestamps
9291
def format_consonance_timestamp(consonance_timestamp):
92+
# This was exported to a method to avoid duplication
93+
# for both the creation and update timestamps
9394
datetime_obj = datetime.strptime(consonance_timestamp, '%Y-%m-%dT%H:%M:%S.%f+0000')
9495
return datetime.strftime(datetime_obj, '%Y-%m-%d %H:%M')
9596

9697

97-
#
98-
# Database initialization, creation if table doesn't exist
99-
#
100-
# Change echo to True to show SQL code... unnecessary for production
101-
#
102-
db = create_engine('postgresql://{}:{}@db/{}'.format(os.getenv("POSTGRES_USER"), os.getenv("POSTGRES_PASSWORD"), os.getenv("POSTGRES_DB")), echo=False)
103-
conn = db.connect()
104-
metadata = MetaData(db)
105-
luigi = Table('luigi', metadata,
106-
Column("luigi_job", String(100), primary_key=True),
107-
Column("status", String(20)),
108-
Column("submitter_specimen_id", String(100)),
109-
Column("specimen_uuid", String(100)),
110-
Column("workflow_name", String(100)),
111-
Column("center_name", String(100)),
112-
Column("submitter_donor_id", String(100)),
113-
Column("consonance_job_uuid", String(100)),
114-
Column("submitter_donor_primary_site", String(100)),
115-
Column("project", String(100)),
116-
Column("analysis_type", String(100)),
117-
Column("program", String(100)),
118-
Column("donor_uuid", String(100)),
119-
Column("submitter_sample_id", String(100)),
120-
Column("submitter_experimental_design", String(100)),
121-
Column("submitter_specimen_type", String(100)),
122-
Column("workflow_version", String(100)),
123-
Column("sample_uuid", String(100)),
124-
Column("start_time", String(100)),
125-
Column("last_updated", String(100)))
126-
127-
if not db.dialect.has_table(db, luigi):
128-
luigi.create()
98+
def validateConsonanceUUID(consonance_uuid):
99+
# Return if consonance uuid only contains hex characters and "-"
100+
uuid_pattern = re.compile("[a-f,A-F,0-9,-]+")
101+
return bool(uuid_pattern.match(consonance_uuid))
102+
129103

104+
monitordb_connection, monitordb_table = luigiDBInit()
130105
jobList = get_job_list()
131106

132107
for job in jobList:
@@ -177,16 +152,16 @@ def format_consonance_timestamp(consonance_timestamp):
177152
# use the Consonance job uuid instead of the Luigi job id because
178153
# Luigi sometimes reuses job ids for different runs
179154
# The Consonance job uuid is always unique
180-
# select_query = select([luigi]).where(luigi.c.luigi_job == job)
155+
# select_query = select([monitordb_table]).where(monitordb_table.c.luigi_job == job)
181156
job_uuid = jsonMetadata['consonance_job_uuid']
182157
print "QUERYING DB FOR JOB UUID:", job_uuid
183-
select_query = select([luigi]).where(luigi.c.consonance_job_uuid == job_uuid)
184-
select_result = query_to_list(conn.execute(select_query))
158+
select_query = select([monitordb_table]).where(monitordb_table.c.consonance_job_uuid == job_uuid)
159+
select_result = query_to_list(monitordb_connection.execute(select_query))
185160
print "JOB RESULT:", select_result
186161
if len(select_result) == 0:
187162
try:
188-
#ins_query = luigi.insert().values(luigi_job=job,
189-
ins_query = luigi.insert().values(luigi_job=job_uuid,
163+
#ins_query = monitordb_table.insert().values(luigi_job=job,
164+
ins_query = monitordb_table.insert().values(luigi_job=job,
190165
submitter_specimen_id=jsonMetadata['submitter_specimen_id'],
191166
specimen_uuid=jsonMetadata['specimen_uuid'],
192167
workflow_name=jsonMetadata['workflow_name'],
@@ -203,7 +178,7 @@ def format_consonance_timestamp(consonance_timestamp):
203178
submitter_specimen_type=jsonMetadata['submitter_specimen_type'],
204179
workflow_version=jsonMetadata['workflow_version'],
205180
sample_uuid=jsonMetadata['sample_uuid'])
206-
exec_result = conn.execute(ins_query)
181+
exec_result = monitordb_connection.execute(ins_query)
207182
except Exception as e:
208183
print >>sys.stderr, e.message, e.args
209184
print "Dumping jsonMetadata to aid debug:\n", jsonMetadata
@@ -218,21 +193,21 @@ def format_consonance_timestamp(consonance_timestamp):
218193
# consonance status using job.consonance_uuid
219194
# update that job using the information from status return
220195
#
221-
select_query = select([luigi])
222-
select_result = conn.execute(select_query)
196+
select_query = select([monitordb_table])
197+
select_result = monitordb_connection.execute(select_query)
223198
result_list = [dict(row) for row in select_result]
224199
for job in result_list:
225200
try:
226201
job_name = job['luigi_job']
227202
job_uuid = job['consonance_job_uuid']
228203

229-
if job_uuid == "no consonance id in test mode":
204+
if not validateConsonanceUUID(job_uuid):
230205
# Skip test mode Consonance ID's
231206
# and force next job
232207
print "\nTest ID, skipping"
233208

234-
stmt = luigi.delete().where(luigi.c.luigi_job == job_name)
235-
exec_result = conn.execute(stmt)
209+
stmt = monitordb_table.delete().where(monitordb_table.c.luigi_job == job_name)
210+
exec_result = monitordb_connection.execute(stmt)
236211
else:
237212
# Consonace job id is real
238213
print "\nJOB NAME:", job_uuid
@@ -247,20 +222,20 @@ def format_consonance_timestamp(consonance_timestamp):
247222
print "CREATED:", created
248223
print "UPDATED:", updated
249224

250-
stmt = luigi.update().\
251-
where(luigi.c.luigi_job == job_name).\
225+
stmt = monitordb_table.update().\
226+
where(monitordb_table.c.consonance_job_uuid == job_uuid).\
252227
values(status=status_json['state'],
253228
start_time=created,
254229
last_updated=updated)
255-
exec_result = conn.execute(stmt)
230+
exec_result = monitordb_connection.execute(stmt)
256231

257232
except Exception as e:
258233
print >>sys.stderr, e.message, e.args
259234
print >>sys.stderr, "Dumping job entry:", job
260235

261-
stmt = luigi.update().\
262-
where((and_(luigi.c.luigi_job == job_name,
263-
luigi.c.status != 'SUCCESS',
264-
luigi.c.status != 'FAILED'))).\
236+
stmt = monitordb_table.update().\
237+
where((and_(monitordb_table.c.luigi_job == job_name,
238+
monitordb_table.c.status != 'SUCCESS',
239+
monitordb_table.c.status != 'FAILED'))).\
265240
values(status='JOB NOT FOUND')
266-
exec_result = conn.execute(stmt)
241+
exec_result = monitordb_connection.execute(stmt)

‎monitordb_lib.py‎

Lines changed: 36 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,36 @@
1+
from sqlalchemy import create_engine, MetaData, Table, String, Float, Column
2+
import os
3+
4+
def luigiDBInit():
5+
# Initializes luigi database with the appropriate env variables
6+
#
7+
# Returns the connection and luigi table for SQL operations,
8+
# but abstracts the table's columns, db engine, and metadata.
9+
db_engine = create_engine('postgresql://{}:{}@db/{}'.format(os.getenv("POSTGRES_USER"), os.getenv("POSTGRES_PASSWORD"), os.getenv("POSTGRES_DB")), echo=False)
10+
monitordb_connection = db_engine.connect()
11+
monitordb_table = Table('luigi', MetaData(db_engine),
12+
Column("luigi_job", String(100)),
13+
Column("status", String(20)),
14+
Column("submitter_specimen_id", String(100)),
15+
Column("specimen_uuid", String(100)),
16+
Column("workflow_name", String(100)),
17+
Column("center_name", String(100)),
18+
Column("submitter_donor_id", String(100)),
19+
Column("consonance_job_uuid", String(100), primary_key=True),
20+
Column("submitter_donor_primary_site", String(100)),
21+
Column("project", String(100)),
22+
Column("analysis_type", String(100)),
23+
Column("program", String(100)),
24+
Column("donor_uuid", String(100)),
25+
Column("submitter_sample_id", String(100)),
26+
Column("submitter_experimental_design", String(100)),
27+
Column("submitter_specimen_type", String(100)),
28+
Column("workflow_version", String(100)),
29+
Column("sample_uuid", String(100)),
30+
Column("start_time", String(100)),
31+
Column("last_updated", String(100)))
32+
33+
if not db_engine.dialect.has_table(db_engine, monitordb_table):
34+
monitordb_table.create()
35+
36+
return monitordb_connection, monitordb_table

0 commit comments

Comments
 (0)