jenkins-bot has submitted this change and it was merged.
Change subject: Merged --dispatcher and --wrapper params
......................................................................
Merged --dispatcher and --wrapper params
* Always assume that a JSON response is given (whether using curl or not). This
means
that runJobs.php dispatchers need --result=json.
* Use the JSON response to increment job run/fail counters.
Change-Id: I0b9d13ffe116d906dd51c928054d2939433308e7
---
M jobrunner.sample.json
M redisJobRunnerService
2 files changed, 31 insertions(+), 49 deletions(-)
Approvals:
Ori.livneh: Looks good to me, approved
jenkins-bot: Verified
diff --git a/jobrunner.sample.json b/jobrunner.sample.json
index 6762058..bca4a9e 100644
--- a/jobrunner.sample.json
+++ b/jobrunner.sample.json
@@ -95,6 +95,5 @@
"statsd": "statsd.eqiad.wmnet:8125",
- // HET deploy wrapper
- "wrapper": "php /usr/local/apache/common/multiversion/MWScript.php"
-}
\ No newline at end of file
+ "dispatcher": "php /usr/local/apache/common/multiversion/MWScript.php
runJobs.php --wiki=%(db)x --type=%(type)x --maxtime=%(maxtime)x
--memory-limit=%(maxmem)x --result=json"
+}
diff --git a/redisJobRunnerService b/redisJobRunnerService
index 75dfac8..262fdf1 100755
--- a/redisJobRunnerService
+++ b/redisJobRunnerService
@@ -40,15 +40,13 @@
/** @var array Map of (job type => integer) */
protected $attemptsMap = array();
- /** @var string MediaWiki script wrapper */
- public $wrapper;
/** @var array Map of (id => (job type list,count) */
public $loopMap = array();
/** @var array Map of (job type => integer) */
public $maxRealMap = array();
/** @var array Map of (job type => integer) */
public $maxMemMap = array();
- /** @var array String URL of dispatcher */
+ /** @var array String command to run jobs and return the status JSON
blob */
public $dispatcher;
/**
@@ -136,6 +134,10 @@
if ( !count( $this->queueSrvs ) ) {
throw new Exception( "Empty list for 'redis.queues'." );
}
+ $this->dispatcher = $config['dispatcher'];
+ if ( !$this->dispatcher ) {
+ throw new Exception( "No command provided for
'dispatcher'." );
+ }
foreach ( $config['groups'] as $group ) {
$runners = $group['runners'];
@@ -193,10 +195,6 @@
$this->maxMemMap['*'] = '300M';
if ( isset( $config['limits']['memory'] ) ) {
$this->maxMemMap = $config['limits']['memory'] +
$this->maxMemMap;
- }
-
- if ( isset( $config['dispatcher'] ) ) {
- $this->dispatcher = $config['dispatcher'];
}
if ( isset( $config['statsd'] ) ) {
@@ -763,7 +761,6 @@
'cmd' => null,
'stime' => 0,
'sigtime' => 0,
- 'isCurl' => false
);
}
@@ -776,6 +773,7 @@
public function refillSlots( $loop, array $prioMap, array &$pending ) {
$free = 0;
$new = 0;
+ $host = gethostname();
$cTime = time();
foreach ( $this->procMap[$loop] as $slot => &$procSlot ) {
$status = $procSlot['handle']
@@ -801,26 +799,25 @@
continue; // slot is busy
}
} elseif ( $status && !$status['running'] ) {
- $error = null;
- if ( $procSlot['isCurl'] && $status['exitcode']
== 0 ) {
- $error = stream_get_contents(
$procSlot['pipes'][1] );
- if ( !json_decode( $error ) ) {
- $status['exitcode'] = 1; //
probably PHP exception
- }
- }
- if ( $status['exitcode'] == 0 ) {
+ $response = stream_get_contents(
$procSlot['pipes'][1] );
+ $result = json_decode( $response ); // an array
if no exceptions happened
+ if ( $status['exitcode'] == 0 && is_array(
$result ) ) {
// If this finished early, lay off of
the queue for a while
if ( ( $cTime - $procSlot['stime'] ) <
$this->srvc->hpMaxTime/2 ) {
unset(
$pending[$procSlot['type']][$procSlot['db']] );
$this->srvc->debug( "Queue
'{$procSlot['db']}/{$procSlot['type']}' emptied." );
}
+ $ok = 0; // jobs that ran OK
+ foreach ( $result['jobs'] as $status ) {
+ $ok += ( $status['status'] ===
'ok' ) ? 1 : 0;
+ }
+ $failed = count( $result['jobs'] ) -
$ok;
+ $this->srvc->incrStats(
"pop.{$procSlot['type']}.ok.{$host}", $ok );
+ $this->srvc->incrStats(
"pop.{$procSlot['type']}.failed.{$host}", $failed );
} else {
// Mention any serious errors that may
have occured
$cmd = $procSlot['cmd'];
- if ( $error === null ) {
- $error = stream_get_contents(
$procSlot['pipes'][2] )
- ?: stream_get_contents(
$procSlot['pipes'][1] );
- }
+ $error = stream_get_contents(
$procSlot['pipes'][2] ) ?: $response;
$this->srvc->error( "Runner loop $loop
process in slot $slot " .
"gave status
'{$status['exitcode']}':\n$cmd\n\t$error" );
$this->srvc->incrStats(
'runner-status.error', 1 );
@@ -906,34 +903,22 @@
protected function spawnRunner( $loop, $slot, $highPrio, array $queue,
array &$procSlot ) {
// Pick a random queue
list( $type, $db ) = $queue;
- $maxTime = $highPrio ? $this->srvc->lpMaxTime :
$this->srvc->hpMaxTime;
- $maxMem = isset( $this->srvc->maxMemMap[$type] )
+ $maxtime = $highPrio ? $this->srvc->lpMaxTime :
$this->srvc->hpMaxTime;
+ $maxmem = isset( $this->srvc->maxMemMap[$type] )
? $this->srvc->maxMemMap[$type]
: $this->srvc->maxMemMap['*'];
- $isCurl = (bool)$this->srvc->dispatcher;
- if ( $isCurl ) {
- $url = $this->srvc->dispatcher .
- "?wiki=" . rawurlencode( $db ) .
- "&type=" . rawurlencode( $type ) .
- "&maxtime=" . rawurlencode( $maxTime );
- $cmd = "curl -s -XPOST -a " . escapeshellarg( $url );
- } else {
- // Apply any HET-deploy style wrapper
- $script = $this->srvc->wrapper
- ? "{$this->srvc->wrapper} runJobs.php"
- : "php maintenance/runJobs.php";
-
- // Make sure the runner is launched with various
time/memory limits.
- // Nice the process so things like ssh and deployment
scripts are fine.
- $cmd = "$script --wiki=" . escapeshellarg( $db ) .
- " --type=" . escapeshellarg( $type ) .
- " --maxtime=" . escapeshellarg( $maxTime ) .
- " --memory-limit=" . escapeshellarg( $maxMem );
- if ( substr( php_uname(), 0, 7 ) !== 'Windows' ) {
- $cmd = "nice -n 19 $cmd";
- }
+ // Make sure the runner is launched with various time/memory
limits.
+ // Nice the process so things like ssh and deployment scripts
are fine.
+ $what = $with = array();
+ foreach ( compact( 'db', 'type', 'maxtime', 'maxmem' ) as $k =>
$v ) {
+ $what[] = "%($k)u";
+ $with[] = rawurlencode( $v );
+ $what[] = "%($k)x";
+ $with[] = escapeshellarg( $v );
}
+ // The dispatcher might be runJobs.php, curl, or wget
+ $cmd = str_replace( $what, $with, $this->srvc->dispatcher );
$descriptors = array(
0 => array( "pipe", "r" ), // stdin (child)
@@ -957,7 +942,6 @@
$procSlot['cmd'] = $cmd;
$procSlot['stime'] = time();
$procSlot['sigtime'] = 0;
- $procSlot['isCurl'] = $isCurl;
if ( $procSlot['handle'] ) {
return true;
@@ -1001,7 +985,6 @@
$procSlot['stime'] = 0;
$procSlot['sigtime'] = 0;
$procSlot['cmd'] = null;
- $procSlot['isCurl'] = false;
}
public function terminateSlots() {
--
To view, visit https://gerrit.wikimedia.org/r/149923
To unsubscribe, visit https://gerrit.wikimedia.org/r/settings
Gerrit-MessageType: merged
Gerrit-Change-Id: I0b9d13ffe116d906dd51c928054d2939433308e7
Gerrit-PatchSet: 6
Gerrit-Project: mediawiki/services/jobrunner
Gerrit-Branch: master
Gerrit-Owner: Aaron Schulz <asch...@wikimedia.org>
Gerrit-Reviewer: Aaron Schulz <asch...@wikimedia.org>
Gerrit-Reviewer: Ori.livneh <o...@wikimedia.org>
Gerrit-Reviewer: jenkins-bot <>
_______________________________________________
MediaWiki-commits mailing list
MediaWiki-commits@lists.wikimedia.org
https://lists.wikimedia.org/mailman/listinfo/mediawiki-commits
