Skip to content

Commit eb219e7

Browse files
aegbert5aegbert5
committed
Create database for single run triggers
1 parent 061182e commit eb219e7

7 files changed

Lines changed: 306 additions & 46 deletions

File tree

‎admin/app/com/lucidchart/piezo/admin/controllers/Triggers.scala‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -354,6 +354,34 @@ class Triggers(
354354
}
355355
}
356356

357+
def triggerJobOneTime(group: String, name: String, id: Long): Action[AnyContent] = Action { request =>
358+
val jobKey = new JobKey(name, group)
359+
360+
if (scheduler.checkExists(jobKey)) {
361+
try {
362+
// Only run a trigger, if we haven't seen this id before
363+
if (jobHistoryModel.addOneTimeJobIfNotExists(jobKey, id)) {
364+
// Single run trigger has its id passed to the scheduler via the job-data-map. The WorkerJobListener will
365+
// use that id to update the existing record in job_history table
366+
val jobDataMap = jobHistoryModel.createJobDataMapForOneTimeJob(id)
367+
scheduler.triggerJob(jobKey, jobDataMap)
368+
}
369+
Ok
370+
} catch {
371+
case e: SchedulerException => {
372+
logger.error(
373+
"Exception caught triggering job one-time %s %s - %s. -- %s"
374+
.format(group, name, id, e.getLocalizedMessage),
375+
e,
376+
)
377+
InternalServerError
378+
}
379+
}
380+
} else {
381+
NotFound
382+
}
383+
}
384+
357385
def patchTrigger(group: String, name: String): Action[AnyContent] = Action { implicit request =>
358386
val triggerKey = new TriggerKey(name, group)
359387
val triggerExists = scheduler.checkExists(triggerKey)

‎admin/conf/routes‎

Lines changed: 32 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -3,39 +3,40 @@
33
# ~~~~
44

55
# Home page
6-
GET / com.lucidchart.piezo.admin.controllers.ApplicationController.index
7-
GET /jobs com.lucidchart.piezo.admin.controllers.Jobs.getIndex
8-
GET /jobs/new com.lucidchart.piezo.admin.controllers.Jobs.getNewJobForm(templateGroup: Option[String] ?= None, templateName: Option[String] ?= None)
9-
GET /jobs/:group/:name com.lucidchart.piezo.admin.controllers.Jobs.getJob(group: String, name: String)
10-
POST /jobs/:group/:name com.lucidchart.piezo.admin.controllers.Jobs.putJob(group: String, name: String)
11-
DELETE /jobs/:group/:name com.lucidchart.piezo.admin.controllers.Jobs.deleteJob(group: String, name: String)
12-
GET /jobs/:group/:name/editor com.lucidchart.piezo.admin.controllers.Jobs.getEditJobAction(group: String, name: String)
13-
POST /jobs com.lucidchart.piezo.admin.controllers.Jobs.postJob
14-
15-
POST /data/jobs com.lucidchart.piezo.admin.controllers.Jobs.postJobs
16-
GET /data/jobs com.lucidchart.piezo.admin.controllers.Jobs.getJobsDetail
17-
GET /data/jobs/:group/:name com.lucidchart.piezo.admin.controllers.Jobs.getJobDetail(group: String, name: String)
18-
19-
GET /typeahead/jobs/:sofar com.lucidchart.piezo.admin.controllers.Jobs.jobGroupTypeAhead(sofar: String)
20-
GET /typeahead/jobs/:group/:sofar com.lucidchart.piezo.admin.controllers.Jobs.jobNameTypeAhead(group: String, sofar: String)
21-
22-
GET /triggers com.lucidchart.piezo.admin.controllers.Triggers.getIndex
23-
GET /triggers/new/:triggerType com.lucidchart.piezo.admin.controllers.Triggers.getNewTriggerForm(triggerType, jobGroup: String ?= "", jobName: String ?= "", templateGroup: Option[String] ?= None, templateName: Option[String] ?= None)
24-
GET /triggers/:group/:name com.lucidchart.piezo.admin.controllers.Triggers.getTrigger(group: String, name: String)
25-
POST /triggers/:group/:name com.lucidchart.piezo.admin.controllers.Triggers.putTrigger(group: String, name: String)
26-
DELETE /triggers/:group/:name com.lucidchart.piezo.admin.controllers.Triggers.deleteTrigger(group: String, name: String)
27-
GET /triggers/:group/:name/editor com.lucidchart.piezo.admin.controllers.Triggers.getEditTriggerAction(group: String, name: String)
28-
POST /triggers/:group/:name/runner com.lucidchart.piezo.admin.controllers.Triggers.triggerJob(group: String, name: String)
29-
POST /triggers com.lucidchart.piezo.admin.controllers.Triggers.postTrigger()
30-
PATCH /triggers/:group/:name com.lucidchart.piezo.admin.controllers.Triggers.patchTrigger(group: String, name: String)
31-
32-
GET /typeahead/triggers/:sofar com.lucidchart.piezo.admin.controllers.Triggers.triggerGroupTypeAhead(sofar: String)
33-
34-
GET /favicon.ico controllers.Assets.at(path="/public/img", file="favicon.ico")
6+
GET / com.lucidchart.piezo.admin.controllers.ApplicationController.index
7+
GET /jobs com.lucidchart.piezo.admin.controllers.Jobs.getIndex
8+
GET /jobs/new com.lucidchart.piezo.admin.controllers.Jobs.getNewJobForm(templateGroup: Option[String] ?= None, templateName: Option[String] ?= None)
9+
GET /jobs/:group/:name com.lucidchart.piezo.admin.controllers.Jobs.getJob(group: String, name: String)
10+
POST /jobs/:group/:name com.lucidchart.piezo.admin.controllers.Jobs.putJob(group: String, name: String)
11+
DELETE /jobs/:group/:name com.lucidchart.piezo.admin.controllers.Jobs.deleteJob(group: String, name: String)
12+
GET /jobs/:group/:name/editor com.lucidchart.piezo.admin.controllers.Jobs.getEditJobAction(group: String, name: String)
13+
POST /jobs com.lucidchart.piezo.admin.controllers.Jobs.postJob
14+
15+
POST /data/jobs com.lucidchart.piezo.admin.controllers.Jobs.postJobs
16+
GET /data/jobs com.lucidchart.piezo.admin.controllers.Jobs.getJobsDetail
17+
GET /data/jobs/:group/:name com.lucidchart.piezo.admin.controllers.Jobs.getJobDetail(group: String, name: String)
18+
19+
GET /typeahead/jobs/:sofar com.lucidchart.piezo.admin.controllers.Jobs.jobGroupTypeAhead(sofar: String)
20+
GET /typeahead/jobs/:group/:sofar com.lucidchart.piezo.admin.controllers.Jobs.jobNameTypeAhead(group: String, sofar: String)
21+
22+
GET /triggers com.lucidchart.piezo.admin.controllers.Triggers.getIndex
23+
GET /triggers/new/:triggerType com.lucidchart.piezo.admin.controllers.Triggers.getNewTriggerForm(triggerType, jobGroup: String ?= "", jobName: String ?= "", templateGroup: Option[String] ?= None, templateName: Option[String] ?= None)
24+
GET /triggers/:group/:name com.lucidchart.piezo.admin.controllers.Triggers.getTrigger(group: String, name: String)
25+
POST /triggers/:group/:name com.lucidchart.piezo.admin.controllers.Triggers.putTrigger(group: String, name: String)
26+
DELETE /triggers/:group/:name com.lucidchart.piezo.admin.controllers.Triggers.deleteTrigger(group: String, name: String)
27+
GET /triggers/:group/:name/editor com.lucidchart.piezo.admin.controllers.Triggers.getEditTriggerAction(group: String, name: String)
28+
POST /triggers/:group/:name/runner com.lucidchart.piezo.admin.controllers.Triggers.triggerJob(group: String, name: String)
29+
POST /triggers/:group/:name/onetime/:id com.lucidchart.piezo.admin.controllers.Triggers.triggerJobOneTime(group: String, name: String, id: Long)
30+
POST /triggers com.lucidchart.piezo.admin.controllers.Triggers.postTrigger()
31+
PATCH /triggers/:group/:name com.lucidchart.piezo.admin.controllers.Triggers.patchTrigger(group: String, name: String)
32+
33+
GET /typeahead/triggers/:sofar com.lucidchart.piezo.admin.controllers.Triggers.triggerGroupTypeAhead(sofar: String)
34+
35+
GET /favicon.ico controllers.Assets.at(path="/public/img", file="favicon.ico")
3536

3637
# Worker Health Check
37-
GET /health com.lucidchart.piezo.admin.controllers.HealthCheck.main()
38+
GET /health com.lucidchart.piezo.admin.controllers.HealthCheck.main()
3839

3940
# Map static resources from the /public folder to the /assets URL path
4041

41-
GET /assets/*file controllers.Assets.at(path="/public", file)
42+
GET /assets/*file controllers.Assets.at(path="/public", file)

‎worker/src/main/resources/piezo_mysql_9.sql‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,3 @@
1+
ALTER TABLE job_history
2+
DROP INDEX start_key,
3+
ADD INDEX start_key (start, trigger_group);

‎worker/src/main/scala/com/lucidchart/piezo/JobHistoryModel.scala‎

Lines changed: 111 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2,10 +2,11 @@ package com.lucidchart.piezo
22

33
import java.sql.{ResultSet, Timestamp}
44
import java.util.Date
5-
import org.quartz.{JobKey, TriggerKey}
5+
import org.quartz.{JobDataMap, JobKey, TriggerKey}
66
import org.slf4j.LoggerFactory
77
import org.slf4j.Logger
88
import java.sql.Connection
9+
import java.time.Instant
910

1011
case class JobRecord(
1112
name: String,
@@ -21,6 +22,18 @@ case class JobRecord(
2122
class JobHistoryModel(getConnection: () => Connection) {
2223
val logger: Logger = LoggerFactory.getLogger(this.getClass)
2324

25+
// Trigger Group for records that aren't deletable
26+
private final val oneTimeJobTriggerGroup = "ONE_TIME_JOB"
27+
final def oneTimeTriggerKey(fireInstanceId: Long): TriggerKey =
28+
TriggerKey(fireInstanceId.toString, oneTimeJobTriggerGroup)
29+
30+
// Methods to store the one-time-job id in a job-data-map
31+
final val jobDataMapOneTimeJobKey = "OneTimeJobId"
32+
final def getOneTimeJobIdFromDataMap(jobDataMap: JobDataMap) = jobDataMap.getString(jobDataMapOneTimeJobKey)
33+
final def createJobDataMapForOneTimeJob(id: Long): JobDataMap = new JobDataMap(
34+
java.util.Map.of(jobDataMapOneTimeJobKey, id),
35+
)
36+
2437
def addJob(
2538
fireInstanceId: String,
2639
jobKey: JobKey,
@@ -66,10 +79,11 @@ class JobHistoryModel(getConnection: () => Connection) {
6679
val connection = getConnection()
6780
try {
6881
val prepared = connection.prepareStatement(
69-
"""
82+
s"""
7083
DELETE
7184
FROM job_history
7285
WHERE start < ?
86+
AND trigger_group != $oneTimeJobTriggerGroup
7387
""".stripMargin,
7488
)
7589
prepared.setTimestamp(1, new Timestamp(minStart))
@@ -190,4 +204,99 @@ class JobHistoryModel(getConnection: () => Connection) {
190204
rs.getString("fire_instance_id"),
191205
)
192206
}
207+
208+
/**
209+
* Check if we have already triggered a one-time-job with the given trigger key and fireInstanceId.
210+
*
211+
* This is useful for seeing if a one-time job has already been triggered, to ensure that triggering a one-time job
212+
* with the same instance id is an idempotent operation. If the one-time job has not been triggered, the same
213+
* transaction is used to add the one-time-job to the database, to avoid race conditions
214+
*/
215+
def addOneTimeJobIfNotExists(jobKey: JobKey, fireInstanceId: Long): Boolean = {
216+
val connection = getConnection()
217+
connection.setAutoCommit(false)
218+
219+
// Use a trigger key that the database won't clean up in "JobHistoryCleanup"
220+
val triggerKey: TriggerKey = oneTimeTriggerKey(fireInstanceId)
221+
222+
try {
223+
val prepared = connection.prepareStatement(
224+
"""
225+
SELECT EXISTS(
226+
SELECT 1
227+
FROM job_history
228+
WHERE fire_instance_id=?
229+
LIMIT 1
230+
) as record_exists
231+
""".stripMargin,
232+
)
233+
prepared.setString(1, fireInstanceId.toString)
234+
val rs = prepared.executeQuery()
235+
if (rs.next() && rs.getBoolean(1)) {
236+
// Record already exists, don't add it again
237+
false
238+
} else {
239+
// Mark this trigger (as already run) in the same transaction, to prevent race conditions where two requests
240+
// trigger the same job
241+
val triggerStartTime: Instant = java.time.Instant.now()
242+
// Same as "addJob()" but without the finish
243+
val prepared = connection.prepareStatement(
244+
"""
245+
INSERT INTO job_history(
246+
fire_instance_id,
247+
job_name,
248+
job_group,
249+
trigger_name,
250+
trigger_group,
251+
success,
252+
start
253+
)
254+
VALUES(?, ?, ?, ?, ?, ?, ?)
255+
""".stripMargin,
256+
)
257+
prepared.setString(1, fireInstanceId.toString)
258+
prepared.setString(2, jobKey.getName)
259+
prepared.setString(3, jobKey.getGroup)
260+
prepared.setString(4, triggerKey.getName)
261+
prepared.setString(5, triggerKey.getGroup)
262+
prepared.setBoolean(6, true)
263+
prepared.setObject(7, triggerStartTime)
264+
prepared.executeUpdate()
265+
connection.commit()
266+
true
267+
}
268+
} finally {
269+
connection.close()
270+
}
271+
}
272+
273+
def completeOneTimeJob(
274+
fireInstanceId: String,
275+
fireTime: Instant,
276+
instanceDurationInMillis: Long,
277+
success: Boolean,
278+
): Unit = {
279+
val connection = getConnection()
280+
try {
281+
val prepared = connection.prepareStatement(
282+
"""
283+
UPDATE job_history
284+
SET
285+
success=?,
286+
start=?,
287+
finish=?
288+
WHERE fire_instance_id=?
289+
""".stripMargin,
290+
)
291+
prepared.setBoolean(1, success)
292+
prepared.setObject(2, fireTime)
293+
prepared.setObject(3, fireTime.plusMillis(instanceDurationInMillis))
294+
prepared.setString(4, fireInstanceId)
295+
prepared.executeUpdate()
296+
} catch {
297+
case e: Exception => logger.error("error in recording completion of one-time-job", e)
298+
} finally {
299+
connection.close()
300+
}
301+
}
193302
}

‎worker/src/main/scala/com/lucidchart/piezo/TriggerHistoryModel.scala‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,11 +1,10 @@
11
package com.lucidchart.piezo
22

3-
import java.sql.Timestamp
3+
import java.sql.{Connection, Timestamp}
44
import org.quartz.TriggerKey
55
import org.slf4j.LoggerFactory
66
import java.util.Date
77
import org.slf4j.Logger
8-
import java.sql.Connection
98

109
case class TriggerRecord(
1110
name: String,
@@ -68,7 +67,7 @@ class TriggerHistoryModel(getConnection: () => Connection) {
6867
def deleteTriggers(minScheduledStart: Long): Int = {
6968
val connection = getConnection()
7069
try {
71-
val prepared = connection.prepareStatement("""DELETE FROM trigger_history WHERE scheduled_start < ?""")
70+
val prepared = connection.prepareStatement(s"""DELETE FROM trigger_history WHERE scheduled_start < ?""")
7271
prepared.setTimestamp(1, new Timestamp(minScheduledStart))
7372
prepared.executeUpdate()
7473
} catch {

‎worker/src/main/scala/com/lucidchart/piezo/WorkerJobListener.scala‎

Lines changed: 19 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -23,14 +23,25 @@ class WorkerJobListener(getConnection: () => Connection, statsd: StatsDClient, u
2323
def jobWasExecuted(context: JobExecutionContext, jobException: JobExecutionException): Unit = {
2424
try {
2525
val success = jobException == null
26-
jobHistoryModel.addJob(
27-
context.getFireInstanceId,
28-
context.getTrigger.getJobKey,
29-
context.getTrigger.getKey,
30-
context.getFireTime,
31-
context.getJobRunTime,
32-
success = success,
33-
)
26+
val oneTimeJobId: String = jobHistoryModel.getOneTimeJobIdFromDataMap(context.getMergedJobDataMap)
27+
if (oneTimeJobId == null) {
28+
jobHistoryModel.addJob(
29+
context.getFireInstanceId,
30+
context.getTrigger.getJobKey,
31+
context.getTrigger.getKey,
32+
context.getFireTime,
33+
context.getJobRunTime,
34+
success = success,
35+
)
36+
} else {
37+
// Update the existing record from the job_history table
38+
jobHistoryModel.completeOneTimeJob(
39+
oneTimeJobId,
40+
context.getFireTime.toInstant,
41+
context.getJobRunTime,
42+
success = success,
43+
)
44+
}
3445

3546
val suffix = if (success) "succeeded" else "failed"
3647
val jobKey = s"${context.getTrigger.getJobKey.getGroup}.${context.getTrigger.getJobKey.getName}"

0 commit comments

Comments
 (0)