--- /dev/null
+<?php
+/**
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 2 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License along
+ * with this program; if not, write to the Free Software Foundation, Inc.,
+ * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
+ * http://www.gnu.org/copyleft/gpl.html
+ *
+ * @file
+ */
+
+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;
+
+/**
+ * Log handler sends log events to a kafka server.
+ *
+ * Constructor options array arguments:
+ * * alias: map from monolog channel to kafka topic name. When no
+ * alias exists the topic "monolog_$channel" will be used.
+ * * swallowExceptions: Swallow exceptions that occur while talking to
+ * kafka. Defaults to false.
+ * * logExceptions: Log exceptions talking to kafka here. Either null,
+ * the name of a channel to log to, or an object implementing
+ * FormatterInterface. Defaults to null.
+ *
+ * Requires the nmred/kafka-php library, version >= 1.3.0
+ *
+ * @since 1.26
+ * @author Erik Bernhardson <ebernhardson@wikimedia.org>
+ * @copyright © 2015 Erik Bernhardson and Wikimedia Foundation.
+ */
+class KafkaHandler extends AbstractProcessingHandler {
+ /**
+ * @var Produce Sends requests to kafka
+ */
+ protected $produce;
+
+ /**
+ * @var array Optional handler configuration
+ */
+ protected $options;
+
+ /**
+ * @var array Map from topic name to partition this request produces to
+ */
+ protected $partitions = [];
+
+ /**
+ * @var array defaults for constructor options
+ */
+ 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,
+ ];
+
+ /**
+ * @param Produce $produce Kafka instance to produce through
+ * @param array $options optional handler configuration
+ * @param int $level The minimum logging level at which this handler will be triggered
+ * @param bool $bubble Whether the messages that are handled can bubble up the stack or not
+ */
+ 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 );
+ }
+
+ /**
+ * Constructs the necessary support objects and returns a KafkaHandler
+ * instance.
+ *
+ * @param string[] $kafkaServers
+ * @param array $options
+ * @param int $level The minimum logging level at which this handle will be triggered
+ * @param bool $bubble Whether the messages that are handled can bubble the stack or not
+ * @return KafkaHandler
+ */
+ 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 );
+ }
+
+ /**
+ * @inheritDoc
+ */
+ protected function write( array $record ) {
+ if ( $record['formatted'] !== null ) {
+ $this->addMessages( $record['channel'], [ $record['formatted'] ] );
+ $this->send();
+ }
+ }
+
+ /**
+ * @inheritDoc
+ */
+ 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();
+ }
+
+ /**
+ * Send any records in the kafka client internal queue.
+ */
+ 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 );
+ }
+ }
+ }
+
+ /**
+ * @param string $topic Name of topic to get partition for
+ * @return int|null The random partition to produce to for this request,
+ * or null if a partition could not be determined.
+ */
+ 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];
+ }
+
+ /**
+ * Adds records for a channel to the Kafka client internal queue.
+ *
+ * @param string $channel Name of Monolog channel records belong to
+ * @param array $records List of records to append
+ */
+ protected function addMessages( $channel, array $records ) {
+ if ( isset( $this->options['alias'][$channel] ) ) {
+ $topic = $this->options['alias'][$channel];
+ } else {
+ $topic = "monolog_$channel";
+ }
+ $partition = $this->getRandomPartition( $topic );
+ if ( $partition !== null ) {
+ $this->produce->setMessages( $topic, $partition, $records );
+ }
+ }
+
+ /**
+ * @param string $message PSR3 compatible message string
+ * @param array $context PSR3 compatible log context
+ * @return bool true if caller should ignore warning
+ */
+ protected function warning( $message, array $context = [] ) {
+ if ( $this->options['logExceptions'] instanceof LoggerInterface ) {
+ $this->options['logExceptions']->warning( $message, $context );
+ }
+ return $this->options['swallowExceptions'];
+ }
+}
scripts.mit.edu / autoinstallsdev / mediawiki.git / blobdiff