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
21 use Psr\Log\LoggerAwareInterface
;
22 use Psr\Log\LoggerInterface
;
23 use Wikimedia\ObjectFactory
;
24 use Wikimedia\WaitConditionLoop
;
27 * Interface for configuration instances
31 class EtcdConfig
implements Config
, LoggerAwareInterface
{
32 /** @var MultiHttpClient */
38 /** @var LoggerInterface */
50 private $baseCacheTTL;
52 private $skewCacheTTL;
57 * @param array $params Parameter map:
58 * - host: the host address and port
59 * - protocol: either http or https
60 * - directory: the etc "directory" were MediaWiki specific variables are located
61 * - encoding: one of ("JSON", "YAML"). Defaults to JSON. [optional]
62 * - cache: BagOStuff instance or ObjectFactory spec thereof for a server cache.
63 * The cache will also be used as a fallback if etcd is down. [optional]
64 * - cacheTTL: logical cache TTL in seconds [optional]
65 * - skewTTL: maximum seconds to randomly lower the assigned TTL on cache save [optional]
66 * - timeout: seconds to wait for etcd before throwing an error [optional]
68 public function __construct( array $params ) {
77 $this->host
= $params['host'];
78 $this->protocol
= $params['protocol'];
79 $this->directory
= trim( $params['directory'], '/' );
80 $this->encoding
= $params['encoding'];
81 $this->skewCacheTTL
= $params['skewTTL'];
82 $this->baseCacheTTL
= max( $params['cacheTTL'] - $this->skewCacheTTL
, 0 );
83 $this->timeout
= $params['timeout'];
85 if ( !isset( $params['cache'] ) ) {
86 $this->srvCache
= new HashBagOStuff();
87 } elseif ( $params['cache'] instanceof BagOStuff
) {
88 $this->srvCache
= $params['cache'];
90 $this->srvCache
= ObjectFactory
::getObjectFromSpec( $params['cache'] );
93 $this->logger
= new Psr\Log\
NullLogger();
94 $this->http
= new MultiHttpClient( [
95 'connTimeout' => $this->timeout
,
96 'reqTimeout' => $this->timeout
,
97 'logger' => $this->logger
101 public function setLogger( LoggerInterface
$logger ) {
102 $this->logger
= $logger;
103 $this->http
->setLogger( $logger );
106 public function has( $name ) {
109 return array_key_exists( $name, $this->procCache
['config'] );
112 public function get( $name ) {
115 if ( !array_key_exists( $name, $this->procCache
['config'] ) ) {
116 throw new ConfigException( "No entry found for '$name'." );
119 return $this->procCache
['config'][$name];
122 public function getModifiedIndex() {
124 return $this->procCache
['modifiedIndex'];
128 * @throws ConfigException
130 private function load() {
131 if ( $this->procCache
!== null ) {
132 return; // already loaded
135 $now = microtime( true );
136 $key = $this->srvCache
->makeGlobalKey(
142 // Get the cached value or block until it is regenerated (by this or another thread)...
143 $data = null; // latest config info
144 $error = null; // last error message
145 $loop = new WaitConditionLoop(
146 function () use ( $key, $now, &$data, &$error ) {
147 // Check if the values are in cache yet...
148 $data = $this->srvCache
->get( $key );
149 if ( is_array( $data ) && $data['expires'] > $now ) {
150 $this->logger
->debug( "Found up-to-date etcd configuration cache." );
152 return WaitConditionLoop
::CONDITION_REACHED
;
155 // Cache is either empty or stale;
156 // refresh the cache from etcd, using a mutex to reduce stampedes...
157 if ( $this->srvCache
->lock( $key, 0, $this->baseCacheTTL
) ) {
159 $etcdResponse = $this->fetchAllFromEtcd();
160 $error = $etcdResponse['error'];
161 if ( is_array( $etcdResponse['config'] ) ) {
162 // Avoid having all servers expire cache keys at the same time
163 $expiry = microtime( true ) +
$this->baseCacheTTL
;
164 $expiry +
= mt_rand( 0, 1e6
) / 1e6
* $this->skewCacheTTL
;
166 'config' => $etcdResponse['config'],
167 'expires' => $expiry,
168 'modifiedIndex' => $etcdResponse['modifiedIndex']
170 $this->srvCache
->set( $key, $data, BagOStuff
::TTL_INDEFINITE
);
172 $this->logger
->info( "Refreshed stale etcd configuration cache." );
174 return WaitConditionLoop
::CONDITION_REACHED
;
176 $this->logger
->error( "Failed to fetch configuration: $error" );
177 if ( !$etcdResponse['retry'] ) {
178 // Fail fast since the error is likely to keep happening
179 return WaitConditionLoop
::CONDITION_FAILED
;
183 $this->srvCache
->unlock( $key ); // release mutex
187 if ( is_array( $data ) ) {
188 $this->logger
->info( "Using stale etcd configuration cache." );
190 return WaitConditionLoop
::CONDITION_REACHED
;
193 return WaitConditionLoop
::CONDITION_CONTINUE
;
198 if ( $loop->invoke() !== WaitConditionLoop
::CONDITION_REACHED
) {
199 // No cached value exists and etcd query failed; throw an error
200 throw new ConfigException( "Failed to load configuration from etcd: $error" );
203 $this->procCache
= $data;
207 * @return array (containing the keys config, error, retry, modifiedIndex)
209 public function fetchAllFromEtcd() {
210 // TODO: inject DnsSrvDiscoverer in order to be able to test this method
211 $dsd = new DnsSrvDiscoverer( $this->host
);
212 $servers = $dsd->getServers();
214 return $this->fetchAllFromEtcdServer( $this->host
);
218 // Pick a random etcd server from dns
219 $server = $dsd->pickServer( $servers );
220 $host = IP
::combineHostAndPort( $server['target'], $server['port'] );
221 // Try to load the config from this particular server
222 $response = $this->fetchAllFromEtcdServer( $host );
223 if ( is_array( $response['config'] ) ||
$response['retry'] ) {
227 // Avoid the server next time if that failed
228 $servers = $dsd->removeServer( $server, $servers );
229 } while ( $servers );
235 * @param string $address Host and port
236 * @return array (containing the keys config, error, retry, modifiedIndex)
238 protected function fetchAllFromEtcdServer( $address ) {
239 // Retrieve all the values under the MediaWiki config directory
240 list( $rcode, $rdesc, /* $rhdrs */, $rbody, $rerr ) = $this->http
->run( [
242 'url' => "{$this->protocol}://{$address}/v2/keys/{$this->directory}/?recursive=true",
243 'headers' => [ 'content-type' => 'application/json' ]
246 $response = [ 'config' => null, 'error' => null, 'retry' => false, 'modifiedIndex' => 0 ];
248 static $terminalCodes = [ 404 => true ];
249 if ( $rcode < 200 ||
$rcode > 399 ) {
250 $response['error'] = strlen( $rerr ) ?
$rerr : "HTTP $rcode ($rdesc)";
251 $response['retry'] = empty( $terminalCodes[$rcode] );
256 $parsedResponse = $this->parseResponse( $rbody );
257 } catch ( EtcdConfigParseError
$e ) {
258 $parsedResponse = [ 'error' => $e->getMessage() ];
260 return array_merge( $response, $parsedResponse );
264 * Parse a response body, throwing EtcdConfigParseError if there is a validation error
266 * @param string $rbody
269 protected function parseResponse( $rbody ) {
270 $info = json_decode( $rbody, true );
271 if ( $info === null ) {
272 throw new EtcdConfigParseError( "Error unserializing JSON response." );
274 if ( !isset( $info['node'] ) ||
!is_array( $info['node'] ) ) {
275 throw new EtcdConfigParseError(
276 "Unexpected JSON response: Missing or invalid node at top level." );
279 $lastModifiedIndex = $this->parseDirectory( '', $info['node'], $config );
280 return [ 'modifiedIndex' => $lastModifiedIndex, 'config' => $config ];
284 * Recursively parse a directory node and populate the array passed by
285 * reference, throwing EtcdConfigParseError if there is a validation error
287 * @param string $dirName The relative directory name
288 * @param array $dirNode The decoded directory node
289 * @param array &$config The output array
290 * @return int lastModifiedIndex The maximum last modified index across all keys in the directory
292 protected function parseDirectory( $dirName, $dirNode, &$config ) {
293 $lastModifiedIndex = 0;
294 if ( !isset( $dirNode['nodes'] ) ) {
295 throw new EtcdConfigParseError(
296 "Unexpected JSON response in dir '$dirName'; missing 'nodes' list." );
298 if ( !is_array( $dirNode['nodes'] ) ) {
299 throw new EtcdConfigParseError(
300 "Unexpected JSON response in dir '$dirName'; 'nodes' is not an array." );
303 foreach ( $dirNode['nodes'] as $node ) {
304 $baseName = basename( $node['key'] );
305 $fullName = $dirName === '' ?
$baseName : "$dirName/$baseName";
306 if ( !empty( $node['dir'] ) ) {
307 $lastModifiedIndex = max(
308 $this->parseDirectory( $fullName, $node, $config ),
309 $lastModifiedIndex );
311 $value = $this->unserialize( $node['value'] );
312 if ( !is_array( $value ) ||
!array_key_exists( 'val', $value ) ) {
313 throw new EtcdConfigParseError( "Failed to parse value for '$fullName'." );
315 $lastModifiedIndex = max( $node['modifiedIndex'], $lastModifiedIndex );
316 $config[$fullName] = $value['val'];
319 return $lastModifiedIndex;
323 * @param string $string
326 private function unserialize( $string ) {
327 if ( $this->encoding
=== 'YAML' ) {
328 return yaml_parse( $string );
330 return json_decode( $string, true );