diff options
author | Timo Tijhof <krinkle@fastmail.com> | 2022-03-25 00:06:08 +0000 |
---|---|---|
committer | Timo Tijhof <krinkle@fastmail.com> | 2022-03-25 00:07:22 +0000 |
commit | 2de79774e1eb4578502a496020079eb3b3a20fe6 (patch) | |
tree | 74bc81f88bcac0e82e406531d1a289d77c7ded1d /includes/debug | |
parent | 1fbbcb6371517cac6d8b8c918be1a94b75f62594 (diff) | |
download | mediawikicore-2de79774e1eb4578502a496020079eb3b3a20fe6.tar.gz mediawikicore-2de79774e1eb4578502a496020079eb3b3a20fe6.zip |
Remove deprecated EventRelayerKafka and KafkaHandler
Also remove the unmaintained kafka-php package from the from "suggested"
and "dev" composer dependencies, as this is now no longer used.
Change-Id: If5668974f417b627df95bce47db18d46fa03327c
Diffstat (limited to 'includes/debug')
-rw-r--r-- | includes/debug/logger/monolog/KafkaHandler.php | 278 |
1 files changed, 0 insertions, 278 deletions
diff --git a/includes/debug/logger/monolog/KafkaHandler.php b/includes/debug/logger/monolog/KafkaHandler.php deleted file mode 100644 index bfc0a47bc862..000000000000 --- a/includes/debug/logger/monolog/KafkaHandler.php +++ /dev/null @@ -1,278 +0,0 @@ -<?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 - * - * @deprecated since 1.38 - * @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 const DEFAULT_OPTIONS = [ - '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 - ) { - wfDeprecated( __CLASS__, '1.38' ); - parent::__construct( $level, $bubble ); - $this->produce = $produce; - $this->options = array_merge( self::DEFAULT_OPTIONS, $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 ): void { - if ( $record['formatted'] !== null ) { - $this->addMessages( $record['channel'], [ $record['formatted'] ] ); - $this->send(); - } - } - - /** - * @inheritDoc - * @phan-param array[] $batch - */ - public function handleBatch( array $batch ): void { - $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 ) { - $topic = $this->options['alias'][$channel] ?? "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']; - } -} |