Server : Apache System : Linux server2.corals.io 4.18.0-348.2.1.el8_5.x86_64 #1 SMP Mon Nov 15 09:17:08 EST 2021 x86_64 User : corals ( 1002) PHP Version : 7.4.33 Disable Function : exec,passthru,shell_exec,system Directory : /home/corals/old/vendor/elasticsearch/elasticsearch/src/Elasticsearch/Connections/ |
<?php
/**
* Elasticsearch PHP client
*
* @link https://github.com/elastic/elasticsearch-php/
* @copyright Copyright (c) Elasticsearch B.V (https://www.elastic.co)
* @license http://www.apache.org/licenses/LICENSE-2.0 Apache License, Version 2.0
* @license https://www.gnu.org/licenses/lgpl-2.1.html GNU Lesser General Public License, Version 2.1
*
* Licensed to Elasticsearch B.V under one or more agreements.
* Elasticsearch B.V licenses this file to you under the Apache 2.0 License or
* the GNU Lesser General Public License, Version 2.1, at your option.
* See the LICENSE file in the project root for more information.
*/
declare(strict_types = 1);
namespace Elasticsearch\Connections;
use Elasticsearch\Client;
use Elasticsearch\Common\Exceptions\BadRequest400Exception;
use Elasticsearch\Common\Exceptions\Conflict409Exception;
use Elasticsearch\Common\Exceptions\Curl\CouldNotConnectToHost;
use Elasticsearch\Common\Exceptions\Curl\CouldNotResolveHostException;
use Elasticsearch\Common\Exceptions\Curl\OperationTimeoutException;
use Elasticsearch\Common\Exceptions\ElasticsearchException;
use Elasticsearch\Common\Exceptions\Forbidden403Exception;
use Elasticsearch\Common\Exceptions\MaxRetriesException;
use Elasticsearch\Common\Exceptions\Missing404Exception;
use Elasticsearch\Common\Exceptions\NoDocumentsToGetException;
use Elasticsearch\Common\Exceptions\NoShardAvailableException;
use Elasticsearch\Common\Exceptions\RequestTimeout408Exception;
use Elasticsearch\Common\Exceptions\RoutingMissingException;
use Elasticsearch\Common\Exceptions\ScriptLangNotSupportedException;
use Elasticsearch\Common\Exceptions\ServerErrorResponseException;
use Elasticsearch\Common\Exceptions\TransportException;
use Elasticsearch\Common\Exceptions\Unauthorized401Exception;
use Elasticsearch\Serializers\SerializerInterface;
use Elasticsearch\Transport;
use Exception;
use GuzzleHttp\Ring\Core;
use GuzzleHttp\Ring\Exception\ConnectException;
use GuzzleHttp\Ring\Exception\RingException;
use Psr\Log\LoggerInterface;
class Connection implements ConnectionInterface
{
/**
* @var callable
*/
protected $handler;
/**
* @var SerializerInterface
*/
protected $serializer;
/**
* @var string
*/
protected $transportSchema = 'http'; // TODO depreciate this default
/**
* @var string
*/
protected $host;
/**
* @var string|null
*/
protected $path;
/**
* @var int
*/
protected $port;
/**
* @var LoggerInterface
*/
protected $log;
/**
* @var LoggerInterface
*/
protected $trace;
/**
* @var array
*/
protected $connectionParams;
/**
* @var array
*/
protected $headers = [];
/**
* @var bool
*/
protected $isAlive = true;
/**
* @var float
*/
private $pingTimeout = 1; //TODO expose this
/**
* @var int
*/
private $lastPing = 0;
/**
* @var int
*/
private $failedPings = 0;
/**
* @var mixed[]
*/
private $lastRequest = array();
/**
* @var string
*/
private $OSVersion = null;
public function __construct(
callable $handler,
array $hostDetails,
array $connectionParams,
SerializerInterface $serializer,
LoggerInterface $log,
LoggerInterface $trace
) {
if (isset($hostDetails['port']) !== true) {
$hostDetails['port'] = 9200;
}
if (isset($hostDetails['scheme'])) {
$this->transportSchema = $hostDetails['scheme'];
}
// Only Set the Basic if API Key is not set and setBasicAuthentication was not called prior
if (isset($connectionParams['client']['headers']['Authorization']) === false
&& isset($connectionParams['client']['curl'][CURLOPT_HTTPAUTH]) === false
&& isset($hostDetails['user'])
&& isset($hostDetails['pass'])
) {
$connectionParams['client']['curl'][CURLOPT_HTTPAUTH] = CURLAUTH_BASIC;
$connectionParams['client']['curl'][CURLOPT_USERPWD] = $hostDetails['user'].':'.$hostDetails['pass'];
}
$connectionParams['client']['curl'][CURLOPT_PORT] = $hostDetails['port'];
if (isset($connectionParams['client']['headers'])) {
$this->headers = $connectionParams['client']['headers'];
unset($connectionParams['client']['headers']);
}
// Add the User-Agent using the format: <client-repo-name>/<client-version> (metadata-values)
$this->headers['User-Agent'] = [sprintf(
"elasticsearch-php/%s (%s %s; PHP %s)",
Client::VERSION,
PHP_OS,
$this->getOSVersion(),
phpversion()
)];
// Add x-elastic-client-meta header, if enabled
if (isset($connectionParams['client']['x-elastic-client-meta']) && $connectionParams['client']['x-elastic-client-meta']) {
$this->headers['x-elastic-client-meta'] = [$this->getElasticMetaHeader($connectionParams)];
}
$host = $hostDetails['host'];
$path = null;
if (isset($hostDetails['path']) === true) {
$path = $hostDetails['path'];
}
$port = $hostDetails['port'];
$this->host = $host;
$this->path = $path;
$this->port = $port;
$this->log = $log;
$this->trace = $trace;
$this->connectionParams = $connectionParams;
$this->serializer = $serializer;
$this->handler = $this->wrapHandler($handler);
}
/**
* @param string $method
* @param string $uri
* @param null|array $params
* @param null $body
* @param array $options
* @param Transport $transport
* @return mixed
*/
public function performRequest(string $method, string $uri, ?array $params = [], $body = null, array $options = [], Transport $transport = null)
{
if ($body !== null) {
$body = $this->serializer->serialize($body);
}
$headers = $this->headers;
if (isset($options['client']['headers']) && is_array($options['client']['headers'])) {
$headers = array_merge($this->headers, $options['client']['headers']);
}
$host = $this->host;
if (isset($this->connectionParams['client']['port_in_header']) && $this->connectionParams['client']['port_in_header']) {
$host .= ':' . $this->port;
}
$request = [
'http_method' => $method,
'scheme' => $this->transportSchema,
'uri' => $this->getURI($uri, $params),
'body' => $body,
'headers' => array_merge(
[
'Host' => [$host]
],
$headers
)
];
$request = array_replace_recursive($request, $this->connectionParams, $options);
// RingPHP does not like if client is empty
if (empty($request['client'])) {
unset($request['client']);
}
$handler = $this->handler;
$future = $handler($request, $this, $transport, $options);
return $future;
}
public function getTransportSchema(): string
{
return $this->transportSchema;
}
public function getLastRequestInfo(): array
{
return $this->lastRequest;
}
private function wrapHandler(callable $handler): callable
{
return function (array $request, Connection $connection, Transport $transport = null, $options) use ($handler) {
$this->lastRequest = [];
$this->lastRequest['request'] = $request;
// Send the request using the wrapped handler.
$response = Core::proxy(
$handler($request),
function ($response) use ($connection, $transport, $request, $options) {
$this->lastRequest['response'] = $response;
if (isset($response['error']) === true) {
if ($response['error'] instanceof ConnectException || $response['error'] instanceof RingException) {
$this->log->warning("Curl exception encountered.");
$exception = $this->getCurlRetryException($request, $response);
$this->logRequestFail($request, $response, $exception);
$node = $connection->getHost();
$this->log->warning("Marking node $node dead.");
$connection->markDead();
// If the transport has not been set, we are inside a Ping or Sniff,
// so we don't want to retrigger retries anyway.
//
// TODO this could be handled better, but we are limited because connectionpools do not
// have access to Transport. Architecturally, all of this needs to be refactored
if (isset($transport) === true) {
$transport->connectionPool->scheduleCheck();
$neverRetry = isset($request['client']['never_retry']) ? $request['client']['never_retry'] : false;
$shouldRetry = $transport->shouldRetry($request);
$shouldRetryText = ($shouldRetry) ? 'true' : 'false';
$this->log->warning("Retries left? $shouldRetryText");
if ($shouldRetry && !$neverRetry) {
return $transport->performRequest(
$request['http_method'],
$request['uri'],
[],
$request['body'],
$options
);
}
}
$this->log->warning("Out of retries, throwing exception from $node");
// Only throw if we run out of retries
throw $exception;
} else {
// Something went seriously wrong, bail
$exception = new TransportException($response['error']->getMessage());
$this->logRequestFail($request, $response, $exception);
throw $exception;
}
} else {
$connection->markAlive();
if (isset($response['headers']['Warning'])) {
$this->logWarning($request, $response);
}
if (isset($response['body']) === true) {
$response['body'] = stream_get_contents($response['body']);
$this->lastRequest['response']['body'] = $response['body'];
}
if ($response['status'] >= 400 && $response['status'] < 500) {
$ignore = $request['client']['ignore'] ?? [];
// Skip 404 if succeeded true in the body (e.g. clear_scroll)
$body = $response['body'] ?? '';
if (strpos($body, '"succeeded":true') !== false) {
$ignore[] = 404;
}
$this->process4xxError($request, $response, $ignore);
} elseif ($response['status'] >= 500) {
$ignore = $request['client']['ignore'] ?? [];
$this->process5xxError($request, $response, $ignore);
}
// No error, deserialize
$response['body'] = $this->serializer->deserialize($response['body'], $response['transfer_stats']);
}
$this->logRequestSuccess($request, $response);
return isset($request['client']['verbose']) && $request['client']['verbose'] === true ? $response : $response['body'];
}
);
return $response;
};
}
private function getURI(string $uri, ?array $params): string
{
if (isset($params) === true && !empty($params)) {
$params = array_map(
function ($value) {
if ($value === true) {
return 'true';
} elseif ($value === false) {
return 'false';
}
return $value;
},
$params
);
$uri .= '?' . http_build_query($params);
}
if ($this->path !== null) {
$uri = $this->path . $uri;
}
return $uri;
}
public function getHeaders(): array
{
return $this->headers;
}
public function logWarning(array $request, array $response): void
{
$this->log->warning('Deprecation', $response['headers']['Warning']);
}
/**
* Log a successful request
*
* @param array $request
* @param array $response
* @return void
*/
public function logRequestSuccess(array $request, array $response): void
{
$port = $request['client']['curl'][CURLOPT_PORT] ?? $response['transfer_stats']['primary_port'] ?? '';
$uri = $this->addPortInUrl($response['effective_url'], (int) $port);
$this->log->debug('Request Body', array($request['body']));
$this->log->info(
'Request Success:',
array(
'method' => $request['http_method'],
'uri' => $uri,
'port' => $port,
'headers' => $request['headers'],
'HTTP code' => $response['status'],
'duration' => $response['transfer_stats']['total_time'],
)
);
$this->log->debug('Response', array($response['body']));
// Build the curl command for Trace.
$curlCommand = $this->buildCurlCommand($request['http_method'], $uri, $request['body']);
$this->trace->info($curlCommand);
$this->trace->debug(
'Response:',
array(
'response' => $response['body'],
'method' => $request['http_method'],
'uri' => $uri,
'port' => $port,
'HTTP code' => $response['status'],
'duration' => $response['transfer_stats']['total_time'],
)
);
}
/**
* Log a failed request
*
* @param array $request
* @param array $response
* @param \Exception $exception
*
* @return void
*/
public function logRequestFail(array $request, array $response, \Exception $exception): void
{
$port = $request['client']['curl'][CURLOPT_PORT] ?? $response['transfer_stats']['primary_port'] ?? '';
$uri = $this->addPortInUrl($response['effective_url'], (int) $port);
$this->log->debug('Request Body', array($request['body']));
$this->log->warning(
'Request Failure:',
array(
'method' => $request['http_method'],
'uri' => $uri,
'port' => $port,
'headers' => $request['headers'],
'HTTP code' => $response['status'],
'duration' => $response['transfer_stats']['total_time'],
'error' => $exception->getMessage(),
)
);
$this->log->warning('Response', array($response['body']));
// Build the curl command for Trace.
$curlCommand = $this->buildCurlCommand($request['http_method'], $uri, $request['body']);
$this->trace->info($curlCommand);
$this->trace->debug(
'Response:',
array(
'response' => $response,
'method' => $request['http_method'],
'uri' => $uri,
'port' => $port,
'HTTP code' => $response['status'],
'duration' => $response['transfer_stats']['total_time'],
)
);
}
public function ping(): bool
{
$options = [
'client' => [
'timeout' => $this->pingTimeout,
'never_retry' => true,
'verbose' => true
]
];
try {
$response = $this->performRequest('HEAD', '/', null, null, $options);
$response = $response->wait();
} catch (TransportException $exception) {
$this->markDead();
return false;
}
if ($response['status'] === 200) {
$this->markAlive();
return true;
} else {
$this->markDead();
return false;
}
}
/**
* @return array|\GuzzleHttp\Ring\Future\FutureArray
*/
public function sniff()
{
$options = [
'client' => [
'timeout' => $this->pingTimeout,
'never_retry' => true
]
];
return $this->performRequest('GET', '/_nodes/', null, null, $options);
}
public function isAlive(): bool
{
return $this->isAlive;
}
public function markAlive(): void
{
$this->failedPings = 0;
$this->isAlive = true;
$this->lastPing = time();
}
public function markDead(): void
{
$this->isAlive = false;
$this->failedPings += 1;
$this->lastPing = time();
}
public function getLastPing(): int
{
return $this->lastPing;
}
public function getPingFailures(): int
{
return $this->failedPings;
}
public function getHost(): string
{
return $this->host;
}
public function getUserPass(): ?string
{
return $this->connectionParams['client']['curl'][CURLOPT_USERPWD] ?? null;
}
public function getPath(): ?string
{
return $this->path;
}
/**
* @return int
*/
public function getPort()
{
return $this->port;
}
protected function getCurlRetryException(array $request, array $response): ElasticsearchException
{
$exception = null;
$message = $response['error']->getMessage();
$exception = new MaxRetriesException($message);
switch ($response['curl']['errno']) {
case 6:
$exception = new CouldNotResolveHostException($message, 0, $exception);
break;
case 7:
$exception = new CouldNotConnectToHost($message, 0, $exception);
break;
case 28:
$exception = new OperationTimeoutException($message, 0, $exception);
break;
}
return $exception;
}
/**
* Get the x-elastic-client-meta header
*
* The header format is specified by the following regex:
* ^[a-z]{1,}=[a-z0-9\.\-]{1,}(?:,[a-z]{1,}=[a-z0-9\.\-]+)*$
*/
private function getElasticMetaHeader(array $connectionParams): string
{
$phpSemVersion = sprintf("%d.%d.%d", PHP_MAJOR_VERSION, PHP_MINOR_VERSION, PHP_RELEASE_VERSION);
// Reduce the size in case of '-snapshot' version
$clientVersion = str_replace('-snapshot', '-s', strtolower(Client::VERSION));
$clientMeta = sprintf(
"es=%s,php=%s,t=%s,a=%d",
$clientVersion,
$phpSemVersion,
$clientVersion,
isset($connectionParams['client']['future']) && $connectionParams['client']['future'] === 'lazy' ? 1 : 0
);
if (function_exists('curl_version')) {
$curlVersion = curl_version();
if (isset($curlVersion['version'])) {
$clientMeta .= sprintf(",cu=%s", $curlVersion['version']); // cu = curl library
}
}
return $clientMeta;
}
/**
* Get the OS version using php_uname if available
* otherwise it returns an empty string
*
* @see https://github.com/elastic/elasticsearch-php/issues/922
*/
private function getOSVersion(): string
{
if ($this->OSVersion === null) {
$this->OSVersion = strpos(strtolower(ini_get('disable_functions')), 'php_uname') !== false
? ''
: php_uname("r");
}
return $this->OSVersion;
}
/**
* Add the port value in the URL if not present
*/
private function addPortInUrl(string $uri, int $port): string
{
if (strpos($uri, ':', 7) !== false) {
return $uri;
}
return preg_replace('#([^/])/([^/])#', sprintf("$1:%s/$2", $port), $uri, 1);
}
/**
* Construct a string cURL command
*/
private function buildCurlCommand(string $method, string $url, ?string $body): string
{
if (strpos($url, '?') === false) {
$url .= '?pretty=true';
} else {
str_replace('?', '?pretty=true', $url);
}
$curlCommand = 'curl -X' . strtoupper($method);
$curlCommand .= " '" . $url . "'";
if (isset($body) === true && $body !== '') {
$curlCommand .= " -d '" . $body . "'";
}
return $curlCommand;
}
private function process4xxError(array $request, array $response, array $ignore): ?ElasticsearchException
{
$statusCode = $response['status'];
/**
* @var \Exception $exception
*/
$exception = $this->tryDeserialize400Error($response);
if (array_search($response['status'], $ignore) !== false) {
return null;
}
$responseBody = $this->convertBodyToString($response['body'], $statusCode, $exception);
if ($statusCode === 401) {
$exception = new Unauthorized401Exception($responseBody, $statusCode);
} elseif ($statusCode === 403) {
$exception = new Forbidden403Exception($responseBody, $statusCode);
} elseif ($statusCode === 404) {
$exception = new Missing404Exception($responseBody, $statusCode);
} elseif ($statusCode === 409) {
$exception = new Conflict409Exception($responseBody, $statusCode);
} elseif ($statusCode === 400 && strpos($responseBody, 'script_lang not supported') !== false) {
$exception = new ScriptLangNotSupportedException($responseBody. $statusCode);
} elseif ($statusCode === 408) {
$exception = new RequestTimeout408Exception($responseBody, $statusCode);
} else {
$exception = new BadRequest400Exception($responseBody, $statusCode);
}
$this->logRequestFail($request, $response, $exception);
throw $exception;
}
private function process5xxError(array $request, array $response, array $ignore): ?ElasticsearchException
{
$statusCode = (int) $response['status'];
$responseBody = $response['body'];
/**
* @var \Exception $exception
*/
$exception = $this->tryDeserialize500Error($response);
$exceptionText = "[$statusCode Server Exception] ".$exception->getMessage();
$this->log->error($exceptionText);
$this->log->error($exception->getTraceAsString());
if (array_search($statusCode, $ignore) !== false) {
return null;
}
if ($statusCode === 500 && strpos($responseBody, "RoutingMissingException") !== false) {
$exception = new RoutingMissingException($exception->getMessage(), $statusCode, $exception);
} elseif ($statusCode === 500 && preg_match('/ActionRequestValidationException.+ no documents to get/', $responseBody) === 1) {
$exception = new NoDocumentsToGetException($exception->getMessage(), $statusCode, $exception);
} elseif ($statusCode === 500 && strpos($responseBody, 'NoShardAvailableActionException') !== false) {
$exception = new NoShardAvailableException($exception->getMessage(), $statusCode, $exception);
} else {
$exception = new ServerErrorResponseException(
$this->convertBodyToString($responseBody, $statusCode, $exception),
$statusCode
);
}
$this->logRequestFail($request, $response, $exception);
throw $exception;
}
private function convertBodyToString($body, int $statusCode, Exception $exception) : string
{
if (empty($body)) {
return sprintf(
"Unknown %d error from Elasticsearch %s",
$statusCode,
$exception->getMessage()
);
}
// if body is not string, we convert it so it can be used as Exception message
if (!is_string($body)) {
return json_encode($body);
}
return $body;
}
private function tryDeserialize400Error(array $response): ElasticsearchException
{
return $this->tryDeserializeError($response, BadRequest400Exception::class);
}
private function tryDeserialize500Error(array $response): ElasticsearchException
{
return $this->tryDeserializeError($response, ServerErrorResponseException::class);
}
private function tryDeserializeError(array $response, string $errorClass): ElasticsearchException
{
$error = $this->serializer->deserialize($response['body'], $response['transfer_stats']);
if (is_array($error) === true) {
if (isset($error['error']) === false) {
// <2.0 "i just blew up" nonstructured exception
// $error is an array but we don't know the format, reuse the response body instead
// added json_encode to convert into a string
return new $errorClass(json_encode($response['body']), (int) $response['status']);
}
// 2.0 structured exceptions
if (is_array($error['error']) && array_key_exists('reason', $error['error']) === true) {
// Try to use root cause first (only grabs the first root cause)
$root = $error['error']['root_cause'];
if (isset($root) && isset($root[0])) {
$cause = $root[0]['reason'];
$type = $root[0]['type'];
} else {
$cause = $error['error']['reason'];
$type = $error['error']['type'];
}
// added json_encode to convert into a string
$original = new $errorClass(json_encode($response['body']), $response['status']);
return new $errorClass("$type: $cause", (int) $response['status'], $original);
}
// <2.0 semi-structured exceptions
// added json_encode to convert into a string
$original = new $errorClass(json_encode($response['body']), $response['status']);
$errorEncoded = $error['error'];
if (is_array($errorEncoded)) {
$errorEncoded = json_encode($errorEncoded);
}
return new $errorClass($errorEncoded, (int) $response['status'], $original);
}
// if responseBody is not string, we convert it so it can be used as Exception message
$responseBody = $response['body'];
if (!is_string($responseBody)) {
$responseBody = json_encode($responseBody);
}
// <2.0 "i just blew up" nonstructured exception
return new $errorClass($responseBody);
}
}