59075e483e794a42db6633eb2dcc2b6682211f2d
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 use Psr\Log\LoggerInterface
;
25 * Basic DB load monitor with no external dependencies
26 * Uses memcached to cache the replication lag for a short time
30 class LoadMonitor
implements ILoadMonitor
{
31 /** @var ILoadBalancer */
37 /** @var LoggerInterface */
38 protected $replLogger;
40 /** @var float Moving average ratio (e.g. 0.1 for 10% weight to new weight) */
41 private $movingAveRatio;
43 public function __construct(
44 ILoadBalancer
$lb, BagOStuff
$srvCache, BagOStuff
$cache, array $options = []
47 $this->srvCache
= $srvCache;
48 $this->mainCache
= $cache;
49 $this->replLogger
= new \Psr\Log\
NullLogger();
51 $this->movingAveRatio
= isset( $options['movingAveRatio'] )
52 ?
$options['movingAveRatio']
56 public function setLogger( LoggerInterface
$logger ) {
57 $this->replLogger
= $logger;
60 public function scaleLoads( array &$weightByServer, $group = false, $domain = false ) {
61 $serverIndexes = array_keys( $weightByServer );
62 $states = $this->getServerStates( $serverIndexes, $domain );
63 $coefficientsByServer = $states['weightScales'];
64 foreach ( $weightByServer as $i => $weight ) {
65 $weightByServer[$i] = $weight * $coefficientsByServer[$i];
69 public function getLagTimes( array $serverIndexes, $domain ) {
70 $states = $this->getServerStates( $serverIndexes, $domain );
72 return $states['lagTimes'];
75 protected function getServerStates( array $serverIndexes, $domain ) {
76 if ( count( $serverIndexes ) == 1 && reset( $serverIndexes ) == 0 ) {
77 # Single server only, just return zero without caching
79 'lagTimes' => [ $this->parent
->getWriterIndex() => 0 ],
80 'weightScales' => [ $this->parent
->getWriterIndex() => 1 ]
84 $key = $this->getCacheKey();
85 # Randomize TTLs to reduce stampedes (4.0 - 5.0 sec)
86 $ttl = mt_rand( 4e6
, 5e6
) / 1e6
;
87 # Keep keys around longer as fallbacks
90 # (a) Check the local APC cache
91 $value = $this->srvCache
->get( $key );
92 if ( $value && $value['timestamp'] > ( microtime( true ) - $ttl ) ) {
93 $this->replLogger
->debug( __METHOD__
. ": got lag times ($key) from local cache" );
94 return $value; // cache hit
96 $staleValue = $value ?
: false;
98 # (b) Check the shared cache and backfill APC
99 $value = $this->mainCache
->get( $key );
100 if ( $value && $value['timestamp'] > ( microtime( true ) - $ttl ) ) {
101 $this->srvCache
->set( $key, $value, $staleTTL );
102 $this->replLogger
->debug( __METHOD__
. ": got lag times ($key) from main cache" );
104 return $value; // cache hit
106 $staleValue = $value ?
: $staleValue;
108 # (c) Cache key missing or expired; regenerate and backfill
109 if ( $this->mainCache
->lock( $key, 0, 10 ) ) {
110 # Let this process alone update the cache value
111 $cache = $this->mainCache
;
112 /** @noinspection PhpUnusedLocalVariableInspection */
113 $unlocker = new ScopedCallback( function () use ( $cache, $key ) {
114 $cache->unlock( $key );
116 } elseif ( $staleValue ) {
117 # Could not acquire lock but an old cache exists, so use it
123 $movAveRatio = $this->movingAveRatio
;
124 foreach ( $serverIndexes as $i ) {
125 if ( $i == $this->parent
->getWriterIndex() ) {
126 $lagTimes[$i] = 0; // master always has no lag
127 $weightScales[$i] = 1.0; // nominal weight
131 $conn = $this->parent
->getAnyOpenConnection( $i );
133 $close = false; // already open
135 $conn = $this->parent
->openConnection( $i, $domain );
136 $close = true; // new connection
139 $lastWeight = isset( $staleValue['weightScales'][$i] )
140 ?
$staleValue['weightScales'][$i]
142 $coefficient = $this->getWeightScale( $i, $conn ?
: null );
143 $newWeight = $movAveRatio * $coefficient +
( 1 - $movAveRatio ) * $lastWeight;
145 // Scale from 10% to 100% of nominal weight
146 $weightScales[$i] = max( $newWeight, .10 );
149 $lagTimes[$i] = false;
150 $host = $this->parent
->getServerName( $i );
151 $this->replLogger
->error( __METHOD__
. ": host $host is unreachable" );
155 $lagTimes[$i] = $conn->getLag();
156 if ( $lagTimes[$i] === false ) {
157 $host = $this->parent
->getServerName( $i );
158 $this->replLogger
->error( __METHOD__
. ": host $host is not replicating?" );
162 # Close the connection to avoid sleeper connections piling up.
163 # Note that the caller will pick one of these DBs and reconnect,
164 # which is slightly inefficient, but this only matters for the lag
165 # time cache miss cache, which is far less common that cache hits.
166 $this->parent
->closeConnection( $conn );
170 # Add a timestamp key so we know when it was cached
172 'lagTimes' => $lagTimes,
173 'weightScales' => $weightScales,
174 'timestamp' => microtime( true )
176 $this->mainCache
->set( $key, $value, $staleTTL );
177 $this->srvCache
->set( $key, $value, $staleTTL );
178 $this->replLogger
->info( __METHOD__
. ": re-calculated lag times ($key)" );
184 * @param integer $index Server index
185 * @param IDatabase|null $conn Connection handle or null on connection failure
188 protected function getWeightScale( $index, IDatabase
$conn = null ) {
189 return $conn ?
1.0 : 0.0;
192 public function clearCaches() {
193 $key = $this->getCacheKey();
194 $this->srvCache
->delete( $key );
195 $this->mainCache
->delete( $key );
198 private function getCacheKey() {
199 // Lag is per-server, not per-DB, so key on the master DB name
200 return $this->srvCache
->makeGlobalKey(
202 $this->parent
->getServerName( $this->parent
->getWriterIndex() )