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