view release on metacpan or  search on metacpan

lib/App/ElasticSearch/Utilities.pm  view on Meta::CPAN

        if( @servers > 0 ) {
            my $pick = @servers > 1 ? $servers[int(rand(@servers))] : $servers[0];
            return App::ElasticSearch::Utilities::Connection->new(%{$pick});
        }
    }
    else {
        # Check for index metadata
        foreach my $k ( keys %conn ) {
            foreach my $name ( $DEF{INDEX}, $DEF{BASE} ) {
                next unless $name;
                if( my $v = es_local_index_meta($k => $name) ) {
                    $conn{$k} = $v;
                    last;
                }
            }
        }
    }

    # Otherwise, cache our handle
    $ES ||= App::ElasticSearch::Utilities::Connection->new(%conn);

    return $ES;
}


sub es_master {
    my ($instance) = @_;
    if(!defined $instance && defined $CLUSTER_MASTER) {
        return $CLUSTER_MASTER;
    }
    my $is_master = 0;
    my @request = ('/_cluster/state/master_node');
    unshift @request, $instance if defined $instance;

    my $cluster = es_request(@request);
    if( defined $cluster && $cluster->{master_node} ) {
        my $local = es_request('/_nodes/_local');
        if ($local->{nodes} && $local->{nodes}{$cluster->{master_node}}) {
            $is_master = 1;
        }
    }
    $CLUSTER_MASTER = $is_master unless defined $instance;
    return $is_master;
}


sub es_request {
    my $instance = ref $_[0] eq 'App::ElasticSearch::Utilities::Connection' ? shift @_ : es_connect();

    $CURRENT_VERSION = _get_es_version() if !defined $CURRENT_VERSION;

    my($url,$options,$body) = _fix_version_request(@_);

    # Normalize the options
    $options->{method} ||= 'GET';
    $options->{command} = $url;
    my $index;

    if( exists $options->{index} ) {
        if( my $index_in = delete $options->{index} ) {
            # No need to validate _all
            if( $index_in eq '_all') {
                $index = $index_in;
            }
            else {
                # Validate each included index
                my @indexes = is_arrayref($index_in) ? @{ $index_in } : split /\,/, $index_in;
                $index = join(',', @indexes);
            }
        }
    }

    # For the cat api, index goes *after* the command
    if( $url =~ /^_(cat|stats)/ && $index ) {
        $url =~ s/\/$//;
        $url = join('/', $url, $index);
        delete $options->{command};
    }
    elsif( $index ) {
        $options->{index} = $index;
    }
    else {
        $index = '';
    }

    # Figure out if we're modifying things
    my $modification = $url eq '_search' && $options->{method} eq 'POST' ? 0
                     : $options->{method} ne 'GET';

    if($modification) {
        # Set NOOP if necessary
        if(!$DEF{NOOP} && $DEF{MASTERONLY}) {
            if( !es_master() ) {
                $DEF{NOOP} = 1;
            }
        }

        # Check for noop
        if( $DEF{NOOP} ) {
            my $flag = $DEF{MASTERONLY} && !es_master() ? '--master-only' : '--noop';
            output({color=>'cyan'}, "Called es_request($index/$options->{command}), but $flag set and method is $options->{method}");
            return;
        }
    }

    # Make the request
    my $resp = $instance->request($url,$options,$body);

    # Check the response is defined, bail if it's not
    die "Unsupported request method: $options->{method}" unless defined $resp;

    # Logging
    verbose({color=>'yellow'}, sprintf "es_request(%s/%s) returned HTTP Status %s",
        $index, $options->{command}, $resp->message,
    ) if $resp->code != 200;

    # Error handling
    if( !$resp->is_success ) {
        my $msg;
        eval {
            my @causes = ();