File Coverage

blib/lib/Search/Elasticsearch/Role/Cxn.pm

Criterion	Covered	Total	%
statement	158	210	75.2
branch	51	94	54.2
condition	35	61	57.3
subroutine	31	33	93.9
pod	9	13	69.2
total	284	411	69.1

# Licensed to Elasticsearch B.V. under one or more contributor
# license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright
# ownership. Elasticsearch B.V. licenses this file to you under
# the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.

package Search::Elasticsearch::Role::Cxn;
$Search::Elasticsearch::Role::Cxn::VERSION = '8.12';
our $PRODUCT_CHECK_HEADER = 'x-elastic-product';
our $PRODUCT_CHECK_VALUE = 'Elasticsearch';

use Moo::Role;
use Search::Elasticsearch::Util qw(parse_params throw to_list);
use List::Util qw(min);
use Try::Tiny;
use URI();
use IO::Compress::Deflate();
use IO::Uncompress::Inflate();
use IO::Compress::Gzip();
use IO::Uncompress::Gunzip qw(gunzip $GunzipError);
use Search::Elasticsearch::Util qw(to_list);
use namespace::clean;
use Net::IP;

requires qw(perform_request error_from_text handle);

has 'client_version'        => ( is => 'ro', required => 1, default => $Search::Elasticsearch::VERSION );
has 'host'                  => ( is => 'ro', required => 1 );
has 'port'                  => ( is => 'ro', required => 1 );
has 'uri'                   => ( is => 'ro', required => 1 );
has 'request_timeout'       => ( is => 'ro', default  => 30 );
has 'ping_timeout'          => ( is => 'ro', default  => 2 );
has 'sniff_timeout'         => ( is => 'ro', default  => 1 );
has 'sniff_request_timeout' => ( is => 'ro', default  => 2 );
has 'next_ping'             => ( is => 'rw', default  => 0 );
has 'ping_failures'         => ( is => 'rw', default  => 0 );
has 'dead_timeout'          => ( is => 'ro', default  => 60 );
has 'max_dead_timeout'      => ( is => 'ro', default  => 3600 );
has 'serializer'            => ( is => 'ro', required => 1 );
has 'logger'                => ( is => 'ro', required => 1 );
has 'handle_args'           => ( is => 'ro', default  => sub { {} } );
has 'default_qs_params'     => ( is => 'ro', default  => sub { {} } );
has 'scheme'             => ( is => 'ro' );
has 'is_https'           => ( is => 'ro' );
has 'userinfo'           => ( is => 'ro' );
has 'max_content_length' => ( is => 'ro' );
has 'default_headers'    => ( is => 'ro' );
has 'deflate'            => ( is => 'ro' );
has 'gzip'               => ( is => 'ro' );
has 'ssl_options'        => ( is => 'ro', predicate => 'has_ssl_options' );
has 'handle'             => ( is => 'lazy', clearer => 1 );
has '_pid'               => ( is => 'rw', default => $$ );

my %Code_To_Error = (
    400 => 'Request',
    401 => 'Unauthorized',
    403 => 'Forbidden',
    404 => 'Missing',
    408 => 'RequestTimeout',
    409 => 'Conflict',
    413 => 'ContentLength',
    502 => 'BadGateway',
    503 => 'Unavailable',
    504 => 'GatewayTimeout'
);

#===================================
sub stringify { shift->uri . '' }
#===================================

#===================================
sub get_user_agent {
#===================================
    return sprintf("elasticsearch-perl/%s (%s; perl %s)", $Search::Elasticsearch::VERSION, $^O, $]);
}

#===================================
sub get_meta_header {
#===================================
    return sprintf("es=%s,pl=%s", $Search::Elasticsearch::VERSION, $^V);
}


#===================================
sub BUILDARGS {
#===================================
    my ( $class, $params ) = parse_params(@_);

    my $node = $params->{node}
        || { host => 'localhost', port => '9200' };

    unless ( ref $node eq 'HASH' ) {
        $node = "[$node]" if Net::IP::ip_is_ipv6($node);
        unless ( $node =~ m{^http(s)?://} ) {
            $node = ( $params->{use_https} ? 'https://' : 'http://' ) . $node;
        }
        if ( $params->{port} && $node !~ m{//[^/\[]+:\d+} ) {
            $node =~ s{(//[^/]+)}{$1:$params->{port}};
        }
        my $uri = URI->new($node);
        $node = {
            scheme   => $uri->scheme,
            host     => $uri->host,
            port     => $uri->port,
            path     => $uri->path,
            userinfo => $uri->userinfo
        };
    }

    my $host = $node->{host} || 'localhost';
    my $userinfo = $node->{userinfo} || $params->{userinfo} || '';
    my $scheme
        = $node->{scheme} || ( $params->{use_https} ? 'https' : 'http' );
    my $port
        = $node->{port}
        || $params->{port}
        || ( $scheme eq 'http' ? 80 : 443 );
    my $path = $node->{path} || $params->{path_prefix} || '';
    $path =~ s{^/?}{/}g;
    $path =~ s{/+$}{};

    my %default_headers = %{ $params->{default_headers} || {} };

    if ($userinfo) {
        require MIME::Base64;
        my $auth = MIME::Base64::encode_base64( $userinfo, "" );
        chomp $auth;
        $default_headers{Authorization} = "Basic $auth";
    }

    if ( $params->{gzip} ) {
        $default_headers{'Accept-Encoding'} = "gzip";
    }

    elsif ( $params->{deflate} ) {
        $default_headers{'Accept-Encoding'} = "deflate";
    }

    $default_headers{'User-Agent'} = $class->get_user_agent();

    # Add Elastic meta header
    $default_headers{'x-elastic-client-meta'} = $class->get_meta_header();

    # Compatibility header
    if (defined $ENV{ELASTIC_CLIENT_APIVERSIONING} &&
        (lc($ENV{ELASTIC_CLIENT_APIVERSIONING}) eq 'true' || $ENV{ELASTIC_CLIENT_APIVERSIONING} eq '1')) {
            $default_headers{'Accept'} = 'application/vnd.elasticsearch+json;compatible-with=7';
            $default_headers{'Content-Type'} = 'application/vnd.elasticsearch+json; compatible-with=7';
    }

    if (defined $params->{elastic_cloud_api_key} && defined $params->{token_api}) {
        throw( 'Request',
            "You cannot set elastic_cloud_api_key and token_api together" );
    }

    # Elastic cloud API key
    if (defined $params->{elastic_cloud_api_key}) {
        $default_headers{'Authorization'} = sprintf("ApiKey %s", $params->{elastic_cloud_api_key});
    }

    # Elasticsearch token API (https://www.elastic.co/guide/en/elasticsearch/reference/current/security-api-get-token.html)
    if (defined $params->{token_api}) {
        $default_headers{'Authorization'} = sprintf("Bearer %s", $params->{token_api});
    }

    # Elasticsearch
    $params->{scheme}   = $scheme;
    $params->{is_https} = $scheme eq 'https';
    $params->{host}     = $host;
    $params->{port}     = $port;
    $params->{path}     = $path;
    $params->{userinfo} = $userinfo;
    $host = "[$host]" if Net::IP::ip_is_ipv6($host);
    $params->{uri}             = URI->new("$scheme://$host:$port$path");
    $params->{default_headers} = \%default_headers;
    # Add the client version
    if (defined $params->{client}) {
        $params->{client_version} = substr($params->{client}, 0, index($params->{client}, '_'));
    }

    return $params;
}

#===================================
before 'handle' => sub {
#===================================
    my $self = shift;
    if ( $$ != $self->_pid ) {
        $self->clear_handle;
        $self->_pid($$);
    }
};

#===================================
sub is_live { !shift->next_ping }
sub is_dead { !!shift->next_ping }
#===================================

#===================================
sub mark_live {
#===================================
    my $self = shift;
    $self->ping_failures(0);
    $self->next_ping(0);
}

#===================================
sub mark_dead {
#===================================
    my $self  = shift;
    my $fails = $self->ping_failures;
    $self->ping_failures( $fails + 1 );

    my $timeout
        = min( $self->dead_timeout * 2**$fails, $self->max_dead_timeout );
    my $next = $self->next_ping( time() + $timeout );

    $self->logger->infof( 'Marking [%s] as dead. Next ping at: %s',
        $self->stringify, scalar localtime($next) );

}

#===================================
sub force_ping {
#===================================
    my $self = shift;
    $self->ping_failures(0);
    $self->next_ping(-1);
}

#===================================
sub pings_ok {
#===================================
    my $self = shift;
    $self->logger->infof( 'Pinging [%s]', $self->stringify );
    return try {
        $self->perform_request(
            {   method  => 'HEAD',
                path    => '/',
                timeout => $self->ping_timeout,
            }
        );
        $self->logger->infof( 'Marking [%s] as live', $self->stringify );
        $self->mark_live;
        1;
    }
    catch {
        $self->logger->debug("$_");
        $self->mark_dead;
        0;
    };
}

#===================================
sub sniff {
#===================================
    my $self = shift;
    $self->logger->infof( 'Sniffing [%s]', $self->stringify );
    return try {
        $self->perform_request(
            {   method  => 'GET',
                path    => '/_nodes/http',
                qs      => { timeout => $self->sniff_timeout . 's' },
                timeout => $self->sniff_request_timeout,
            }
        )->{nodes};
    }
    catch {
        $self->logger->debug($_);
        return;
    };
}

#===================================
sub build_uri {
#===================================
    my ( $self, $params ) = @_;
    my $uri = $self->uri->clone;
    $uri->path( $uri->path . $params->{path} );
    my %qs = ( %{ $self->default_qs_params }, %{ $params->{qs} || {} } );
    $uri->query_form( \%qs );
    return $uri;
}

#===================================
before 'perform_request' => sub {
#===================================
    my ( $self, $params ) = @_;
    return unless defined $params->{data};

    $self->_compress_body($params);

    my $max = $self->max_content_length
        or return;

    return if length( $params->{data} ) < $max;

    $self->logger->throw_error( 'ContentLength',
        "Body is longer than max_content_length ($max)",
    );
};

#===================================
sub _compress_body {
#===================================
    my ( $self, $params ) = @_;
    my $output;
    if ( $self->gzip ) {
        IO::Compress::Gzip::gzip( \( $params->{data} ), \$output )
            or throw( 'Request',
            "Couldn't gzip request: $IO::Compress::Gzip::GzipError" );
        $params->{data}     = $output;
        $params->{encoding} = 'gzip';
    }
    elsif ( $self->deflate ) {
        IO::Compress::Deflate::deflate( \( $params->{data} ), \$output )
            or throw( 'Request',
            "Couldn't deflate request: $IO::Compress::Deflate::DeflateError" );
        $params->{data}     = $output;
        $params->{encoding} = 'deflate';
    }
}

#===================================
sub _decompress_body {
#===================================
    my ( $self, $body_ref, $headers ) = @_;
    if ( my $encoding = $headers->{'content-encoding'} ) {
        my $output;
        if ( $encoding eq 'gzip' ) {
            IO::Uncompress::Gunzip::gunzip( $body_ref, \$output )
                or throw(
                'Request',
                "Couldn't gunzip response: $IO::Uncompress::Gunzip::GunzipError"
                );
        }
        elsif ( $encoding eq 'deflate' ) {
            IO::Uncompress::Inflate::inflate( $body_ref, \$output,
                Transparent => 0 )
                or throw(
                'Request',
                "Couldn't inflate response: $IO::Uncompress::Inflate::InflateError"
                );
        }
        else {
            throw( 'Request', "Unknown content-encoding: $encoding" );
        }
        ${$body_ref} = $output;
    }
}

#===================================
sub process_response {
#===================================
    my ( $self, $params, $code, $msg, $body, $headers ) = @_;

    # Product check only for 8+ client API version
    if ( $self->client_version >= 8 and $code >= 200 and $code < 300 ) {
        my $product = $headers->{$PRODUCT_CHECK_HEADER} // '';
        if ($product ne $PRODUCT_CHECK_VALUE) {
            throw( "ProductCheck", "The client noticed that the server is not Elasticsearch and we do not support this unknown product" );
        }
    }

    $self->_decompress_body( \$body, $headers );

    my ($mime_type) = split /\s*;\s*/, ( $headers->{'content-type'} || '' );

    my $is_encoded = $mime_type && $mime_type ne 'text/plain';

    # Deprecation warnings
    if ( my $warnings = $headers->{warning} ) {
        my $warning_string = _parse_warnings($warnings);
        my %temp           = (%$params);
        delete $temp{data};
        $self->logger->deprecation( $warning_string, \%temp );
    }

    # Request is successful
    if ( $code >= 200 and $code <= 209 ) {
        if ( defined $body and length $body ) {
            $body = $self->serializer->decode($body)
                if $is_encoded;
            return $code, $body;
        }
        return ( $code, 1 ) if $params->{method} eq 'HEAD';
        return ( $code, '' );
    }

    # Check if the error should be ignored
    my @ignore = to_list( $params->{ignore} );
    push @ignore, 404 if $params->{method} eq 'HEAD';
    return ($code) if grep { $_ eq $code } @ignore;

    # Determine error type
    my $error_type = $Code_To_Error{$code};
    unless ($error_type) {
        if ( defined $body and length $body ) {
            $msg  = $body;
            $body = undef;
        }
        $error_type = $self->error_from_text( $code, $msg );
    }

    delete $params->{data} if $params->{body};
    my %error_args = ( status_code => $code, request => $params );

    # Extract error message from the body, if present

    if ( $body = $self->serializer->decode($body) ) {
        $error_args{body} = $body;
        $msg = $self->_munge_elasticsearch_exception($body) || $msg;

        $error_args{current_version} = $1
            if $error_type eq 'Conflict'
            and $msg =~ /: version conflict, current (?:version )?\[(\d+)\]/;
    }
    $msg ||= $error_type;

    chomp $msg;
    throw( $error_type, "[" . $self->stringify . "]-[$code] $msg",
        \%error_args );
}

#===================================
sub _parse_warnings {
#===================================
    my @warnings = ref $_[0] eq 'ARRAY' ? @{ shift() } : shift();
    my @str;
    for (@warnings) {
        if ( $_ =~ /^\d+\s+\S+\s+"((?:\\"|[^"])+)"/ ) {
            my $msg = $1;
            $msg =~ s/\\"/"/g, push @str, $msg;
        }
        else {
            push @str, $_;
        }
    }
    return join "; ", @str;
}

#===================================
sub _munge_elasticsearch_exception {
#===================================
    my ( $self, $body ) = @_;
    return $body unless ref $body eq 'HASH';
    my $error = $body->{error} || return;
    return $error unless ref $error eq 'HASH';

    my $root_causes = $error->{root_cause} || [];
    unless (@$root_causes) {
        my $msg = "[" . $error->{type} . "] " if $error->{type};
        $msg .= $error->{reason} if $error->{reason};
        return $msg;
    }

    my $json = $self->serializer;
    my @msgs;
    for (@$root_causes) {
        my %cause = (%$_);
        my $msg
            = "[" . ( delete $cause{type} ) . "] " . ( delete $cause{reason} );
        if ( keys %cause ) {
            $msg .= ", with: " . $json->encode( \%cause );
        }
        push @msgs, $msg;
    }
    return ( join ", ", @msgs );
}

1;

# ABSTRACT: Provides common functionality to HTTP Cxn implementations

__END__