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( Produce $produce, array $options, $level = Logger::DEBUG, $bubble = true ) {
  80                 parent::__construct( $level, $bubble );
  81                 $this->produce = $produce;
  82                 $this->options = array_merge( self::$defaultOptions, $options );
  83         }
  84
  85         /**
  86          * Constructs the necessary support objects and returns a KafkaHandler
  87          * instance.
  88          *
  89          * @param string[] $kafkaServers
  90          * @param array $options
  91          * @param int $level The minimum logging level at which this handle will be triggered
  92          * @param bool $bubble Whether the messages that are handled can bubble the stack or not
  93          * @return KafkaHandler
  94          */
  95         public static function factory( $kafkaServers, array $options = array(), $level = Logger::DEBUG, $bubble = true ) {
  96                 $metadata = new MetaDataFromKafka( $kafkaServers );
  97                 $produce = new Produce( $metadata );
  98                 if ( isset( $options['logExceptions'] ) && is_string( $options['logExceptions'] ) ) {
  99                         $options['logExceptions'] = LoggerFactory::getInstance( $options['logExceptions'] );
 100                 }
 101                 return new self( $produce, $options, $level, $bubble );
 102         }
 103
 104         /**
 105          * {@inheritDoc}
 106          */
 107         protected function write( array $record ) {
 108                 if ( $record['formatted'] !== null ) {
 109                         $this->addMessages( $record['channel'], array( $record['formatted'] ) );
 110                         $this->send();
 111                 }
 112         }
 113
 114         /**
 115          * {@inheritDoc}
 116          */
 117         public function handleBatch( array $batch ) {
 118                 $channels = array();
 119                 foreach ( $batch as $record ) {
 120                         if ( $record['level'] < $this->level ) {
 121                                 continue;
 122                         }
 123                         $channels[$record['channel']][] = $this->processRecord( $record );
 124                 }
 125
 126                 $formatter = $this->getFormatter();
 127                 foreach ( $channels as $channel => $records ) {
 128                         $messages = array();
 129                         foreach ( $records as $idx => $record ) {
 130                                 $message = $formatter->format( $record );
 131                                 if ( $message !== null ) {
 132                                         $messages[] = $message;
 133                                 }
 134                         }
 135                         if ( $messages ) {
 136                                 $this->addMessages( $channel, $messages );
 137                         }
 138                 }
 139
 140                 $this->send();
 141         }
 142
 143         /**
 144          * Send any records in the kafka client internal queue.
 145          */
 146         protected function send() {
 147                 try {
 148                         $this->produce->send();
 149                 } catch ( \Kafka\Exception $e ) {
 150                         $ignore = $this->warning(
 151                                 'Error sending records to kafka: {exception}',
 152                                 array( 'exception' => $e ) );
 153                         if ( !$ignore ) {
 154                                 throw $e;
 155                         }
 156                 }
 157         }
 158
 159         /**
 160          * @param string $topic Name of topic to get partition for
 161          * @return int|null The random partition to produce to for this request,
 162          *  or null if a partition could not be determined.
 163          */
 164         protected function getRandomPartition( $topic ) {
 165                 if ( !array_key_exists( $topic, $this->partitions ) ) {
 166                         try {
 167                                 $partitions = $this->produce->getAvailablePartitions( $topic );
 168                         } catch ( \Kafka\Exception $e ) {
 169                                 $ignore = $this->warning(
 170                                         'Error getting metadata for kafka topic {topic}: {exception}',
 171                                         array( 'topic' => $topic, 'exception' => $e ) );
 172                                 if ( $ignore ) {
 173                                         return null;
 174                                 }
 175                                 throw $e;
 176                         }
 177                         if ( $partitions ) {
 178                                 $key = array_rand( $partitions );
 179                                 $this->partitions[$topic] = $partitions[$key];
 180                         } else {
 181                                 $details = $this->produce->getClient()->getTopicDetail( $topic );
 182                                 $ignore = $this->warning(
 183                                         'No partitions available for kafka topic {topic}',
 184                                         array( 'topic' => $topic, 'kafka' => $details )
 185                                 );
 186                                 if ( !$ignore ) {
 187                                         throw new \RuntimeException( "No partitions available for kafka topic $topic" );
 188                                 }
 189                                 $this->partitions[$topic] = null;
 190                         }
 191                 }
 192                 return $this->partitions[$topic];
 193         }
 194
 195         /**
 196          * Adds records for a channel to the Kafka client internal queue.
 197          *
 198          * @param string $channel Name of Monolog channel records belong to
 199          * @param array $records List of records to append
 200          */
 201         protected function addMessages( $channel, array $records ) {
 202                 if ( isset( $this->options['alias'][$channel] ) ) {
 203                         $topic = $this->options['alias'][$channel];
 204                 } else {
 205                         $topic = "monolog_$channel";
 206                 }
 207                 $partition = $this->getRandomPartition( $topic );
 208                 if ( $partition !== null ) {
 209                         $this->produce->setMessages( $topic, $partition, $records );
 210                 }
 211         }
 212
 213         /**
 214          * @param string $message PSR3 compatible message string
 215          * @param array $context PSR3 compatible log context
 216          * @return bool true if caller should ignore warning
 217          */
 218         protected function warning( $message, array $context = array() ) {
 219                 if ( $this->options['logExceptions'] instanceof LoggerInterface ) {
 220                         $this->options['logExceptions']->warning( $message, $context );
 221                 }
 222                 return $this->options['swallowExceptions'];
 223         }
 224 }