KafkaHandler.php

Go to the documentation of this file.

<?php
namespace MediaWiki\Logger\Monolog;
 
use Kafka\MetaDataFromKafka;
use Kafka\Produce;
use Kafka\Protocol\Decoder;
use MediaWiki\Logger\LoggerFactory;
use Monolog\Handler\AbstractProcessingHandler;
use Monolog\Logger;
use Psr\Log\LoggerInterface;
 
class KafkaHandler extends AbstractProcessingHandler {
    protected $produce;
 
    protected $options;
 
    protected $partitions = [];
 
    private static $defaultOptions = [
        'alias' => [], // map from monolog channel to kafka topic
        'swallowExceptions' => false, // swallow exceptions sending records
        'logExceptions' => null, // A PSR3 logger to inform about errors
        'requireAck' => 0,
    ];
 
    public function __construct(
        Produce $produce, array $options, $level = Logger::DEBUG, $bubble = true
    ) {
        parent::__construct( $level, $bubble );
        $this->produce = $produce;
        $this->options = array_merge( self::$defaultOptions, $options );
    }
 
    public static function factory(
        $kafkaServers, array $options = [], $level = Logger::DEBUG, $bubble = true
    ) {
        $metadata = new MetaDataFromKafka( $kafkaServers );
        $produce = new Produce( $metadata );
 
        if ( isset( $options['sendTimeout'] ) ) {
            $timeOut = $options['sendTimeout'];
            $produce->getClient()->setStreamOption( 'SendTimeoutSec', 0 );
            $produce->getClient()->setStreamOption( 'SendTimeoutUSec',
                intval( $timeOut * 1000000 )
            );
        }
        if ( isset( $options['recvTimeout'] ) ) {
            $timeOut = $options['recvTimeout'];
            $produce->getClient()->setStreamOption( 'RecvTimeoutSec', 0 );
            $produce->getClient()->setStreamOption( 'RecvTimeoutUSec',
                intval( $timeOut * 1000000 )
            );
        }
        if ( isset( $options['logExceptions'] ) && is_string( $options['logExceptions'] ) ) {
            $options['logExceptions'] = LoggerFactory::getInstance( $options['logExceptions'] );
        }
 
        if ( isset( $options['requireAck'] ) ) {
            $produce->setRequireAck( $options['requireAck'] );
        }
 
        return new self( $produce, $options, $level, $bubble );
    }
 
    protected function write( array $record ) {
        if ( $record['formatted'] !== null ) {
            $this->addMessages( $record['channel'], [ $record['formatted'] ] );
            $this->send();
        }
    }
 
    public function handleBatch( array $batch ) {
        $channels = [];
        foreach ( $batch as $record ) {
            if ( $record['level'] < $this->level ) {
                continue;
            }
            $channels[$record['channel']][] = $this->processRecord( $record );
        }
 
        $formatter = $this->getFormatter();
        foreach ( $channels as $channel => $records ) {
            $messages = [];
            foreach ( $records as $idx => $record ) {
                $message = $formatter->format( $record );
                if ( $message !== null ) {
                    $messages[] = $message;
                }
            }
            if ( $messages ) {
                $this->addMessages( $channel, $messages );
            }
        }
 
        $this->send();
    }
 
    protected function send() {
        try {
            $response = $this->produce->send();
        } catch ( \Kafka\Exception $e ) {
            $ignore = $this->warning(
                'Error sending records to kafka: {exception}',
                [ 'exception' => $e ] );
            if ( !$ignore ) {
                throw $e;
            } else {
                return;
            }
        }
 
        if ( is_bool( $response ) ) {
            return;
        }
 
        $errors = [];
        foreach ( $response as $topicName => $partitionResponse ) {
            foreach ( $partitionResponse as $partition => $info ) {
                if ( $info['errCode'] === 0 ) {
                    // no error
                    continue;
                }
                $errors[] = sprintf(
                    'Error producing to %s (errno %d): %s',
                    $topicName,
                    $info['errCode'],
                    Decoder::getError( $info['errCode'] )
                );
            }
        }
 
        if ( $errors ) {
            $error = implode( "\n", $errors );
            if ( !$this->warning( $error ) ) {
                throw new \RuntimeException( $error );
            }
        }
    }
 
    protected function getRandomPartition( $topic ) {
        if ( !array_key_exists( $topic, $this->partitions ) ) {
            try {
                $partitions = $this->produce->getAvailablePartitions( $topic );
            } catch ( \Kafka\Exception $e ) {
                $ignore = $this->warning(
                    'Error getting metadata for kafka topic {topic}: {exception}',
                    [ 'topic' => $topic, 'exception' => $e ] );
                if ( $ignore ) {
                    return null;
                }
                throw $e;
            }
            if ( $partitions ) {
                $key = array_rand( $partitions );
                $this->partitions[$topic] = $partitions[$key];
            } else {
                $details = $this->produce->getClient()->getTopicDetail( $topic );
                $ignore = $this->warning(
                    'No partitions available for kafka topic {topic}',
                    [ 'topic' => $topic, 'kafka' => $details ]
                );
                if ( !$ignore ) {
                    throw new \RuntimeException( "No partitions available for kafka topic $topic" );
                }
                $this->partitions[$topic] = null;
            }
        }
        return $this->partitions[$topic];
    }
 
    protected function addMessages( $channel, array $records ) {
        $topic = $this->options['alias'][$channel] ?? "monolog_$channel";
        $partition = $this->getRandomPartition( $topic );
        if ( $partition !== null ) {
            $this->produce->setMessages( $topic, $partition, $records );
        }
    }
 
    protected function warning( $message, array $context = [] ) {
        if ( $this->options['logExceptions'] instanceof LoggerInterface ) {
            $this->options['logExceptions']->warning( $message, $context );
        }
        return $this->options['swallowExceptions'];
    }
}