includes/debug/logger/monolog/KafkaHandler.php

   1 <?php
   2 /**
   3  * This program is free software; you can redistribute it and/or modify
   4  * it under the terms of the GNU General Public License as published by
   5  * the Free Software Foundation; either version 2 of the License, or
   6  * (at your option) any later version.
   7  *
   8  * This program is distributed in the hope that it will be useful,
   9  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  10  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  11  * GNU General Public License for more details.
  12  *
  13  * You should have received a copy of the GNU General Public License along
  14  * with this program; if not, write to the Free Software Foundation, Inc.,
  15  * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
  16  * http://www.gnu.org/copyleft/gpl.html
  17  *
  18  * @file
  19  */
  20
  21 namespace MediaWiki\Logger\Monolog;
  22
  23 use Kafka\MetaDataFromKafka;
  24 use Kafka\Produce;
  25 use MediaWiki\Logger\LoggerFactory;
  26 use Monolog\Handler\AbstractProcessingHandler;
  27 use Monolog\Logger;
  28 use Psr\Log\LoggerInterface;
  29
  30 /**
  31  * Log handler sends log events to a kafka server.
  32  *
  33  * Constructor options array arguments:
  34  * * alias: map from monolog channel to kafka topic name. When no
  35  *        alias exists the topic "monolog_$channel" will be used.
  36  * * swallowExceptions: Swallow exceptions that occur while talking to
  37  *    kafka. Defaults to false.
  38  * * logExceptions: Log exceptions talking to kafka here. Either null,
  39  *    the name of a channel to log to, or an object implementing
  40  *    FormatterInterface. Defaults to null.
  41  *
  42  * Requires the nmred/kafka-php library, version >= 1.3.0
  43  *
  44  * @since 1.26
  45  * @author Erik Bernhardson <ebernhardson@wikimedia.org>
  46  * @copyright © 2015 Erik Bernhardson and Wikimedia Foundation.
  47  */
  48 class KafkaHandler extends AbstractProcessingHandler {
  49         /**
  50          * @var Produce Sends requests to kafka
  51          */
  52         protected $produce;
  53
  54         /**
  55          * @var array Optional handler configuration
  56          */
  57         protected $options;
  58
  59         /**
  60          * @var array Map from topic name to partition this request produces to
  61          */
  62         protected $partitions = array();
  63
  64         /**
  65          * @var array defaults for constructor options
  66          */
  67         private static $defaultOptions = array(
  68                 'alias' => array(), // map from monolog channel to kafka topic
  69                 'swallowExceptions' => false, // swallow exceptions sending records
  70                 'logExceptions' => null, // A PSR3 logger to inform about errors
  71         );
  72
  73         /**
  74          * @param Produce $produce Kafka instance to produce through
  75          * @param array $options optional handler configuration
  76          * @param int $level The minimum logging level at which this handler will be triggered
  77          * @param bool $bubble Whether the messages that are handled can bubble up the stack or not
  78          */
  79         public function __construct(
  80                 Produce $produce, array $options, $level = Logger::DEBUG, $bubble = true
  81         ) {
  82                 parent::__construct( $level, $bubble );
  83                 $this->produce = $produce;
  84                 $this->options = array_merge( self::$defaultOptions, $options );
  85         }
  86
  87         /**
  88          * Constructs the necessary support objects and returns a KafkaHandler
  89          * instance.
  90          *
  91          * @param string[] $kafkaServers
  92          * @param array $options
  93          * @param int $level The minimum logging level at which this handle will be triggered
  94          * @param bool $bubble Whether the messages that are handled can bubble the stack or not
  95          * @return KafkaHandler
  96          */
  97         public static function factory(
  98                 $kafkaServers, array $options = array(), $level = Logger::DEBUG, $bubble = true
  99         ) {
 100                 $metadata = new MetaDataFromKafka( $kafkaServers );
 101                 $produce = new Produce( $metadata );
 102                 if ( isset( $options['logExceptions'] ) && is_string( $options['logExceptions'] ) ) {
 103                         $options['logExceptions'] = LoggerFactory::getInstance( $options['logExceptions'] );
 104                 }
 105                 return new self( $produce, $options, $level, $bubble );
 106         }
 107
 108         /**
 109          * {@inheritDoc}
 110          */
 111         protected function write( array $record ) {
 112                 if ( $record['formatted'] !== null ) {
 113                         $this->addMessages( $record['channel'], array( $record['formatted'] ) );
 114                         $this->send();
 115                 }
 116         }
 117
 118         /**
 119          * {@inheritDoc}
 120          */
 121         public function handleBatch( array $batch ) {
 122                 $channels = array();
 123                 foreach ( $batch as $record ) {
 124                         if ( $record['level'] < $this->level ) {
 125                                 continue;
 126                         }
 127                         $channels[$record['channel']][] = $this->processRecord( $record );
 128                 }
 129
 130                 $formatter = $this->getFormatter();
 131                 foreach ( $channels as $channel => $records ) {
 132                         $messages = array();
 133                         foreach ( $records as $idx => $record ) {
 134                                 $message = $formatter->format( $record );
 135                                 if ( $message !== null ) {
 136                                         $messages[] = $message;
 137                                 }
 138                         }
 139                         if ( $messages ) {
 140                                 $this->addMessages( $channel, $messages );
 141                         }
 142                 }
 143
 144                 $this->send();
 145         }
 146
 147         /**
 148          * Send any records in the kafka client internal queue.
 149          */
 150         protected function send() {
 151                 try {
 152                         $this->produce->send();
 153                 } catch ( \Kafka\Exception $e ) {
 154                         $ignore = $this->warning(
 155                                 'Error sending records to kafka: {exception}',
 156                                 array( 'exception' => $e ) );
 157                         if ( !$ignore ) {
 158                                 throw $e;
 159                         }
 160                 }
 161         }
 162
 163         /**
 164          * @param string $topic Name of topic to get partition for
 165          * @return int|null The random partition to produce to for this request,
 166          *  or null if a partition could not be determined.
 167          */
 168         protected function getRandomPartition( $topic ) {
 169                 if ( !array_key_exists( $topic, $this->partitions ) ) {
 170                         try {
 171                                 $partitions = $this->produce->getAvailablePartitions( $topic );
 172                         } catch ( \Kafka\Exception $e ) {
 173                                 $ignore = $this->warning(
 174                                         'Error getting metadata for kafka topic {topic}: {exception}',
 175                                         array( 'topic' => $topic, 'exception' => $e ) );
 176                                 if ( $ignore ) {
 177                                         return null;
 178                                 }
 179                                 throw $e;
 180                         }
 181                         if ( $partitions ) {
 182                                 $key = array_rand( $partitions );
 183                                 $this->partitions[$topic] = $partitions[$key];
 184                         } else {
 185                                 $details = $this->produce->getClient()->getTopicDetail( $topic );
 186                                 $ignore = $this->warning(
 187                                         'No partitions available for kafka topic {topic}',
 188                                         array( 'topic' => $topic, 'kafka' => $details )
 189                                 );
 190                                 if ( !$ignore ) {
 191                                         throw new \RuntimeException( "No partitions available for kafka topic $topic" );
 192                                 }
 193                                 $this->partitions[$topic] = null;
 194                         }
 195                 }
 196                 return $this->partitions[$topic];
 197         }
 198
 199         /**
 200          * Adds records for a channel to the Kafka client internal queue.
 201          *
 202          * @param string $channel Name of Monolog channel records belong to
 203          * @param array $records List of records to append
 204          */
 205         protected function addMessages( $channel, array $records ) {
 206                 if ( isset( $this->options['alias'][$channel] ) ) {
 207                         $topic = $this->options['alias'][$channel];
 208                 } else {
 209                         $topic = "monolog_$channel";
 210                 }
 211                 $partition = $this->getRandomPartition( $topic );
 212                 if ( $partition !== null ) {
 213                         $this->produce->setMessages( $topic, $partition, $records );
 214                 }
 215         }
 216
 217         /**
 218          * @param string $message PSR3 compatible message string
 219          * @param array $context PSR3 compatible log context
 220          * @return bool true if caller should ignore warning
 221          */
 222         protected function warning( $message, array $context = array() ) {
 223                 if ( $this->options['logExceptions'] instanceof LoggerInterface ) {
 224                         $this->options['logExceptions']->warning( $message, $context );
 225                 }
 226                 return $this->options['swallowExceptions'];
 227         }
 228 }