MediaWiki master
EtcdConfig.php
Go to the documentation of this file.
1<?php
21namespace MediaWiki\Config;
22
23use BagOStuff;
26use Psr\Log\LoggerAwareInterface;
27use Psr\Log\LoggerInterface;
29use Wikimedia\IPUtils;
30use Wikimedia\ObjectFactory\ObjectFactory;
31use Wikimedia\WaitConditionLoop;
32
38class EtcdConfig implements Config, LoggerAwareInterface {
40 private $http;
42 private $srvCache;
44 private $procCache;
46 private $dsd;
47
49 private $service;
51 private $host;
53 private $port;
55 private $protocol;
57 private $directory;
59 private $baseCacheTTL;
61 private $skewCacheTTL;
63 private $timeout;
64
78 public function __construct( array $params ) {
79 $params += [
80 'service' => 'etcd',
81 'port' => null,
82 'protocol' => 'http',
83 'cacheTTL' => 10,
84 'skewTTL' => 1,
85 'timeout' => 2
86 ];
87
88 $this->service = $params['service'];
89 $this->host = $params['host'];
90 $this->port = $params['port'];
91 $this->protocol = $params['protocol'];
92 $this->directory = trim( $params['directory'], '/' );
93 $this->skewCacheTTL = $params['skewTTL'];
94 $this->baseCacheTTL = max( $params['cacheTTL'] - $this->skewCacheTTL, 0 );
95 $this->timeout = $params['timeout'];
96
97 // For backwards compatibility, check the host for an embedded port
98 $hostAndPort = IPUtils::splitHostAndPort( $this->host );
99
100 if ( $hostAndPort ) {
101 $this->host = $hostAndPort[0];
102
103 if ( $hostAndPort[1] ) {
104 $this->port = $hostAndPort[1];
105 }
106 }
107
108 // Also for backwards compatibility, check for a host in the format of
109 // an SRV record and use the service specified therein
110 if ( preg_match( '/^_([^\.]+)\._tcp\.(.+)$/', $this->host, $m ) ) {
111 $this->service = $m[1];
112 $this->host = $m[2];
113 }
114
115 if ( !isset( $params['cache'] ) ) {
116 $this->srvCache = new HashBagOStuff();
117 } elseif ( $params['cache'] instanceof BagOStuff ) {
118 $this->srvCache = $params['cache'];
119 } else {
120 $this->srvCache = ObjectFactory::getObjectFromSpec( $params['cache'] );
121 }
122
123 $this->http = new MultiHttpClient( [
124 'connTimeout' => $this->timeout,
125 'reqTimeout' => $this->timeout,
126 ] );
127 $this->dsd = new DnsSrvDiscoverer( $this->service, 'tcp', $this->host );
128 }
129
133 public function setLogger( LoggerInterface $logger ) {
134 trigger_error( __METHOD__ . ' is deprecated since 1.41', E_USER_DEPRECATED );
135 }
136
137 public function has( $name ) {
138 $this->load();
139
140 return array_key_exists( $name, $this->procCache['config'] );
141 }
142
143 public function get( $name ) {
144 $this->load();
145
146 if ( !array_key_exists( $name, $this->procCache['config'] ) ) {
147 throw new ConfigException( "No entry found for '$name'." );
148 }
149
150 return $this->procCache['config'][$name];
151 }
152
153 public function getModifiedIndex() {
154 $this->load();
155 return $this->procCache['modifiedIndex'];
156 }
157
161 private function load() {
162 if ( $this->procCache !== null ) {
163 return; // already loaded
164 }
165
166 $now = microtime( true );
167 $key = $this->srvCache->makeGlobalKey(
168 __CLASS__,
169 $this->host,
170 $this->directory
171 );
172
173 // Get the cached value or block until it is regenerated (by this or another thread)...
174 $data = null; // latest config info
175 $error = null; // last error message
176 $loop = new WaitConditionLoop(
177 function () use ( $key, $now, &$data, &$error ) {
178 // Check if the values are in cache yet...
179 $data = $this->srvCache->get( $key );
180 if ( is_array( $data ) && $data['expires'] > $now ) {
181 return WaitConditionLoop::CONDITION_REACHED;
182 }
183
184 // Cache is either empty or stale;
185 // refresh the cache from etcd, using a mutex to reduce stampedes...
186 if ( $this->srvCache->lock( $key, 0, $this->baseCacheTTL ) ) {
187 try {
188 $etcdResponse = $this->fetchAllFromEtcd();
189 $error = $etcdResponse['error'];
190 if ( is_array( $etcdResponse['config'] ) ) {
191 // Avoid having all servers expire cache keys at the same time
192 $expiry = microtime( true ) + $this->baseCacheTTL;
193 // @phan-suppress-next-line PhanTypeMismatchArgumentInternal
194 $expiry += mt_rand( 0, 1e6 ) / 1e6 * $this->skewCacheTTL;
195 $data = [
196 'config' => $etcdResponse['config'],
197 'expires' => $expiry,
198 'modifiedIndex' => $etcdResponse['modifiedIndex']
199 ];
200 $this->srvCache->set( $key, $data, BagOStuff::TTL_INDEFINITE );
201
202 return WaitConditionLoop::CONDITION_REACHED;
203 } else {
204 trigger_error( "EtcdConfig failed to fetch data: $error", E_USER_WARNING );
205 if ( !$etcdResponse['retry'] && !is_array( $data ) ) {
206 // Fail fast since the error is likely to keep happening
207 return WaitConditionLoop::CONDITION_FAILED;
208 }
209 }
210 } finally {
211 $this->srvCache->unlock( $key ); // release mutex
212 }
213 } else {
214 $error = 'lost lock';
215 }
216
217 if ( is_array( $data ) ) {
218 trigger_error( "EtcdConfig using stale data: $error", E_USER_NOTICE );
219
220 return WaitConditionLoop::CONDITION_REACHED;
221 }
222
223 return WaitConditionLoop::CONDITION_CONTINUE;
224 },
225 $this->timeout
226 );
227
228 if ( $loop->invoke() !== WaitConditionLoop::CONDITION_REACHED ) {
229 // No cached value exists and etcd query failed; throw an error
230 // @phan-suppress-next-line PhanTypeSuspiciousStringExpression WaitConditionLoop throws or error set
231 throw new ConfigException( "Failed to load configuration from etcd: $error" );
232 }
233
234 // @phan-suppress-next-line PhanTypeMismatchProperty WaitConditionLoop throws ore data set
235 $this->procCache = $data;
236 }
237
241 public function fetchAllFromEtcd() {
242 $servers = $this->dsd->getServers() ?: [ [ $this->host, $this->port ] ];
243
244 foreach ( $servers as [ $host, $port ] ) {
245 // Try to load the config from this particular server
246 $response = $this->fetchAllFromEtcdServer( $host, $port );
247 if ( is_array( $response['config'] ) || $response['retry'] ) {
248 break;
249 }
250 }
251
252 return $response;
253 }
254
260 protected function fetchAllFromEtcdServer( string $address, ?int $port = null ) {
261 $host = $address;
262
263 if ( $port !== null ) {
264 $host = IPUtils::combineHostAndPort( $address, $port );
265 }
266
267 // Retrieve all the values under the MediaWiki config directory
268 [ $rcode, $rdesc, /* $rhdrs */, $rbody, $rerr ] = $this->http->run( [
269 'method' => 'GET',
270 'url' => "{$this->protocol}://{$host}/v2/keys/{$this->directory}/?recursive=true",
271 'headers' => [
272 'content-type' => 'application/json',
273 ]
274 ] );
275
276 $response = [ 'config' => null, 'error' => null, 'retry' => false, 'modifiedIndex' => 0 ];
277
278 static $terminalCodes = [ 404 => true ];
279 if ( $rcode < 200 || $rcode > 399 ) {
280 $response['error'] = strlen( $rerr ?? '' ) ? $rerr : "HTTP $rcode ($rdesc)";
281 $response['retry'] = empty( $terminalCodes[$rcode] );
282 return $response;
283 }
284
285 try {
286 $parsedResponse = $this->parseResponse( $rbody );
287 } catch ( EtcdConfigParseError $e ) {
288 $parsedResponse = [ 'error' => $e->getMessage() ];
289 }
290 return array_merge( $response, $parsedResponse );
291 }
292
299 protected function parseResponse( $rbody ) {
300 $info = json_decode( $rbody, true );
301 if ( $info === null ) {
302 throw new EtcdConfigParseError( "Error unserializing JSON response." );
303 }
304 if ( !isset( $info['node'] ) || !is_array( $info['node'] ) ) {
305 throw new EtcdConfigParseError(
306 "Unexpected JSON response: Missing or invalid node at top level." );
307 }
308 $config = [];
309 $lastModifiedIndex = $this->parseDirectory( '', $info['node'], $config );
310 return [ 'modifiedIndex' => $lastModifiedIndex, 'config' => $config ];
311 }
312
322 protected function parseDirectory( $dirName, $dirNode, &$config ) {
323 $lastModifiedIndex = 0;
324 if ( !isset( $dirNode['nodes'] ) ) {
325 throw new EtcdConfigParseError(
326 "Unexpected JSON response in dir '$dirName'; missing 'nodes' list." );
327 }
328 if ( !is_array( $dirNode['nodes'] ) ) {
329 throw new EtcdConfigParseError(
330 "Unexpected JSON response in dir '$dirName'; 'nodes' is not an array." );
331 }
332
333 foreach ( $dirNode['nodes'] as $node ) {
334 '@phan-var array $node';
335 $baseName = basename( $node['key'] );
336 $fullName = $dirName === '' ? $baseName : "$dirName/$baseName";
337 if ( !empty( $node['dir'] ) ) {
338 $lastModifiedIndex = max(
339 $this->parseDirectory( $fullName, $node, $config ),
340 $lastModifiedIndex );
341 } else {
342 $value = $this->unserialize( $node['value'] );
343 if ( !is_array( $value ) || !array_key_exists( 'val', $value ) ) {
344 throw new EtcdConfigParseError( "Failed to parse value for '$fullName'." );
345 }
346 $lastModifiedIndex = max( $node['modifiedIndex'], $lastModifiedIndex );
347 $config[$fullName] = $value['val'];
348 }
349 }
350 return $lastModifiedIndex;
351 }
352
357 private function unserialize( $string ) {
358 return json_decode( $string, true );
359 }
360}
361
363class_alias( EtcdConfig::class, 'EtcdConfig' );
array $params
The job parameters.
Class representing a cache/ephemeral data store.
Definition BagOStuff.php:85
Simple store for keeping values in an associative array for the current process.
Exceptions for config failures.
Interface for configuration instances.
parseResponse( $rbody)
Parse a response body, throwing EtcdConfigParseError if there is a validation error.
parseDirectory( $dirName, $dirNode, &$config)
Recursively parse a directory node and populate the array passed by reference, throwing EtcdConfigPar...
fetchAllFromEtcdServer(string $address, ?int $port=null)
has( $name)
Check whether a configuration option is set for the given name.
setLogger(LoggerInterface $logger)
Class to handle multiple HTTP requests.
Interface for configuration instances.
Definition Config.php:32