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.
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.
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
21 namespace MediaWiki\Logger\Monolog
;
23 use Kafka\MetaDataFromKafka
;
25 use MediaWiki\Logger\LoggerFactory
;
26 use Monolog\Handler\AbstractProcessingHandler
;
28 use Psr\Log\LoggerInterface
;
31 * Log handler sends log events to a kafka server.
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.
42 * Requires the nmred/kafka-php library, version >= 1.3.0
45 * @author Erik Bernhardson <ebernhardson@wikimedia.org>
46 * @copyright © 2015 Erik Bernhardson and Wikimedia Foundation.
48 class KafkaHandler
extends AbstractProcessingHandler
{
50 * @var Produce Sends requests to kafka
55 * @var array Optional handler configuration
60 * @var array Map from topic name to partition this request produces to
62 protected $partitions = array();
65 * @var array defaults for constructor options
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
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
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 );
86 * Constructs the necessary support objects and returns a KafkaHandler
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
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'] );
101 return new self( $produce, $options, $level, $bubble );
107 protected function write( array $record ) {
108 if ( $record['formatted'] !== null ) {
109 $this->addMessages( $record['channel'], array( $record['formatted'] ) );
117 public function handleBatch( array $batch ) {
119 foreach ( $batch as $record ) {
120 if ( $record['level'] < $this->level
) {
123 $channels[$record['channel']][] = $this->processRecord( $record );
126 $formatter = $this->getFormatter();
127 foreach ( $channels as $channel => $records ) {
129 foreach ( $records as $idx => $record ) {
130 $message = $formatter->format( $record );
131 if ( $message !== null ) {
132 $messages[] = $message;
136 $this->addMessages($channel, $messages);
144 * Send any records in the kafka client internal queue.
146 protected function send() {
148 $this->produce
->send();
149 } catch ( \Kafka\Exception
$e ) {
150 $ignore = $this->warning(
151 'Error sending records to kafka: {exception}',
152 array( 'exception' => $e ) );
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.
164 protected function getRandomPartition( $topic ) {
165 if ( !array_key_exists( $topic, $this->partitions
) ) {
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 ) );
178 $key = array_rand( $partitions );
179 $this->partitions
[$topic] = $partitions[$key];
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 )
187 throw new \
RuntimeException( "No partitions available for kafka topic $topic" );
189 $this->partitions
[$topic] = null;
192 return $this->partitions
[$topic];
196 * Adds records for a channel to the Kafka client internal queue.
198 * @param string $channel Name of Monolog channel records belong to
199 * @param array $records List of records to append
201 protected function addMessages( $channel, array $records ) {
202 if ( isset( $this->options
['alias'][$channel] ) ) {
203 $topic = $this->options
['alias'][$channel];
205 $topic = "monolog_$channel";
207 $partition = $this->getRandomPartition( $topic );
208 if ( $partition !== null ) {
209 $this->produce
->setMessages( $topic, $partition, $records );
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
218 protected function warning( $message, array $context = array() ) {
219 if ( $this->options
['logExceptions'] instanceof LoggerInterface
) {
220 $this->options
['logExceptions']->warning( $message, $context );
222 return $this->options
['swallowExceptions'];