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
22 namespace Wikimedia\Rdbms
;
24 use Psr\Log\LoggerInterface
;
25 use Psr\Log\NullLogger
;
26 use Wikimedia\ScopedCallback
;
30 * Basic DB load monitor with no external dependencies
31 * Uses memcached to cache the replication lag for a short time
35 class LoadMonitor
implements ILoadMonitor
{
36 /** @var ILoadBalancer */
42 /** @var LoggerInterface */
43 protected $replLogger;
45 /** @var float Moving average ratio (e.g. 0.1 for 10% weight to new weight) */
46 private $movingAveRatio;
48 const VERSION
= 1; // cache key version
50 public function __construct(
51 ILoadBalancer
$lb, BagOStuff
$srvCache, BagOStuff
$cache, array $options = []
54 $this->srvCache
= $srvCache;
55 $this->mainCache
= $cache;
56 $this->replLogger
= new NullLogger();
58 $this->movingAveRatio
= isset( $options['movingAveRatio'] )
59 ?
$options['movingAveRatio']
63 public function setLogger( LoggerInterface
$logger ) {
64 $this->replLogger
= $logger;
67 public function scaleLoads( array &$weightByServer, $domain ) {
68 $serverIndexes = array_keys( $weightByServer );
69 $states = $this->getServerStates( $serverIndexes, $domain );
70 $coefficientsByServer = $states['weightScales'];
71 foreach ( $weightByServer as $i => $weight ) {
72 if ( isset( $coefficientsByServer[$i] ) ) {
73 $weightByServer[$i] = $weight * $coefficientsByServer[$i];
74 } else { // server recently added to config?
75 $host = $this->parent
->getServerName( $i );
76 $this->replLogger
->error( __METHOD__
. ": host $host not in cache" );
81 public function getLagTimes( array $serverIndexes, $domain ) {
82 $states = $this->getServerStates( $serverIndexes, $domain );
84 return $states['lagTimes'];
87 protected function getServerStates( array $serverIndexes, $domain ) {
88 $writerIndex = $this->parent
->getWriterIndex();
89 if ( count( $serverIndexes ) == 1 && reset( $serverIndexes ) == $writerIndex ) {
90 # Single server only, just return zero without caching
92 'lagTimes' => [ $writerIndex => 0 ],
93 'weightScales' => [ $writerIndex => 1.0 ]
97 $key = $this->getCacheKey( $serverIndexes );
98 # Randomize TTLs to reduce stampedes (4.0 - 5.0 sec)
99 $ttl = mt_rand( 4e6
, 5e6
) / 1e6
;
100 # Keep keys around longer as fallbacks
103 # (a) Check the local APC cache
104 $value = $this->srvCache
->get( $key );
105 if ( $value && $value['timestamp'] > ( microtime( true ) - $ttl ) ) {
106 $this->replLogger
->debug( __METHOD__
. ": got lag times ($key) from local cache" );
107 return $value; // cache hit
109 $staleValue = $value ?
: false;
111 # (b) Check the shared cache and backfill APC
112 $value = $this->mainCache
->get( $key );
113 if ( $value && $value['timestamp'] > ( microtime( true ) - $ttl ) ) {
114 $this->srvCache
->set( $key, $value, $staleTTL );
115 $this->replLogger
->debug( __METHOD__
. ": got lag times ($key) from main cache" );
117 return $value; // cache hit
119 $staleValue = $value ?
: $staleValue;
121 # (c) Cache key missing or expired; regenerate and backfill
122 if ( $this->mainCache
->lock( $key, 0, 10 ) ) {
123 # Let this process alone update the cache value
124 $cache = $this->mainCache
;
125 /** @noinspection PhpUnusedLocalVariableInspection */
126 $unlocker = new ScopedCallback( function () use ( $cache, $key ) {
127 $cache->unlock( $key );
129 } elseif ( $staleValue ) {
130 # Could not acquire lock but an old cache exists, so use it
136 $movAveRatio = $this->movingAveRatio
;
137 foreach ( $serverIndexes as $i ) {
138 if ( $i == $this->parent
->getWriterIndex() ) {
139 $lagTimes[$i] = 0; // master always has no lag
140 $weightScales[$i] = 1.0; // nominal weight
144 $conn = $this->parent
->getAnyOpenConnection( $i );
146 $close = false; // already open
148 $conn = $this->parent
->openConnection( $i, $domain );
149 $close = true; // new connection
152 $lastWeight = isset( $staleValue['weightScales'][$i] )
153 ?
$staleValue['weightScales'][$i]
155 $coefficient = $this->getWeightScale( $i, $conn ?
: null );
156 $newWeight = $movAveRatio * $coefficient +
( 1 - $movAveRatio ) * $lastWeight;
158 // Scale from 10% to 100% of nominal weight
159 $weightScales[$i] = max( $newWeight, .10 );
162 $lagTimes[$i] = false;
163 $host = $this->parent
->getServerName( $i );
164 $this->replLogger
->error(
165 __METHOD__
. ": host {db_server} is unreachable",
166 [ 'db_server' => $host ]
171 if ( $conn->getLBInfo( 'is static' ) ) {
174 $lagTimes[$i] = $conn->getLag();
175 if ( $lagTimes[$i] === false ) {
176 $host = $this->parent
->getServerName( $i );
177 $this->replLogger
->error(
178 __METHOD__
. ": host {db_server} is not replicating?",
179 [ 'db_server' => $host ]
185 # Close the connection to avoid sleeper connections piling up.
186 # Note that the caller will pick one of these DBs and reconnect,
187 # which is slightly inefficient, but this only matters for the lag
188 # time cache miss cache, which is far less common that cache hits.
189 $this->parent
->closeConnection( $conn );
193 # Add a timestamp key so we know when it was cached
195 'lagTimes' => $lagTimes,
196 'weightScales' => $weightScales,
197 'timestamp' => microtime( true )
199 $this->mainCache
->set( $key, $value, $staleTTL );
200 $this->srvCache
->set( $key, $value, $staleTTL );
201 $this->replLogger
->info( __METHOD__
. ": re-calculated lag times ($key)" );
207 * @param integer $index Server index
208 * @param IDatabase|null $conn Connection handle or null on connection failure
211 protected function getWeightScale( $index, IDatabase
$conn = null ) {
212 return $conn ?
1.0 : 0.0;
215 private function getCacheKey( array $serverIndexes ) {
216 sort( $serverIndexes );
217 // Lag is per-server, not per-DB, so key on the master DB name
218 return $this->srvCache
->makeGlobalKey(
221 $this->parent
->getServerName( $this->parent
->getWriterIndex() ),
222 implode( '-', $serverIndexes )