Skip to content

Commit fd7d267

Browse files
vertigo17vertigo17
committed
Secure that only 1 job scheduled will trigger at the same time.
Lock entry on myversion table has been introduced.
1 parent 35074b4 commit fd7d267

File tree

6 files changed

+185
-40
lines changed

6 files changed

+185
-40
lines changed

‎source/src/main/java/org/cerberus/core/crud/dao/IMyVersionDAO.java

+22
Original file line numberDiff line numberDiff line change
@@ -37,10 +37,32 @@ public interface IMyVersionDAO {
3737
*/
3838
MyVersion findMyVersionByKey(String key);
3939

40+
/**
41+
*
42+
* @param myVersion
43+
* @return
44+
*/
4045
boolean update(MyVersion myVersion);
4146

47+
/**
48+
*
49+
* @param myVersion
50+
* @return
51+
*/
4252
boolean updateMyVersionString(MyVersion myVersion);
4353

54+
/**
55+
*
56+
* @param myVersion
57+
* @return
58+
*/
59+
boolean updateAndLockSchedulerVersion(long myVersion);
60+
61+
/**
62+
*
63+
* @param key
64+
* @return
65+
*/
4466
boolean flagMyVersionString(String key);
4567

4668
}

‎source/src/main/java/org/cerberus/core/crud/dao/impl/MyVersionDAO.java

+44
Original file line numberDiff line numberDiff line change
@@ -169,6 +169,50 @@ public boolean updateMyVersionString(MyVersion myVersion) {
169169
return result;
170170
}
171171

172+
@Override
173+
public boolean updateAndLockSchedulerVersion(long myVersion) {
174+
boolean result = false;
175+
final String query = "UPDATE myversion SET value = ? WHERE `key` = 'scheduler_active_instance_version' and value < ?";
176+
177+
// Debug message on SQL.
178+
if (LOG.isDebugEnabled()) {
179+
LOG.debug("SQL : " + query);
180+
LOG.debug("SQL.param.value : " + myVersion);
181+
LOG.debug("SQL.param.value : " + (myVersion - 10000));
182+
}
183+
184+
Connection connection = this.databaseSpring.connect();
185+
try {
186+
PreparedStatement preStat = connection.prepareStatement(query);
187+
try {
188+
preStat.setLong(1, myVersion);
189+
preStat.setLong(2, myVersion - 10000);
190+
191+
if (preStat.executeUpdate() >= 1) {
192+
result = true;
193+
} else {
194+
result = false;
195+
}
196+
197+
} catch (SQLException exception) {
198+
LOG.warn("Unable to execute query : " + exception.toString());
199+
} finally {
200+
preStat.close();
201+
}
202+
} catch (SQLException exception) {
203+
LOG.warn("Unable to execute query : " + exception.toString());
204+
} finally {
205+
try {
206+
if (connection != null) {
207+
connection.close();
208+
}
209+
} catch (SQLException e) {
210+
LOG.warn(e.toString());
211+
}
212+
}
213+
return result;
214+
}
215+
172216
@Override
173217
public boolean flagMyVersionString(String key) {
174218
boolean result = false;

‎source/src/main/java/org/cerberus/core/crud/service/IMyVersionService.java

+7
Original file line numberDiff line numberDiff line change
@@ -52,6 +52,13 @@ public interface IMyVersionService {
5252
*/
5353
boolean updateMyVersionString(String key, String value);
5454

55+
/**
56+
*
57+
* @param value
58+
* @return true if the update was done. False in case there were an issue.
59+
*/
60+
boolean updateAndLockSchedulerVersion(long value);
61+
5562
/**
5663
* Flag the key. Means that the method will return true if the previous
5764
* value was N and update manage to move it to Y. It returns false if the

‎source/src/main/java/org/cerberus/core/crud/service/impl/MyVersionService.java

+5
Original file line numberDiff line numberDiff line change
@@ -65,6 +65,11 @@ public boolean updateMyVersionString(String key, String value) {
6565
return this.myVersionDAO.updateMyVersionString(DtbVersion);
6666
}
6767

68+
@Override
69+
public boolean updateAndLockSchedulerVersion(long value) {
70+
return this.myVersionDAO.updateAndLockSchedulerVersion(value);
71+
}
72+
6873
@Override
6974
public boolean flagMyVersionString(String key) {
7075
return this.myVersionDAO.flagMyVersionString(key);

‎source/src/main/java/org/cerberus/core/engine/scheduledtasks/ScheduledTaskRunner.java

+101-39
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,13 @@
1919
*/
2020
package org.cerberus.core.engine.scheduledtasks;
2121

22+
import java.text.DateFormat;
23+
import java.text.SimpleDateFormat;
24+
import java.time.LocalDate;
25+
import java.time.LocalTime;
26+
import java.util.Date;
27+
import java.util.TimeZone;
28+
import org.cerberus.core.crud.service.IMyVersionService;
2229
import org.cerberus.core.crud.service.IParameterService;
2330
import org.cerberus.core.crud.service.ITestCaseExecutionQueueDepService;
2431
import org.cerberus.core.crud.service.ITestCaseExecutionQueueService;
@@ -47,98 +54,153 @@ public class ScheduledTaskRunner {
4754
private SchedulerInit schedulerInit;
4855
@Autowired
4956
private ITestCaseExecutionQueueDepService testCaseExecutionQueueDepService;
57+
@Autowired
58+
private IMyVersionService myVersionService;
5059

5160
private int b1TickNumberTarget = 60;
5261
private int b1TickNumber = 1;
5362
private int b2TickNumberTarget = 30;
5463
private int b2TickNumber = 1;
5564
private int b3TickNumberTarget = 1;
5665
private int b3TickNumber = 1;
66+
private int b4TickNumberTarget = 1;
67+
private int b4TickNumber = 1;
68+
69+
private long loadingTimestamp = 0;
70+
private boolean instanceActive = true;
71+
72+
private static final String DATE_FORMAT = "yyyy-MM-dd'T'HH:mm:ss.S";
5773

5874
private static final org.apache.logging.log4j.Logger LOG = org.apache.logging.log4j.LogManager.getLogger(ScheduledTaskRunner.class);
5975

6076
@Scheduled(fixedRate = 60000, initialDelay = 30000 /* Every minute */)
6177
public void nextStep() {
62-
LOG.debug("Schedule Start. " + b1TickNumber + "/" + b1TickNumberTarget + " - " + b2TickNumber + "/" + b2TickNumberTarget + " - " + b3TickNumber + "/" + b3TickNumberTarget);
6378

64-
// We get the new period from parameter and trigger the Queue automatic cancellation job.
65-
b1TickNumberTarget = parameterService.getParameterIntegerByKey("cerberus_automaticqueuecancellationjob_period", "", 60);
66-
b2TickNumberTarget = parameterService.getParameterIntegerByKey("cerberus_automaticqueueprocessingjob_period", "", 30);
67-
68-
if (b1TickNumber < b1TickNumberTarget) {
69-
b1TickNumber++;
70-
} else {
71-
b1TickNumber = 1;
72-
performBatch1_CancelOldQueueEntries();
79+
/**
80+
* Secure only 1 Task trigger on a given cerberus instance. Multiple
81+
* jobs could be triggered if several instance is running. The first
82+
* scheduler trigger will define loadingTimestamp and instanceActive
83+
* based on a lock value defined at database level. Only the 1st update
84+
* (within 10 second delay) will be considered as active. All others
85+
* will be disabled.
86+
*/
87+
if (loadingTimestamp == 0) {
88+
Date d = new Date();
89+
TimeZone tz = TimeZone.getTimeZone("UTC");
90+
DateFormat df = new SimpleDateFormat(DATE_FORMAT);
91+
df.setTimeZone(tz);
92+
long newVersion = new java.util.Date().getTime();
93+
loadingTimestamp = newVersion;
94+
LOG.debug("Setting local scheduler Version to : {}", newVersion);
95+
instanceActive = myVersionService.updateAndLockSchedulerVersion(newVersion);
7396
}
7497

75-
if (b2TickNumber < b2TickNumberTarget) {
76-
b2TickNumber++;
77-
} else {
78-
b2TickNumber = 1;
79-
// We trigger the Queue Processing job.
80-
performBatch2_ProcessQueue();
81-
}
98+
if (instanceActive) {
99+
100+
LOG.debug("Schedule ({}) Start. "
101+
+ b1TickNumber + "/" + b1TickNumberTarget + " - "
102+
+ b2TickNumber + "/" + b2TickNumberTarget + " - "
103+
+ b3TickNumber + "/" + b3TickNumberTarget + " - "
104+
+ b4TickNumber + "/" + b4TickNumberTarget,
105+
loadingTimestamp);
106+
107+
// We get the new period of each job from parameter.
108+
b1TickNumberTarget = parameterService.getParameterIntegerByKey("cerberus_automaticqueuecancellationjob_period", "", 60);
109+
b2TickNumberTarget = parameterService.getParameterIntegerByKey("cerberus_automaticqueueprocessingjob_period", "", 30);
110+
b3TickNumberTarget = 1;
111+
b4TickNumberTarget = 1;
112+
113+
if (b1TickNumber < b1TickNumberTarget) {
114+
b1TickNumber++;
115+
} else {
116+
b1TickNumber = 1;
117+
performBatch1_CancelOldQueueEntries();
118+
}
119+
120+
if (b2TickNumber < b2TickNumberTarget) {
121+
b2TickNumber++;
122+
} else {
123+
b2TickNumber = 1;
124+
// We trigger the Queue Processing job.
125+
performBatch2_ProcessQueue();
126+
}
127+
128+
if (b3TickNumber < b3TickNumberTarget) {
129+
b3TickNumber++;
130+
} else {
131+
b3TickNumber = 1;
132+
// We trigger the Scheduler init job.
133+
performBatch3_SchedulerInit();
134+
}
135+
136+
if (b4TickNumber < b4TickNumberTarget) {
137+
b4TickNumber++;
138+
} else {
139+
b4TickNumber = 1;
140+
// We trigger the Queue dependencies release by timing.
141+
performBatch4_ProcessTimingBasedQueueDependencies();
142+
}
143+
144+
LOG.debug("Schedule ({}) Stop. "
145+
+ b1TickNumber + "/" + b1TickNumberTarget + " - "
146+
+ b2TickNumber + "/" + b2TickNumberTarget + " - "
147+
+ b3TickNumber + "/" + b3TickNumberTarget + " - "
148+
+ b4TickNumber + "/" + b4TickNumberTarget,
149+
loadingTimestamp);
82150

83-
if (b3TickNumber < b3TickNumberTarget) {
84-
b3TickNumber++;
85151
} else {
86-
b3TickNumber = 1;
87-
// We trigger the Scheduler init job.
88-
performBatch3_SchedulerInit();
89-
}
152+
LOG.debug("Schedule ({}) disabled. ", loadingTimestamp);
90153

91-
performBatch3_ProcessTimingBasedQueueDependencies();
154+
}
92155

93-
LOG.debug("Schedule Stop. " + b1TickNumber + "/" + b1TickNumberTarget + " - " + b2TickNumber + "/" + b2TickNumberTarget + " - " + b3TickNumber + "/" + b3TickNumberTarget);
94156
}
95157

96158
private void performBatch1_CancelOldQueueEntries() {
97-
LOG.info("automaticqueuecancellationjob Task triggered.");
159+
LOG.info("Schedule ({}) : automaticqueuecancellationjob Task triggered.", loadingTimestamp);
98160
if (parameterService.getParameterBooleanByKey("cerberus_automaticqueuecancellationjob_active", "", true)) {
99161
testCaseExecutionQueueService.cancelRunningOldQueueEntries();
100162
} else {
101-
LOG.info("automaticqueuecancellationjob Task disabled by config (cerberus_automaticqueuecancellationjob_active).");
163+
LOG.info("Schedule ({}) : automaticqueuecancellationjob Task disabled by config (cerberus_automaticqueuecancellationjob_active).", loadingTimestamp);
102164
}
103-
LOG.info("automaticqueuecancellationjob Task ended.");
165+
LOG.info("Schedule ({}) : automaticqueuecancellationjob Task ended.", loadingTimestamp);
104166
}
105167

106168
private void performBatch2_ProcessQueue() {
107-
LOG.info("automaticqueueprocessingjob Task triggered.");
169+
LOG.info("Schedule ({}) : automaticqueueprocessingjob Task triggered.", loadingTimestamp);
108170
if (parameterService.getParameterBooleanByKey("cerberus_automaticqueueprocessingjob_active", "", true)) {
109171
try {
110172
executionThreadPoolService.executeNextInQueue(false);
111173
} catch (CerberusException ex) {
112174
LOG.error(ex.toString(), ex);
113175
}
114176
} else {
115-
LOG.info("automaticqueueprocessingjob Task disabled by config (cerberus_automaticqueueprocessingjob_active).");
177+
LOG.info("Schedule ({}) : automaticqueueprocessingjob Task disabled by config (cerberus_automaticqueueprocessingjob_active).", loadingTimestamp);
116178
}
117-
LOG.info("automaticqueueprocessingjob Task ended.");
179+
LOG.info("Schedule ({}) : automaticqueueprocessingjob Task ended.", loadingTimestamp);
118180
}
119181

120182
private void performBatch3_SchedulerInit() {
121183
try {
122-
LOG.debug("SchedulerInit Task triggered.");
184+
LOG.debug("Schedule ({}) : SchedulerInit Task triggered.", loadingTimestamp);
123185
schedulerInit.init();
124-
LOG.debug("SchedulerInit Task ended.");
186+
LOG.debug("Schedule ({}) : SchedulerInit Task ended.", loadingTimestamp);
125187
} catch (Exception e) {
126-
LOG.error("ScheduleEntry init from scheduletaskrunner failed : " + e);
188+
LOG.error("ScheduleEntry init from scheduletaskrunner failed : " + e, e);
127189
}
128190

129191
}
130192

131-
private void performBatch3_ProcessTimingBasedQueueDependencies() {
193+
private void performBatch4_ProcessTimingBasedQueueDependencies() {
132194
try {
133-
LOG.debug("Queue dep timing Task triggered.");
195+
LOG.debug("Schedule ({}) : Queue dep timing Task triggered.", loadingTimestamp);
134196
int nbReleased = testCaseExecutionQueueDepService.manageDependenciesCheckTimingWaiting();
135197
if (nbReleased > 0) {
136-
LOG.info(nbReleased + " Queue entry(ies) has(have) been released due to TIMING dependencies. We trigger now the processing of the queue entry.");
198+
LOG.info("Schedule ({}) : " + nbReleased + " Queue entry(ies) has(have) been released due to TIMING dependencies. We trigger now the processing of the queue entry.", loadingTimestamp);
137199
executionThreadPoolService.executeNextInQueue(false);
138200
}
139-
LOG.debug("Queue dep timing Task ended.");
201+
LOG.debug("Schedule ({}) : Queue dep timing Task ended.", loadingTimestamp);
140202
} catch (Exception e) {
141-
LOG.error("Queue dep timing Task from scheduletaskrunner failed : " + e);
203+
LOG.error("Queue dep timing Task from scheduletaskrunner failed : " + e, e);
142204
}
143205

144206
}

‎source/src/main/resources/database.sql

+6-1
Original file line numberDiff line numberDiff line change
@@ -6588,4 +6588,9 @@ UPDATE testcase SET `UsrModif`=REPLACE(REPLACE(`UsrModif`, '&#64;', '@'), '%40',
65886588
UPDATE testcasecountryproperties SET `Type` = 'getFromHtml' WHERE `Type` = 'getFromHTML';
65896589

65906590
-- 1874
6591-
UPDATE robot SET lbexemethod='BYRANKING' WHERE lbexemethod='';
6591+
UPDATE robot SET lbexemethod='BYRANKING' WHERE lbexemethod='';
6592+
6593+
-- 1875-1876
6594+
ALTER TABLE myversion MODIFY COLUMN Value BIGINT NULL;
6595+
INSERT INTO myversion (`Key`,Value) VALUES ('scheduler_active_instance_version',0);
6596+

0 commit comments

Comments
 (0)