lib/stompqueuemanager.php

   1 <?php
   2 /**
   3  * StatusNet, the distributed open-source microblogging tool
   4  *
   5  * Abstract class for queue managers
   6  *
   7  * PHP version 5
   8  *
   9  * LICENCE: This program is free software: you can redistribute it and/or modify
  10  * it under the terms of the GNU Affero General Public License as published by
  11  * the Free Software Foundation, either version 3 of the License, or
  12  * (at your option) any later version.
  13  *
  14  * This program is distributed in the hope that it will be useful,
  15  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  16  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  17  * GNU Affero General Public License for more details.
  18  *
  19  * You should have received a copy of the GNU Affero General Public License
  20  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
  21  *
  22  * @category  QueueManager
  23  * @package   StatusNet
  24  * @author    Evan Prodromou <evan@status.net>
  25  * @author    Sarven Capadisli <csarven@status.net>
  26  * @copyright 2009 StatusNet, Inc.
  27  * @license   http://www.fsf.org/licensing/licenses/agpl-3.0.html GNU Affero General Public License version 3.0
  28  * @link      http://status.net/
  29  */
  30
  31 require_once 'Stomp.php';
  32 require_once 'Stomp/Exception.php';
  33
  34 class StompQueueManager extends QueueManager
  35 {
  36     protected $servers;
  37     protected $username;
  38     protected $password;
  39     protected $base;
  40     protected $control;
  41
  42     protected $useTransactions;
  43     protected $useAcks;
  44
  45     protected $sites = array();
  46     protected $subscriptions = array();
  47
  48     protected $cons = array(); // all open connections
  49     protected $disconnect = array();
  50     protected $transaction = array();
  51     protected $transactionCount = array();
  52     protected $defaultIdx = 0;
  53
  54     function __construct()
  55     {
  56         parent::__construct();
  57         $server = common_config('queue', 'stomp_server');
  58         if (is_array($server)) {
  59             $this->servers = $server;
  60         } else {
  61             $this->servers = array($server);
  62         }
  63         $this->username        = common_config('queue', 'stomp_username');
  64         $this->password        = common_config('queue', 'stomp_password');
  65         $this->base            = common_config('queue', 'queue_basename');
  66         $this->control         = common_config('queue', 'control_channel');
  67         $this->breakout        = common_config('queue', 'breakout');
  68         $this->useTransactions = common_config('queue', 'stomp_transactions');
  69         $this->useAcks         = common_config('queue', 'stomp_acks');
  70     }
  71
  72     /**
  73      * Tell the i/o master we only need a single instance to cover
  74      * all sites running in this process.
  75      */
  76     public static function multiSite()
  77     {
  78         return IoManager::INSTANCE_PER_PROCESS;
  79     }
  80
  81     /**
  82      * Optional; ping any running queue handler daemons with a notification
  83      * such as announcing a new site to handle or requesting clean shutdown.
  84      * This avoids having to restart all the daemons manually to update configs
  85      * and such.
  86      *
  87      * Currently only relevant for multi-site queue managers such as Stomp.
  88      *
  89      * @param string $event event key
  90      * @param string $param optional parameter to append to key
  91      * @return boolean success
  92      */
  93     public function sendControlSignal($event, $param='')
  94     {
  95         $message = $event;
  96         if ($param != '') {
  97             $message .= ':' . $param;
  98         }
  99         $this->_connect();
 100         $con = $this->cons[$this->defaultIdx];
 101         $result = $con->send($this->control,
 102                              $message,
 103                              array ('created' => common_sql_now()));
 104         if ($result) {
 105             $this->_log(LOG_INFO, "Sent control ping to queue daemons: $message");
 106             return true;
 107         } else {
 108             $this->_log(LOG_ERR, "Failed sending control ping to queue daemons: $message");
 109             return false;
 110         }
 111     }
 112
 113     /**
 114      * Saves an object into the queue item table.
 115      *
 116      * @param mixed $object
 117      * @param string $queue
 118      *
 119      * @return boolean true on success
 120      * @throws StompException on connection or send error
 121      */
 122     public function enqueue($object, $queue)
 123     {
 124         $this->_connect();
 125         return $this->_doEnqueue($object, $queue, $this->defaultIdx);
 126     }
 127
 128     /**
 129      * Saves a notice object reference into the queue item table
 130      * on the given connection.
 131      *
 132      * @return boolean true on success
 133      * @throws StompException on connection or send error
 134      */
 135     protected function _doEnqueue($object, $queue, $idx)
 136     {
 137         $rep = $this->logrep($object);
 138         $envelope = array('site' => common_config('site', 'nickname'),
 139                           'handler' => $queue,
 140                           'payload' => $this->encode($object));
 141         $msg = serialize($envelope);
 142
 143         $props = array('created' => common_sql_now());
 144         if ($this->isPersistent($queue)) {
 145             $props['persistent'] = 'true';
 146         }
 147
 148         $con = $this->cons[$idx];
 149         $host = $con->getServer();
 150         $target = $this->queueName($queue);
 151         $result = $con->send($target, $msg, $props);
 152
 153         if (!$result) {
 154             $this->_log(LOG_ERR, "Error sending $rep to $queue queue on $host $target");
 155             return false;
 156         }
 157
 158         $this->_log(LOG_DEBUG, "complete remote queueing $rep for $queue on $host $target");
 159         $this->stats('enqueued', $queue);
 160         return true;
 161     }
 162
 163     /**
 164      * Determine whether messages to this queue should be marked as persistent.
 165      * Actual persistent storage depends on the queue server's configuration.
 166      * @param string $queue
 167      * @return bool
 168      */
 169     protected function isPersistent($queue)
 170     {
 171         $mode = common_config('queue', 'stomp_persistent');
 172         if (is_array($mode)) {
 173             return in_array($queue, $mode);
 174         } else {
 175             return (bool)$mode;
 176         }
 177     }
 178
 179     /**
 180      * Send any sockets we're listening on to the IO manager
 181      * to wait for input.
 182      *
 183      * @return array of resources
 184      */
 185     public function getSockets()
 186     {
 187         $sockets = array();
 188         foreach ($this->cons as $con) {
 189             if ($con) {
 190                 $sockets[] = $con->getSocket();
 191             }
 192         }
 193         return $sockets;
 194     }
 195
 196     /**
 197      * Get the Stomp connection object associated with the given socket.
 198      * @param resource $socket
 199      * @return int index into connections list
 200      * @throws Exception
 201      */
 202     protected function connectionFromSocket($socket)
 203     {
 204         foreach ($this->cons as $i => $con) {
 205             if ($con && $con->getSocket() === $socket) {
 206                 return $i;
 207             }
 208         }
 209         throw new Exception(__CLASS__ . " asked to read from unrecognized socket");
 210     }
 211
 212     /**
 213      * We've got input to handle on our socket!
 214      * Read any waiting Stomp frame(s) and process them.
 215      *
 216      * @param resource $socket
 217      * @return boolean ok on success
 218      */
 219     public function handleInput($socket)
 220     {
 221         $idx = $this->connectionFromSocket($socket);
 222         $con = $this->cons[$idx];
 223         $host = $con->getServer();
 224         $this->defaultIdx = $idx;
 225
 226         $ok = true;
 227         try {
 228             $frames = $con->readFrames();
 229         } catch (StompException $e) {
 230             $this->_log(LOG_ERR, "Lost connection to $host: " . $e->getMessage());
 231             fclose($socket); // ???
 232             $this->cons[$idx] = null;
 233             $this->transaction[$idx] = null;
 234             $this->disconnect[$idx] = time();
 235             return false;
 236         }
 237         foreach ($frames as $frame) {
 238             $dest = $frame->headers['destination'];
 239             if ($dest == $this->control) {
 240                 if (!$this->handleControlSignal($frame)) {
 241                     // We got a control event that requests a shutdown;
 242                     // close out and stop handling anything else!
 243                     break;
 244                 }
 245             } else {
 246                 $ok = $this->handleItem($frame) && $ok;
 247             }
 248             $this->ack($idx, $frame);
 249             $this->commit($idx);
 250             $this->begin($idx);
 251         }
 252         return $ok;
 253     }
 254
 255     /**
 256      * Attempt to reconnect in background if we lost a connection.
 257      */
 258     function idle()
 259     {
 260         $now = time();
 261         foreach ($this->cons as $idx => $con) {
 262             if (empty($con)) {
 263                 $age = $now - $this->disconnect[$idx];
 264                 if ($age >= 60) {
 265                     $this->_reconnect($idx);
 266                 }
 267             }
 268         }
 269         return true;
 270     }
 271
 272     /**
 273      * Initialize our connection and subscribe to all the queues
 274      * we're going to need to handle... If multiple queue servers
 275      * are configured for failover, we'll listen to all of them.
 276      *
 277      * Side effects: in multi-site mode, may reset site configuration.
 278      *
 279      * @param IoMaster $master process/event controller
 280      * @return bool return false on failure
 281      */
 282     public function start($master)
 283     {
 284         parent::start($master);
 285         $this->_connectAll();
 286
 287         foreach ($this->cons as $i => $con) {
 288             if ($con) {
 289                 $this->doSubscribe($con);
 290                 $this->begin($i);
 291             }
 292         }
 293         return true;
 294     }
 295
 296     /**
 297      * Close out any active connections.
 298      *
 299      * @return bool return false on failure
 300      */
 301     public function finish()
 302     {
 303         // If there are any outstanding delivered messages we haven't processed,
 304         // free them for another thread to take.
 305         foreach ($this->cons as $i => $con) {
 306             if ($con) {
 307                 $this->rollback($i);
 308                 $con->disconnect();
 309                 $this->cons[$i] = null;
 310             }
 311         }
 312         return true;
 313     }
 314
 315     /**
 316      * Lazy open a single connection to Stomp queue server.
 317      * If multiple servers are configured, we let the Stomp client library
 318      * worry about finding a working connection among them.
 319      */
 320     protected function _connect()
 321     {
 322         if (empty($this->cons)) {
 323             $list = $this->servers;
 324             if (count($list) > 1) {
 325                 shuffle($list); // Randomize to spread load
 326                 $url = 'failover://(' . implode(',', $list) . ')';
 327             } else {
 328                 $url = $list[0];
 329             }
 330             $con = $this->_doConnect($url);
 331             $this->cons = array($con);
 332             $this->transactionCount = array(0);
 333             $this->transaction = array(null);
 334             $this->disconnect = array(null);
 335         }
 336     }
 337
 338     /**
 339      * Lazy open connections to all Stomp servers, if in manual failover
 340      * mode. This means the queue servers don't speak to each other, so
 341      * we have to listen to all of them to make sure we get all events.
 342      */
 343     protected function _connectAll()
 344     {
 345         if (!common_config('queue', 'stomp_manual_failover')) {
 346             return $this->_connect();
 347         }
 348         if (empty($this->cons)) {
 349             $this->cons = array();
 350             $this->transactionCount = array();
 351             $this->transaction = array();
 352             foreach ($this->servers as $idx => $server) {
 353                 try {
 354                     $this->cons[] = $this->_doConnect($server);
 355                     $this->disconnect[] = null;
 356                 } catch (Exception $e) {
 357                     // s'okay, we'll live
 358                     $this->cons[] = null;
 359                     $this->disconnect[] = time();
 360                 }
 361                 $this->transactionCount[] = 0;
 362                 $this->transaction[] = null;
 363             }
 364             if (empty($this->cons)) {
 365                 throw new ServerException("No queue servers reachable...");
 366                 return false;
 367             }
 368         }
 369     }
 370
 371     /**
 372      * Attempt to manually reconnect to the Stomp server for the given
 373      * slot. If successful, set up our subscriptions on it.
 374      */
 375     protected function _reconnect($idx)
 376     {
 377         try {
 378             $con = $this->_doConnect($this->servers[$idx]);
 379         } catch (Exception $e) {
 380             $this->_log(LOG_ERR, $e->getMessage());
 381             $con = null;
 382         }
 383         if ($con) {
 384             $this->cons[$idx] = $con;
 385             $this->disconnect[$idx] = null;
 386
 387             $this->doSubscribe($con);
 388             $this->begin($idx);
 389         } else {
 390             // Try again later...
 391             $this->disconnect[$idx] = time();
 392         }
 393     }
 394
 395     protected function _doConnect($server)
 396     {
 397         $this->_log(LOG_INFO, "Connecting to '$server' as '$this->username'...");
 398         $con = new LiberalStomp($server);
 399
 400         if ($con->connect($this->username, $this->password)) {
 401             $this->_log(LOG_INFO, "Connected.");
 402         } else {
 403             $this->_log(LOG_ERR, 'Failed to connect to queue server');
 404             throw new ServerException('Failed to connect to queue server');
 405         }
 406
 407         return $con;
 408     }
 409
 410     /**
 411      * Set up all our raw queue subscriptions on the given connection
 412      * @param LiberalStomp $con
 413      */
 414     protected function doSubscribe(LiberalStomp $con)
 415     {
 416         $host = $con->getServer();
 417         foreach ($this->subscriptions() as $sub) {
 418             $this->_log(LOG_INFO, "Subscribing to $sub on $host");
 419             $con->subscribe($sub);
 420         }
 421     }
 422
 423     /**
 424      * Grab a full list of stomp-side queue subscriptions.
 425      * Will include:
 426      *  - control broadcast channel
 427      *  - shared group queues for active groups
 428      *  - per-handler and per-site breakouts from $config['queue']['breakout']
 429      *    that are rooted in the active groups.
 430      *
 431      * @return array of strings
 432      */
 433     protected function subscriptions()
 434     {
 435         $subs = array();
 436         $subs[] = $this->control;
 437
 438         foreach ($this->activeGroups as $group) {
 439             $subs[] = $this->base . $group;
 440         }
 441
 442         foreach ($this->breakout as $spec) {
 443             $parts = explode('/', $spec);
 444             if (count($parts) < 2 || count($parts) > 3) {
 445                 common_log(LOG_ERR, "Bad queue breakout specifier $spec");
 446             }
 447             if (in_array($parts[0], $this->activeGroups)) {
 448                 $subs[] = $this->base . $spec;
 449             }
 450         }
 451         return array_unique($subs);
 452     }
 453
 454     /**
 455      * Handle and acknowledge an event that's come in through a queue.
 456      *
 457      * If the queue handler reports failure, the message is requeued for later.
 458      * Missing notices or handler classes will drop the message.
 459      *
 460      * Side effects: in multi-site mode, may reset site configuration to
 461      * match the site that queued the event.
 462      *
 463      * @param StompFrame $frame
 464      * @return bool success
 465      */
 466     protected function handleItem($frame)
 467     {
 468         $host = $this->cons[$this->defaultIdx]->getServer();
 469         $message = unserialize($frame->body);
 470         $site = $message['site'];
 471         $queue = $message['handler'];
 472
 473         if ($this->isDeadletter($frame, $message)) {
 474             $this->stats('deadletter', $queue);
 475                 return false;
 476         }
 477
 478         // @fixme detect failing site switches
 479         $this->switchSite($site);
 480
 481         $item = $this->decode($message['payload']);
 482         if (empty($item)) {
 483             $this->_log(LOG_ERR, "Skipping empty or deleted item in queue $queue from $host");
 484             $this->stats('baditem', $queue);
 485             return false;
 486         }
 487         $info = $this->logrep($item) . " posted at " .
 488                 $frame->headers['created'] . " in queue $queue from $host";
 489         $this->_log(LOG_DEBUG, "Dequeued $info");
 490
 491         $handler = $this->getHandler($queue);
 492         if (!$handler) {
 493             $this->_log(LOG_ERR, "Missing handler class; skipping $info");
 494             $this->stats('badhandler', $queue);
 495             return false;
 496         }
 497
 498         try {
 499             $ok = $handler->handle($item);
 500         } catch (Exception $e) {
 501             $this->_log(LOG_ERR, "Exception on queue $queue: " . $e->getMessage());
 502             $ok = false;
 503         }
 504
 505         if ($ok) {
 506             $this->_log(LOG_INFO, "Successfully handled $info");
 507             $this->stats('handled', $queue);
 508         } else {
 509             $this->_log(LOG_WARNING, "Failed handling $info");
 510             // Requeing moves the item to the end of the line for its next try.
 511             // @fixme add a manual retry count
 512             $this->enqueue($item, $queue);
 513             $this->stats('requeued', $queue);
 514         }
 515
 516         return $ok;
 517     }
 518
 519     /**
 520      * Check if a redelivered message has been run through enough
 521      * that we're going to give up on it.
 522      *
 523      * @param StompFrame $frame
 524      * @param array $message unserialized message body
 525      * @return boolean true if we should discard
 526      */
 527     protected function isDeadLetter($frame, $message)
 528     {
 529         if (isset($frame->headers['redelivered']) && $frame->headers['redelivered'] == 'true') {
 530                 // Message was redelivered, possibly indicating a previous failure.
 531             $msgId = $frame->headers['message-id'];
 532             $site = $message['site'];
 533             $queue = $message['handler'];
 534                 $msgInfo = "message $msgId for $site in queue $queue";
 535
 536                 $deliveries = $this->incDeliveryCount($msgId);
 537                 if ($deliveries > common_config('queue', 'max_retries')) {
 538                         $info = "DEAD-LETTER FILE: Gave up after retry $deliveries on $msgInfo";
 539
 540                         $outdir = common_config('queue', 'dead_letter_dir');
 541                         if ($outdir) {
 542                         $filename = $outdir . "/$site-$queue-" . rawurlencode($msgId);
 543                         $info .= ": dumping to $filename";
 544                         file_put_contents($filename, $message['payload']);
 545                         }
 546
 547                         common_log(LOG_ERR, $info);
 548                         return true;
 549                 } else {
 550                     common_log(LOG_INFO, "retry $deliveries on $msgInfo");
 551                 }
 552         }
 553         return false;
 554     }
 555
 556     /**
 557      * Update count of times we've re-encountered this message recently,
 558      * triggered when we get a message marked as 'redelivered'.
 559      *
 560      * Requires a CLI-friendly cache configuration.
 561      *
 562      * @param string $msgId message-id header from message
 563      * @return int number of retries recorded
 564      */
 565     function incDeliveryCount($msgId)
 566     {
 567             $count = 0;
 568             $cache = common_memcache();
 569             if ($cache) {
 570                     $key = 'statusnet:stomp:message-retries:' . $msgId;
 571                     $count = $cache->increment($key);
 572                     if (!$count) {
 573                             $count = 1;
 574                             $cache->set($key, $count, null, 3600);
 575                             $got = $cache->get($key);
 576                     }
 577             }
 578             return $count;
 579     }
 580
 581     /**
 582      * Process a control signal broadcast.
 583      *
 584      * @param int $idx connection index
 585      * @param array $frame Stomp frame
 586      * @return bool true to continue; false to stop further processing.
 587      */
 588     protected function handleControlSignal($idx, $frame)
 589     {
 590         $message = trim($frame->body);
 591         if (strpos($message, ':') !== false) {
 592             list($event, $param) = explode(':', $message, 2);
 593         } else {
 594             $event = $message;
 595             $param = '';
 596         }
 597
 598         $shutdown = false;
 599
 600         if ($event == 'shutdown') {
 601             $this->master->requestShutdown();
 602             $shutdown = true;
 603         } else if ($event == 'restart') {
 604             $this->master->requestRestart();
 605             $shutdown = true;
 606         } else if ($event == 'update') {
 607             $this->updateSiteConfig($param);
 608         } else {
 609             $this->_log(LOG_ERR, "Ignoring unrecognized control message: $message");
 610         }
 611         return $shutdown;
 612     }
 613
 614     /**
 615      * Switch site, if necessary, and reset current handler assignments
 616      * @param string $site
 617      */
 618     function switchSite($site)
 619     {
 620         if ($site != StatusNet::currentSite()) {
 621             $this->stats('switch');
 622             StatusNet::switchSite($site);
 623             $this->initialize();
 624         }
 625     }
 626
 627     /**
 628      * (Re)load runtime configuration for a given site by nickname,
 629      * triggered by a broadcast to the 'statusnet-control' topic.
 630      *
 631      * Configuration changes in database should update, but config
 632      * files might not.
 633      *
 634      * @param array $frame Stomp frame
 635      * @return bool true to continue; false to stop further processing.
 636      */
 637     protected function updateSiteConfig($nickname)
 638     {
 639         $sn = Status_network::staticGet($nickname);
 640         if ($sn) {
 641             $this->switchSite($nickname);
 642             if (!in_array($nickname, $this->sites)) {
 643                 $this->addSite();
 644             }
 645             $this->stats('siteupdate');
 646         } else {
 647             $this->_log(LOG_ERR, "Ignoring ping for unrecognized new site $nickname");
 648         }
 649     }
 650
 651     /**
 652      * Combines the queue_basename from configuration with the
 653      * group name for this queue to give eg:
 654      *
 655      * /queue/statusnet/main
 656      * /queue/statusnet/main/distrib
 657      * /queue/statusnet/xmpp/xmppout/site01
 658      *
 659      * @param string $queue
 660      * @return string
 661      */
 662     protected function queueName($queue)
 663     {
 664         $group = $this->queueGroup($queue);
 665         $site = StatusNet::currentSite();
 666
 667         $specs = array("$group/$queue/$site",
 668                        "$group/$queue");
 669         foreach ($specs as $spec) {
 670             if (in_array($spec, $this->breakout)) {
 671                 return $this->base . $spec;
 672             }
 673         }
 674         return $this->base . $group;
 675     }
 676
 677     /**
 678      * Get the breakout mode for the given queue on the current site.
 679      *
 680      * @param string $queue
 681      * @return string one of 'shared', 'handler', 'site'
 682      */
 683     protected function breakoutMode($queue)
 684     {
 685         $breakout = common_config('queue', 'breakout');
 686         if (isset($breakout[$queue])) {
 687             return $breakout[$queue];
 688         } else if (isset($breakout['*'])) {
 689             return $breakout['*'];
 690         } else {
 691             return 'shared';
 692         }
 693     }
 694
 695     protected function begin($idx)
 696     {
 697         if ($this->useTransactions) {
 698             if (!empty($this->transaction[$idx])) {
 699                 throw new Exception("Tried to start transaction in the middle of a transaction");
 700             }
 701             $this->transactionCount[$idx]++;
 702             $this->transaction[$idx] = $this->master->id . '-' . $this->transactionCount[$idx] . '-' . time();
 703             $this->cons[$idx]->begin($this->transaction[$idx]);
 704         }
 705     }
 706
 707     protected function ack($idx, $frame)
 708     {
 709         if ($this->useAcks) {
 710             if ($this->useTransactions) {
 711                 if (empty($this->transaction[$idx])) {
 712                     throw new Exception("Tried to ack but not in a transaction");
 713                 }
 714                 $this->cons[$idx]->ack($frame, $this->transaction[$idx]);
 715             } else {
 716                 $this->cons[$idx]->ack($frame);
 717             }
 718         }
 719     }
 720
 721     protected function commit($idx)
 722     {
 723         if ($this->useTransactions) {
 724             if (empty($this->transaction[$idx])) {
 725                 throw new Exception("Tried to commit but not in a transaction");
 726             }
 727             $this->cons[$idx]->commit($this->transaction[$idx]);
 728             $this->transaction[$idx] = null;
 729         }
 730     }
 731
 732     protected function rollback($idx)
 733     {
 734         if ($this->useTransactions) {
 735             if (empty($this->transaction[$idx])) {
 736                 throw new Exception("Tried to rollback but not in a transaction");
 737             }
 738             $this->cons[$idx]->commit($this->transaction[$idx]);
 739             $this->transaction[$idx] = null;
 740         }
 741     }
 742 }
 743