· 9 years ago · Oct 03, 2016, 04:24 PM
1<?php
2/**
3 * InterWorx Hosting Control Panel
4 *
5 * <pre>
6 * +----------------------------------------------------------------------+
7 * | Copyright (c) 2000-2010 InterWorx L.L.C., All Rights Reserved. |
8 * +----------------------------------------------------------------------+
9 * | Redistribution and use in source form, with or without modification |
10 * | is NOT permitted without consent from the copyright holder. |
11 * | |
12 * | THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS "AS IS" AND |
13 * | ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, |
14 * | THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A |
15 * | PARTICULAR PURPOSE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, |
16 * | EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, |
17 * | PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR |
18 * | PROFITS; OF BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY |
19 * | OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT |
20 * | (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE |
21 * | USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH |
22 * | DAMAGE. |
23 * +----------------------------------------------------------------------+
24 * </pre>
25 *
26 * @package InterWorx
27 * @subpackage CommandQueue
28 * @author Tim College <tcollege@interworx.com>
29 * @date Feb 3, 2010
30 */
31
32/**
33 * Command Queue class.
34 *
35 * @package InterWorx
36 * @subpackage CommandQueue
37 */
38class IWorx_Command_Queue implements Service_Interface {
39 const DAEMON_NAME = 'iworx-queue-d';
40
41 const MODE_CLUSTER_NODES = 1;
42 const MODE_REPLICATION = 2;
43 const MODE_RUN_ROOT = 4;
44
45 const CASCADE_FLAG_NAME = 'cascade_to_nodes';
46 const NODE_CHANGES_FLAG_NAME = 'enable_local_changes';
47
48 /**
49 * Singleton instance.
50 *
51 * @var IWorx_Command_Queue
52 */
53 static private $_instance;
54
55 /**
56 * Mode to replay events in.
57 *
58 * @var integer
59 */
60 private $_replay_mode = self::MODE_CLUSTER_NODES;
61
62 /**
63 * The last command ID inserted into the queue.
64 *
65 * @var integer
66 */
67 private $_last_command_id = 0;
68
69 /**
70 * Gets a singleton of the Command Queue.
71 *
72 * @return IWorx_Command_Queue
73 */
74 static public function getInstance() {
75 if( self::$_instance === null ) {
76 self::$_instance = new IWorx_Command_Queue();
77 }
78
79 return self::$_instance;
80 }
81
82 /**
83 * Gets the last command ID inserted into the queue.
84 *
85 * @return integer
86 */
87 public function getLastCommandId() {
88 return intval( $this->_last_command_id );
89 }
90
91 /**
92 * Replays the action it was called from.
93 *
94 * @return boolean
95 */
96 public function replayThisAction() {
97 if( IW::Env()->isManager() === false ) {
98 return true;
99 }
100 $ctrl_name = IW::FC()->getControllerName();
101 $action_name = IW::FC()->getAction();
102 $input = IW::Env()->getAllInput();
103
104 return $this->replayAnyAction( $ctrl_name, $action_name, $input );
105 }
106
107 /**
108 * Checks if an action should be replayed or not.
109 *
110 * @param integer $mode
111 * @return boolean
112 */
113 public function isReplayableAction( $mode ) {
114 // The run root mode can do whatever, whenever
115 if( $mode === self::MODE_RUN_ROOT ) {
116 return true;
117 }
118
119 // If we're not a cluster manager, nothing is replayable
120 if( IW::Env()->isManager() === false ) {
121 return false;
122 }
123
124 // If the cascade flag wasn't set, it's not replayable
125 $Input = new Input_Flag( self::CASCADE_FLAG_NAME );
126 if( $Input->getCurrentValue() !== '1' ) {
127 IWorxLog::debugCommandQueueLog( 'Cascade flag not set, discarding' );
128 return false;
129 }
130
131 return true;
132 }
133
134 /**
135 * Records a reply in the history table.
136 *
137 * @param Reply $Reply
138 * @param string $command_id
139 */
140 public function recordReply( Reply $Reply, $command_id ) {
141 $q_ipaddr = q( Ini::get( Ini::CLUSTER, 'node_id' ) );
142 $q_command_id = q( $command_id );
143 $q_success = q( ( $Reply->wasSuccessful() ) ? '1' : '0' );
144 $q_code = q( $Reply->getCode() );
145 $q_data = q( json_encode( $Reply->getData() ) );
146
147 /*
148 * This is stupid.
149 *
150 * Since the queries to get the command data get the most recent attempt, this
151 * particular failure can only happen AFTER success has happened. Since this
152 * was then a failed attempt, the queue would process it again.
153 *
154 * I'm not thrilled about this, but it was the best I could come up with. This
155 * error is recorded as having been successful, since it WAS...at one point...
156 * just not this most recent point.
157 *
158 * By doing it this way, the daemon won't attempt to re-run the command, but
159 * PHP will still see it as having failed.
160 */
161 if( $Reply->getCode() === Reply::ERROR_CQ_ALREADY_RUN ) {
162 $q_success = q( '1' );
163 }
164
165 $sql = "INSERT INTO cmd_history
166 SET
167 ipaddr = $q_ipaddr,
168 command_id = $q_command_id,
169 was_successful = $q_success,
170 reply_code = $q_code,
171 reply_data = $q_data,
172 replay_time = NOW()";
173
174 IWorxDbUtil::db_query( IW::DB(), $sql );
175
176 if( $q_success === '0' ) {
177 $sql = "SELECT history_id
178 FROM cmd_history
179 WHERE ipaddr = {$q_ipaddr}
180 AND command_id = {$q_command_id}
181 ORDER BY history_id DESC LIMIT 50";
182
183 $history = IWorxDbUtil::db_getCol( IW::DB(), $sql );
184
185 if( count( $history ) > 50 ) {
186 IWorxLog::log( "Truncating command history for command id {$command_id}" );
187
188 $q_history = q( $history );
189
190 $sql = "DELETE FROM cmd_history
191 WHERE ipaddr = {$q_ipaddr}
192 AND command_id = {$q_command_id}
193 AND history_id NOT IN( {$q_history} )";
194
195 IWorxDbUtil::db_query( IW::DB(), $sql );
196 }//end if
197 }//end if
198 }
199
200 /**
201 * Logs any action to be replayed on nodes.
202 *
203 * @param string $ctrl_name Controller
204 * @param string $action_name Action to replay
205 * @param array $input Input for the action
206 * @param integer $mode Replay mode - use constants - defaults to mode
207 * set with setReplayMode()
208 * @return boolean
209 */
210 public function replayAnyAction( $ctrl_name,
211 $action_name,
212 array $input = [],
213 $mode = null ) {
214 if( $mode === null ) {
215 $mode = $this->_replay_mode;
216 }
217
218 if( $this->isReplayableAction( $mode ) === false ) {
219 return true;
220 }
221
222 $input['IWLOG_REQUEST_ID'] = IW::Env()->getLogRequestId();
223 $input['IWLOG_SESSION_ID'] = IW::Env()->getLogSessionId();
224
225 $q_ctrl = q( $ctrl_name );
226 $q_action = q( $action_name );
227 $q_auth = q( $this->getAuthInfo() );
228 $q_input = q( json_encode( $input ) );
229 $q_mode = q( $mode );
230
231 $db = IW::DB();
232
233 $table = 'cmd_queue';
234
235 /*
236 * Only do this for cascaded actions.
237 */
238 if( $mode === self::MODE_CLUSTER_NODES ) {
239 if( IW::Env()->getEnvFlag( 'cache-command-queue' ) === true ) {
240 $table = 'cmd_queue_cache';
241
242 /*
243 * We do this here, and not in the upgrade script, to avoid a potential
244 * race where the table doesn't exist in time. Doing this here ensures we
245 * capture any strays, that might be triggered not by the upgrade script,
246 * but by a coincidental fively or a user action.
247 */
248 $sql = "CREATE TABLE IF NOT EXISTS `cmd_queue_cache` (
249 `command_id` int(6) unsigned NOT NULL AUTO_INCREMENT,
250 `ctrl_name` varchar(60) NOT NULL DEFAULT '',
251 `action_name` varchar(60) NOT NULL DEFAULT '',
252 `replay_mode` int(11) NOT NULL DEFAULT '0',
253 `auth` varchar(255) NOT NULL DEFAULT '',
254 `timestamp` datetime DEFAULT NULL,
255 `input` text NOT NULL,
256 PRIMARY KEY (`command_id`),
257 KEY `replay_mode` (`replay_mode`)
258 ) ENGINE=MyISAM";
259 IWorxDbUtil::db_query( $db, $sql );
260 }//end if
261 }//end if
262
263 $sql = "INSERT INTO {$table}
264 SET ctrl_name = $q_ctrl,
265 action_name = $q_action,
266 replay_mode = $q_mode,
267 auth = $q_auth,
268 timestamp = NOW(),
269 input = $q_input";
270
271 try {
272 IWorxDbUtil::db_query( $db, $sql );
273 } catch( IWorx_Exception_DbUtil $e ) {
274 return false;
275 }
276 $this->_last_command_id = IWorxDbUtil::db_lastInsertId( $db );
277 return true;
278 }
279
280 /**
281 * Checks if an action is recorded.
282 *
283 * @param string $ctrl_name
284 * @param string $action_name
285 * @return boolean
286 */
287 public function isARecordedAction( $ctrl_name, $action_name ) {
288 $q_ctrl = q( $ctrl_name );
289 $q_action = q( $action_name );
290
291 $msg = "Checking if $ctrl_name::$action_name is recorded";
292 IWorxLog::debugCommandQueueLog( $msg );
293
294 $sql = "SELECT trigger_id FROM cmd_triggers
295 WHERE ctrl_name = $q_ctrl
296 AND action_name = $q_action";
297
298 $trigger_id = IWorxDbUtil::db_getOne( IW::DB(), $sql );
299
300 return ( $trigger_id ) ? true : false;
301 }
302
303 /**
304 * Checks if an action is forced to be recorded no matter what.
305 *
306 * @param string $ctrl_name
307 * @param string $action_name
308 * @return boolean
309 */
310 public function isForceRecordedAction( $ctrl_name, $action_name ) {
311 $q_ctrl = q( $ctrl_name );
312 $q_action = q( $action_name );
313
314 $msg = "Checking if $ctrl_name::$action_name is forced to be recorded";
315 IWorxLog::debugCommandQueueLog( $msg );
316
317 $sql = "SELECT trigger_id FROM cmd_triggers
318 WHERE ctrl_name = $q_ctrl
319 AND action_name = $q_action
320 AND is_forced = 1";
321
322 $trigger_id = IWorxDbUtil::db_getOne( IW::DB(), $sql );
323
324 return ( $trigger_id ) ? true : false;
325 }
326
327 /**
328 * Gets an array of information about the nodes.
329 *
330 * @return array
331 */
332 public function getNodeStatuses() {
333 $nodes = NodeManager::getNodeInfoList();
334
335 $i = 0;
336
337 foreach( $nodes as $node ) {
338 $commands = $this->getUnplayedCommands( $node['ipaddr'] );
339
340 $nodes[ $i ]['commands'] = $commands;
341 $nodes[ $i ]['command_count'] = count( $commands );
342
343 $q_ip = q( $node['ipaddr'] );
344
345 $sql = "SELECT UNIX_TIMESTAMP( last_replay_time )
346 FROM cmd_replay_log
347 WHERE ipaddr = $q_ip";
348
349 $last_replay = IWorxDbUtil::db_getOne( IW::DB(), $sql );
350
351 $nodes[ $i ]['last_replay'] = $last_replay;
352
353 ++$i;
354 }
355
356 return $nodes;
357 }
358
359 /**
360 * Gets the unplayed commands for a node.
361 *
362 * @param string $node_id IP Address of the node
363 * @param integer $mode
364 * @return IWorx_Command_Queue_Command[]
365 */
366 public function getUnplayedCommands( $node_id, $mode = null ) {
367 if( $mode === null ) {
368 $mode = $this->_replay_mode;
369 }
370
371 $q_mode = q( $mode );
372 $q_node_id = q( $node_id );
373 $db = IW::DB();
374 /*
375 * OK, this query deserves some explanation, cause it's not blatantly obvious
376 * what's going on.
377 *
378 * q is the log of commands. That's pretty much what we want to get
379 *
380 * l contains the current position of the nodes. This is important, because
381 * without it, new nodes would try to play back the entire history, which would
382 * suck a fatty. Also, using the position in the join limits the number of rows
383 * to look at, which will help as the queue grows
384 *
385 * h is the history. This gets us whether the command has been attempted to be
386 * replayed, and the result.
387 *
388 * h2 is a self-join on the history. By looking for rows that h2 IS NULL, we
389 * get commands that have never been tried OR just the last attempt for commands
390 * that have been tried and failed. If we didn't have this self-join, failed
391 * commands would show up multiple times in the results, which would not be good
392 */
393 $sql = "SELECT q.command_id,
394 q.ctrl_name,
395 q.action_name,
396 q.replay_mode,
397 q.auth,
398 q.input,
399 h.reply_code,
400 h.was_successful
401 FROM cmd_queue q
402 INNER JOIN cmd_replay_log l ON l.ipaddr = $q_node_id
403 LEFT JOIN cmd_history h ON h.ipaddr = l.ipaddr
404 AND q.command_id = h.command_id
405 LEFT JOIN cmd_history h2 ON h2.command_id = h.command_id
406 AND h2.history_id > h.history_id
407 AND h2.ipaddr = h.ipaddr
408 WHERE q.command_id > l.last_command_id
409 AND ( h.was_successful IS NULL OR h.was_successful != 1 )
410 AND q.replay_mode = $q_mode
411 AND h2.history_id IS NULL
412 ORDER BY q.command_id";
413
414 $commands = IWorxDbUtil::db_getAll( $db, $sql );
415
416 return $commands;
417 }
418
419 /**
420 * Gets the command history.
421 *
422 * @param string $node_id
423 * @param integer $mode
424 * @param integer $limit
425 * @return array
426 */
427 public function getCommandHistory( $node_id, $mode = null, $limit = 10 ) {
428 if( $mode === null ) {
429 $mode = $this->_replay_mode;
430 }
431
432 $q_mode = q( $mode );
433 $q_node_id = q( $node_id );
434 $q_limit = intval( $limit );
435 $q_dupe = q( Reply::ERROR_CQ_ALREADY_RUN );
436
437 $sql = "SELECT q.command_id,
438 q.ctrl_name,
439 q.action_name,
440 UNIX_TIMESTAMP( h.replay_time ) as replay_time,
441 h.reply_code,
442 h.reply_data,
443 h.was_successful
444 FROM cmd_queue q
445 LEFT JOIN cmd_history h ON h.ipaddr = $q_node_id
446 AND q.command_id = h.command_id
447 LEFT JOIN cmd_history h2 ON h2.command_id = h.command_id
448 AND h2.history_id > h.history_id
449 AND h2.ipaddr = h.ipaddr
450 WHERE h.reply_code <> $q_dupe
451 AND h.was_successful IS NOT NULL
452 AND q.replay_mode = $q_mode
453 AND h2.history_id IS NULL
454 ORDER BY q.command_id DESC
455 LIMIT $q_limit";
456
457 $output = IWorxDbUtil::db_getAll( IW::DB(), $sql );
458
459 return $output;
460 }
461
462 /**
463 * Gets a single command from the queue.
464 *
465 * @param integer $command_id
466 * @return IWorx_Command_Queue_Command
467 */
468 public function getCommandData( $command_id ) {
469 $q_id = q( $command_id );
470 $q_node_id = q( Node::getNodeID() );
471 $db = IW::DB();
472
473 $sql = "SELECT q.command_id,
474 q.ctrl_name,
475 q.action_name,
476 q.replay_mode,
477 q.auth,
478 q.input,
479 h.reply_code,
480 h.was_successful,
481 h.reply_data
482 FROM cmd_queue q
483 LEFT JOIN cmd_history h ON h.ipaddr = $q_node_id
484 AND q.command_id = h.command_id
485 LEFT JOIN cmd_history h2 ON h2.command_id = h.command_id
486 AND h2.history_id > h.history_id
487 AND h.ipaddr = h2.ipaddr
488 WHERE q.command_id = $q_id
489 AND h2.history_id IS NULL";
490
491 return new IWorx_Command_Queue_Command( IWorxDbUtil::db_getRow( $db, $sql ) );
492 }
493
494 /**
495 * Gets an InterWorx object for routing with.
496 *
497 * @param object $user
498 * @return InterWorx
499 */
500 public function getInterWorxFromAuth( $user ) {
501 if( isset( $user->domain ) === true ) {
502 // SiteWorx
503 $IW = new SiteWorx( $user->domain );
504 $User = new SiteWorx_User( $user->user_id );
505 } else {
506 // NodeWorx
507 $IW = new NodeWorx( $user->nodeworx_id );
508 $User = new NodeWorx_User( $user->user_id );
509 }
510
511 $IW->setWorkingUser( $User );
512 return $IW;
513 }
514
515 /**
516 * Gets the authorization information for replaying the action later.
517 *
518 * @return string
519 */
520 public function getAuthInfo() {
521 $Session = IW::Env()->getActiveSession();
522
523 $output = (object) null;
524
525 switch( get_class( $Session ) ) {
526 case 'Session_NodeWorx':
527 // Currently hard-coded to the master account
528 $output->user_id = NodeWorx_User::MASTER_ID;
529 $output->nodeworx_id = NodeWorx::MASTER_ID;
530 break;
531
532 case 'Session_SiteWorx':
533 // Actions are always replayed as if the master user did them
534 /* @var $Session Session_SiteWorx */
535 $output->user_id = $Session->getInterWorx()->getMasterUser()->getId();
536 $output->siteworx_id = $Session->getInterWorx()->getId();
537 $output->domain = $Session->getDomain();
538 }
539
540 return json_encode( $output );
541 }
542
543 /**
544 * Sets the replay mode.
545 *
546 * @param integer $mode
547 */
548 public function setReplayMode( $mode ) {
549 $this->_replay_mode = $mode;
550 }
551
552 /**
553 * Gets the class name of the service.
554 *
555 * @return string
556 */
557 public function getServiceClassName() {
558 return __CLASS__;
559 }
560
561 /**
562 * Gets the controller name.
563 *
564 * @return string
565 */
566 public function getControllerName() {
567 return Ctrl_Util::convertClassNameToPath( 'Ctrl_Nodeworx_CommandQueue' );
568 }
569
570 /**
571 * Gets the generic type of the service.
572 *
573 * @return string
574 */
575 public function getServiceType() {
576 return '';
577 }
578
579 /**
580 * Gets a "normal looking" name.
581 *
582 * @return string
583 */
584 public function getGeneralName() {
585 return '##LG_COMMAND_QUEUE##';
586 }
587
588 /**
589 * Gets the service name.
590 *
591 * @return string
592 */
593 public function getServiceName() {
594 return self::DAEMON_NAME;
595 }
596
597 /**
598 * Gets an array of permissions required for the service.
599 *
600 * @return array
601 */
602 public function getRequiredPermissions() {
603 return Ctrl_Nodeworx_CommandQueue::getRequiredPerms();
604 }
605
606 /**
607 * Returns the page that the service should be linked to.
608 *
609 * @return string
610 */
611 public function getPage() {
612 return '/nodeworx/command/queue';
613 }
614
615 /**
616 * Gets an array of ports applicable to the service.
617 *
618 * @return array
619 */
620 static public function getPortNumbers() {
621 return array( 'N/A' );
622 }
623
624 /**
625 * Gets a string of the port numbers, comma-separated.
626 *
627 * @return string
628 */
629 public function getPortNumberString() {
630 return implode( ',', $this->getPortNumbers() );
631 }
632
633 /**
634 * Determines if the driver service is running or not.
635 *
636 * @return boolean
637 */
638 public function isRunning() {
639 $Daemon = new IWorx_Daemon( $this->getServiceName() );
640
641 return $Daemon->isRunning();
642 }
643
644 /**
645 * Starts the driver service.
646 *
647 * @return boolean
648 */
649 public function start() {
650 $cmd = Ini::get( Ini::IWORX_BIN, 'command-queue' );
651 $cmd .= ' --start';
652
653 $result = [];
654 $retval = -1;
655
656 IWorxExec::exec( $cmd, $result, $retval, IWorxExec::IN_BACKGROUND );
657
658 $Daemon = new IWorx_Daemon( self::DAEMON_NAME );
659 $i = 0;
660
661 // Because it launches in the background, we can't use the retval
662 while( $i < 10 && $Daemon->isRunning() === false ) {
663 usleep( 100000 );
664 ++$i;
665 }
666
667 return $Daemon->isRunning();
668 }
669
670 /**
671 * Stops the driver service.
672 *
673 * @param string $msg
674 * @return boolean
675 */
676 public function stop( $msg = 'Stopped by service interface' ) {
677 $cmd = Ini::get( Ini::IWORX_BIN, 'command-queue' );
678 $cmd .= ' --stop --stop-message "' . $msg . '"';
679
680 $result = [];
681 $retval = -1;
682
683 IWorxExec::exec( $cmd, $result, $retval );
684
685 return $retval === 0;
686 }
687
688 /**
689 * Restarts the driver service.
690 *
691 * @return boolean
692 */
693 public function restart() {
694 /*
695 * OK, we can't use the --restart command line, even though it exists. PHP
696 * hangs when the child process forks, and never returns. It was a real bitch
697 * to figure out a way around this that didn't require sleeping for 5 seconds
698 * or something stupid, and the start() and stop() already work
699 */
700 $this->stop( 'Restarted by service interface' );
701 return $this->start();
702 }
703}