TransactionProfiler.php

Go to the documentation of this file.

<?php
namespace Wikimedia\Rdbms;
 
use Psr\Log\LoggerAwareInterface;
use Psr\Log\LoggerInterface;
use Psr\Log\NullLogger;
use RuntimeException;
use Wikimedia\ScopedCallback;
 
class TransactionProfiler implements LoggerAwareInterface {
    private $logger;
    private $expect;
    private $hits;
 
    private $dbTrxHoldingLocks;
 
    private $dbTrxMethodTimes;
 
    private $silenced;
 
    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 function __construct() {
        $this->initPlaceholderExpectations();
 
        $this->dbTrxHoldingLocks = [];
        $this->dbTrxMethodTimes = [];
 
        $this->silenced = false;
 
        $this->setLogger( new NullLogger() );
    }
 
    public function setLogger( LoggerInterface $logger ) {
        $this->logger = $logger;
    }
 
    public function setSilenced( bool $value ) {
        $old = $this->silenced;
        $this->silenced = $value;
 
        return $old;
    }
 
    public function silenceForScope() {
        $oldSilenced = $this->setSilenced( true );
        return new ScopedCallback( function () use ( $oldSilenced ) {
            $this->setSilenced( $oldSilenced );
        } );
    }
 
    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 = "{$server} ({$db}) (TRX#$id)";
        if ( isset( $this->dbTrxHoldingLocks[$name] ) ) {
            $this->logger->warning( "Nested transaction for '$name' - out of sync." );
        }
        $this->dbTrxHoldingLocks[$name] = [
            'start' => microtime( true ),
            '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 = microtime( true );
        $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
    ) {
        $name = "{$server} ({$db}) (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 {$server} ({$db})]",
                $writeTime,
                $id
            );
            $slow = true;
        }
        // Warn if too many rows were changed...
        if ( $this->isAboveThreshold( $affected, 'maxAffected' ) ) {
            $this->reportExpectationViolated(
                'maxAffected',
                "[transaction writes to {$server} ({$db})]",
                $affected,
                $id
            );
        }
        // Fill in the last non-query period...
        $lastQuery = end( $this->dbTrxMethodTimes[$name] );
        if ( $lastQuery ) {
            $now = microtime( true );
            $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(
                    "%d\t%.6f\t%s\n", $i, ( $end - $sTime ), $this->getGeneralizedSql( $query ) );
            }
            $this->logger->warning( "Sub-optimal transaction on DB(s) [{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 );
    }
 
    private function isAboveThreshold( $value, string $expectation ) {
        return ( $value > $this->expect[$expectation][self::FLD_LIMIT] );
    }
 
    private function pingAndCheckThreshold( string $expectation ) {
        $newValue = ++$this->hits[$expectation];
 
        return ( $newValue > $this->expect[$expectation][self::FLD_LIMIT] );
    }
 
    private function reportExpectationViolated(
        $expectation,
        $query,
        $actual,
        ?string $trxId = null,
        ?string $serverName = null
    ) {
        if ( $this->silenced ) {
            return;
        }
 
        $max = $this->expect[$expectation][self::FLD_LIMIT];
        $by = $this->expect[$expectation][self::FLD_FNAME];
        $message = "Expectation ($expectation <=) $max by $by not met (actual: {actual})";
        if ( $trxId ) {
            $message .= ' in trx #{trxId}';
        }
        $message .= ":\n{query}\n";
        $this->logger->warning(
            $message,
            [
                'measure' => $expectation,
                '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;
    }
}