TransactionProfiler.php

Go to the documentation of this file.

<?php
namespace Wikimedia\Rdbms;
 
use Liuggio\StatsdClient\Factory\StatsdDataFactoryInterface;
use NullStatsdDataFactory;
use Psr\Log\LoggerAwareInterface;
use Psr\Log\LoggerInterface;
use Psr\Log\NullLogger;
use RuntimeException;
use StatsdAwareInterface;
use Wikimedia\ScopedCallback;
 
class TransactionProfiler implements LoggerAwareInterface, StatsdAwareInterface {
    private $logger;
    private $stats;
    private $expect;
    private $hits;
    private $violations;
    private $silenced;
 
    private $dbTrxHoldingLocks;
 
    private $dbTrxMethodTimes;
 
    private $method;
 
    private $wallClockOverride;
 
    private const DB_LOCK_THRESHOLD_SEC = 3.0;
    private const EVENT_THRESHOLD_SEC = 0.25;
 
    private const EVENT_NAMES = [
        'writes',
        'queries',
        'conns',
        'masterConns',
        'maxAffected',
        'readQueryRows',
        'readQueryTime',
        'writeQueryTime'
    ];
 
    private const COUNTER_EVENT_NAMES = [
        'writes',
        'queries',
        'conns',
        'masterConns'
    ];
 
    private const FLD_LIMIT = 0;
    private const FLD_FNAME = 1;
 
    public const EXPECTATION_ANY = 'any';
    public const EXPECTATION_REPLICAS_ONLY = 'replicas-only';
 
    public function __construct() {
        $this->initPlaceholderExpectations();
 
        $this->dbTrxHoldingLocks = [];
        $this->dbTrxMethodTimes = [];
 
        $this->silenced = array_fill_keys( self::EVENT_NAMES, 0 );
 
        $this->setLogger( new NullLogger() );
        $this->setStatsdDataFactory( new NullStatsdDataFactory() );
    }
 
    public function setLogger( LoggerInterface $logger ) {
        $this->logger = $logger;
    }
 
    public function setStatsdDataFactory( StatsdDataFactoryInterface $statsFactory ) {
        $this->stats = $statsFactory;
    }
 
    public function setRequestMethod( ?string $method ) {
        $this->method = $method;
    }
 
    public function setSilenced( bool $value ) {
        wfDeprecated( __METHOD__, '1.40' );
 
        $delta = $value ? 1 : -1;
        foreach ( self::EVENT_NAMES as $event ) {
            $this->silenced[$event] += $delta;
        }
 
        return false;
    }
 
    public function silenceForScope( string $type = self::EXPECTATION_ANY ) {
        if ( $type === self::EXPECTATION_REPLICAS_ONLY ) {
            $events = [];
            foreach ( [ 'writes', 'masterConns' ] as $event ) {
                if ( $this->expect[$event][self::FLD_LIMIT] === 0 ) {
                    $events[] = $event;
                }
            }
        } else {
            $events = self::EVENT_NAMES;
        }
 
        foreach ( $events as $event ) {
            ++$this->silenced[$event];
        }
 
        return new ScopedCallback( function () use ( $events ) {
            foreach ( $events as $event ) {
                --$this->silenced[$event];
            }
        } );
    }
 
    public function setExpectation( string $event, $limit, string $fname ) {
        if ( !isset( $this->expect[$event] ) ) {
            return; // obsolete/bogus expectation
        }
 
        if ( $limit <= $this->expect[$event][self::FLD_LIMIT] ) {
            // New limit is more restrictive
            $this->expect[$event] = [
                self::FLD_LIMIT => $limit,
                self::FLD_FNAME => $fname
            ];
        }
    }
 
    public function setExpectations( array $expects, string $fname ) {
        foreach ( $expects as $event => $value ) {
            $this->setExpectation( $event, $value, $fname );
        }
    }
 
    public function resetExpectations() {
        $this->initPlaceholderExpectations();
    }
 
    public function redefineExpectations( array $expects, string $fname ) {
        $this->initPlaceholderExpectations();
        $this->setExpectations( $expects, $fname );
    }
 
    public function recordConnection( $server, $db, bool $isPrimary ) {
        // Report when too many connections happen...
        if ( $this->pingAndCheckThreshold( 'conns' ) ) {
            $this->reportExpectationViolated(
                'conns',
                "[connect to $server ($db)]",
                $this->hits['conns']
            );
        }
 
        // Report when too many primary connections happen...
        if ( $isPrimary && $this->pingAndCheckThreshold( 'masterConns' ) ) {
            $this->reportExpectationViolated(
                'masterConns',
                "[connect to $server ($db)]",
                $this->hits['masterConns']
            );
        }
    }
 
    public function transactionWritingIn( $server, $db, string $id ) {
        $name = "{$db} {$server} TRX#$id";
        if ( isset( $this->dbTrxHoldingLocks[$name] ) ) {
            $this->logger->warning( "Nested transaction for '$name' - out of sync." );
        }
        $this->dbTrxHoldingLocks[$name] = [
            'start' => $this->getCurrentTime(),
            'conns' => [], // all connections involved
        ];
        $this->dbTrxMethodTimes[$name] = [];
 
        foreach ( $this->dbTrxHoldingLocks as $name => &$info ) {
            // Track all DBs in transactions for this transaction
            $info['conns'][$name] = 1;
        }
    }
 
    public function recordQueryCompletion(
        $query,
        float $sTime,
        bool $isWrite,
        ?int $rowCount,
        string $trxId,
        ?string $serverName = null
    ) {
        $eTime = $this->getCurrentTime();
        $elapsed = ( $eTime - $sTime );
 
        if ( $isWrite && $this->isAboveThreshold( $rowCount, 'maxAffected' ) ) {
            $this->reportExpectationViolated( 'maxAffected', $query, $rowCount, $trxId, $serverName );
        } elseif ( !$isWrite && $this->isAboveThreshold( $rowCount, 'readQueryRows' ) ) {
            $this->reportExpectationViolated( 'readQueryRows', $query, $rowCount, $trxId, $serverName );
        }
 
        // Report when too many writes/queries happen...
        if ( $this->pingAndCheckThreshold( 'queries' ) ) {
            $this->reportExpectationViolated( 'queries', $query, $this->hits['queries'], $trxId, $serverName );
        }
        if ( $isWrite && $this->pingAndCheckThreshold( 'writes' ) ) {
            $this->reportExpectationViolated( 'writes', $query, $this->hits['writes'], $trxId, $serverName );
        }
        // Report slow queries...
        if ( !$isWrite && $this->isAboveThreshold( $elapsed, 'readQueryTime' ) ) {
            $this->reportExpectationViolated( 'readQueryTime', $query, $elapsed, $trxId, $serverName );
        }
        if ( $isWrite && $this->isAboveThreshold( $elapsed, 'writeQueryTime' ) ) {
            $this->reportExpectationViolated( 'writeQueryTime', $query, $elapsed, $trxId, $serverName );
        }
 
        if ( !$this->dbTrxHoldingLocks ) {
            // Short-circuit
            return;
        } elseif ( !$isWrite && $elapsed < self::EVENT_THRESHOLD_SEC ) {
            // Not an important query nor slow enough
            return;
        }
 
        foreach ( $this->dbTrxHoldingLocks as $name => $info ) {
            $lastQuery = end( $this->dbTrxMethodTimes[$name] );
            if ( $lastQuery ) {
                // Additional query in the trx...
                $lastEnd = $lastQuery[2];
                if ( $sTime >= $lastEnd ) {
                    if ( ( $sTime - $lastEnd ) > self::EVENT_THRESHOLD_SEC ) {
                        // Add an entry representing the time spent doing non-queries
                        $this->dbTrxMethodTimes[$name][] = [ '...delay...', $lastEnd, $sTime ];
                    }
                    $this->dbTrxMethodTimes[$name][] = [ $query, $sTime, $eTime ];
                }
            } else {
                // First query in the trx...
                if ( $sTime >= $info['start'] ) {
                    $this->dbTrxMethodTimes[$name][] = [ $query, $sTime, $eTime ];
                }
            }
        }
    }
 
    public function transactionWritingOut(
        $server,
        $db,
        string $id,
        float $writeTime,
        int $affected
    ) {
        // Must match $name in transactionWritingIn()
        $name = "{$db} {$server} TRX#$id";
        if ( !isset( $this->dbTrxMethodTimes[$name] ) ) {
            $this->logger->warning( "Detected no transaction for '$name' - out of sync." );
            return;
        }
 
        $slow = false;
 
        // Warn if too much time was spend writing...
        if ( $this->isAboveThreshold( $writeTime, 'writeQueryTime' ) ) {
            $this->reportExpectationViolated(
                'writeQueryTime',
                "[transaction writes to {$db} at {$server}]",
                $writeTime,
                $id
            );
            $slow = true;
        }
        // Warn if too many rows were changed...
        if ( $this->isAboveThreshold( $affected, 'maxAffected' ) ) {
            $this->reportExpectationViolated(
                'maxAffected',
                "[transaction writes to {$db} at {$server}]",
                $affected,
                $id
            );
        }
        // Fill in the last non-query period...
        $lastQuery = end( $this->dbTrxMethodTimes[$name] );
        if ( $lastQuery ) {
            $now = $this->getCurrentTime();
            $lastEnd = $lastQuery[2];
            if ( ( $now - $lastEnd ) > self::EVENT_THRESHOLD_SEC ) {
                $this->dbTrxMethodTimes[$name][] = [ '...delay...', $lastEnd, $now ];
            }
        }
        // Check for any slow queries or non-query periods...
        foreach ( $this->dbTrxMethodTimes[$name] as $info ) {
            $elapsed = ( $info[2] - $info[1] );
            if ( $elapsed >= self::DB_LOCK_THRESHOLD_SEC ) {
                $slow = true;
                break;
            }
        }
        if ( $slow ) {
            $trace = '';
            foreach ( $this->dbTrxMethodTimes[$name] as $i => [ $query, $sTime, $end ] ) {
                $trace .= sprintf(
                    "%-2d %.3fs %s\n", $i, ( $end - $sTime ), $this->getGeneralizedSql( $query ) );
            }
            $this->logger->warning( "Sub-optimal transaction [{dbs}]:\n{trace}", [
                'dbs' => implode( ', ', array_keys( $this->dbTrxHoldingLocks[$name]['conns'] ) ),
                'trace' => $trace
            ] );
        }
        unset( $this->dbTrxHoldingLocks[$name] );
        unset( $this->dbTrxMethodTimes[$name] );
    }
 
    private function initPlaceholderExpectations() {
        $this->expect = array_fill_keys(
            self::EVENT_NAMES,
            [ self::FLD_LIMIT => INF, self::FLD_FNAME => null ]
        );
 
        $this->hits = array_fill_keys( self::COUNTER_EVENT_NAMES, 0 );
        $this->violations = array_fill_keys( self::EVENT_NAMES, 0 );
    }
 
    private function isAboveThreshold( $value, string $event ) {
        if ( $this->silenced[$event] > 0 ) {
            return false;
        }
 
        return ( $value > $this->expect[$event][self::FLD_LIMIT] );
    }
 
    private function pingAndCheckThreshold( string $event ) {
        if ( $this->silenced[$event] > 0 ) {
            return false;
        }
 
        $newValue = ++$this->hits[$event];
        $limit = $this->expect[$event][self::FLD_LIMIT];
 
        return ( $newValue > $limit );
    }
 
    private function reportExpectationViolated(
        $event,
        $query,
        $actual,
        ?string $trxId = null,
        ?string $serverName = null
    ) {
        $violations = ++$this->violations[$event];
        // First violation; check if this is a web request
        if ( $violations === 1 && $this->method !== null ) {
            $this->stats->increment( "rdbms_trxprofiler_warnings.$event.{$this->method}" );
        }
 
        $max = $this->expect[$event][self::FLD_LIMIT];
        $by = $this->expect[$event][self::FLD_FNAME];
 
        $message = "Expectation ($event <= $max) by $by not met (actual: {actualSeconds})";
        if ( $trxId ) {
            $message .= ' in trx #{trxId}';
        }
        $message .= ":\n{query}\n";
 
        $this->logger->warning(
            $message,
            [
                'measure' => $event,
                'maxSeconds' => $max,
                'by' => $by,
                'actualSeconds' => $actual,
                'query' => $this->getGeneralizedSql( $query ),
                'exception' => new RuntimeException(),
                'trxId' => $trxId,
                'fullQuery' => $this->getRawSql( $query ),
                'dbHost' => $serverName
            ]
        );
    }
 
    private function getGeneralizedSql( $query ) {
        return $query instanceof GeneralizedSql ? $query->stringify() : $query;
    }
 
    private function getRawSql( $query ) {
        return $query instanceof GeneralizedSql ? $query->getRawSql() : $query;
    }
 
    private function getCurrentTime() {
        return $this->wallClockOverride ?: microtime( true );
    }
 
    public function setMockTime( &$time ) {
        $this->wallClockOverride =& $time;
    }
}