File Coverage

blib/lib/App/Prove/Elasticsearch/Indexer.pm

Criterion	Covered	Total	%
statement	81	88	92.0
branch	31	36	86.1
condition	10	23	43.4
subroutine	11	12	91.6
pod	5	5	100.0
total	138	164	84.1

# ABSTRACT: Define what data is to be uploaded to elasticsearch, and handle it's uploading
# PODNAME: App::Prove::Elasticsearch::Indexer

package App::Prove::Elasticsearch::Indexer;
$App::Prove::Elasticsearch::Indexer::VERSION = '0.001';
use strict;
use warnings;

use App::Prove::Elasticsearch::Utils();

use Search::Elasticsearch();
use List::Util 1.33;

our $index = 'testsuite';

sub index {
    return $index;
}

our $max_query_size = 1000;
our $e;
our $bulk_helper;
our $idx;

sub check_index {
    my $conf = shift;

    my $port = $conf->{'server.port'} ? ':' . $conf->{'server.port'} : '';
    die "server must be specified" unless $conf->{'server.host'};
    die("port must be specified")  unless $port;
    my $serveraddress = "$conf->{'server.host'}$port";
    $e //= Search::Elasticsearch->new(
        nodes => $serveraddress,
    );

    #XXX for debugging
    #$e->indices->delete( index => $index );

    if (!$e->indices->exists(index => $index)) {
        $e->indices->create(
            index => $index,
            body  => {
                index => {
                    number_of_shards   => "3",
                    number_of_replicas => "2",
                    similarity         => {default => {type => "classic"}}
                },
                analysis => {
                    analyzer => {
                        default => {
                            type      => "custom",
                            tokenizer => "whitespace",
                            filter    => [
                                'lowercase', 'std_english_stop', 'custom_stop'
                            ]
                        }
                    },
                    filter => {
                        std_english_stop => {
                            type      => "stop",
                            stopwords => "_english_"
                        },
                        custom_stop => {
                            type      => "stop",
                            stopwords => [ "test", "ok", "not" ]
                        }
                    }
                },
                mappings => {
                    testsuite => {
                        properties => {
                            id       => {type => "integer"},
                            elapsed  => {type => "integer"},
                            occurred => {
                                type   => "date",
                                format => "yyyy-MM-dd HH:mm:ss"
                            },
                            executor => {
                                type        => "text",
                                analyzer    => "default",
                                fielddata   => "true",
                                term_vector => "yes",
                                similarity  => "classic",
                                fields      => {keyword => {type => "keyword"}}
                            },
                            status => {
                                type        => "text",
                                analyzer    => "default",
                                fielddata   => "true",
                                term_vector => "yes",
                                similarity  => "classic",
                                fields      => {keyword => {type => "keyword"}}
                            },
                            version => {
                                type        => "text",
                                analyzer    => "default",
                                fielddata   => "true",
                                term_vector => "yes",
                                similarity  => "classic",
                                fields      => {keyword => {type => "keyword"}}
                            },
                            test_version => {
                                type        => "text",
                                analyzer    => "default",
                                fielddata   => "true",
                                term_vector => "yes",
                                similarity  => "classic",
                                fields      => {keyword => {type => "keyword"}}
                            },
                            platform => {
                                type        => "text",
                                analyzer    => "default",
                                fielddata   => "true",
                                term_vector => "yes",
                                similarity  => "classic",
                                fields      => {keyword => {type => "keyword"}}
                            },
                            path => {
                                type        => "text",
                                analyzer    => "default",
                                fielddata   => "true",
                                term_vector => "yes",
                                similarity  => "classic",
                                fields      => {keyword => {type => "keyword"}}
                            },
                            defect => {
                                type        => "text",
                                analyzer    => "default",
                                fielddata   => "true",
                                term_vector => "yes",
                                similarity  => "classic",
                                fields      => {keyword => {type => "keyword"}}
                            },
                            steps_planned => {type => "integer"},
                            body          => {
                                type        => "text",
                                analyzer    => "default",
                                fielddata   => "true",
                                term_vector => "yes",
                                similarity  => "classic",
                            },
                            name => {
                                type        => "text",
                                analyzer    => "default",
                                fielddata   => "true",
                                term_vector => "yes",
                                similarity  => "classic",
                                fields      => {keyword => {type => "keyword"}}
                            },
                            steps => {
                                properties => {
                                    number  => {type => "integer"},
                                    text    => {type => "text"},
                                    status  => {type => "text"},
                                    elapsed => {type => "integer"},
                                }
                            },
                        }
                    }
                }
            }
        );
        return 1;
    }
    return 0;
}

sub index_results {
    my ($result) = @_;

    die("check_index must be run first") unless $e;

    $idx //= App::Prove::Elasticsearch::Utils::get_last_index($e, $index);
    $idx++;

    $e->index(
        index => $index,
        id    => $idx,
        type  => $index,
        body  => $result,
    );

    my $doc_exists =
      $e->exists(index => $index, type => 'testsuite', id => $idx);
    if (!defined($doc_exists) || !int($doc_exists)) {
        die
          "Failed to Index $result->{'name'}, could find no record with ID $idx\n";
    } else {
        print
          "Successfully Indexed test: $result->{'name'} with result ID $idx\n";
    }
}

sub bulk_index_results {
    my @results = @_;
    $bulk_helper //= $e->bulk_helper(
        index => $index,
        type  => $index,
    );

    $idx //= App::Prove::Elasticsearch::Utils::get_last_index($e, $index);

    $bulk_helper->index(map { $idx++; {id => $idx, source => $_} } @results);
    $bulk_helper->flush();
}

sub associate_case_with_result {
    my %opts = @_;

    die("check_index must be run first") unless $e;

    my %q = (
        index => $index,
        body  => {
            query => {
                bool => {
                    must => [
                        {
                            match => {
                                name => $opts{case},
                            }
                        },
                    ],
                },
            },
        },
    );

    #It's normal to have multiple platforms in a document.
    foreach my $plat (@{$opts{platforms}}) {
        push(@{$q{body}{query}{bool}{must}}, {match => {platform => $plat}});
    }

    #It's NOT normal to have multiple versions in a document.
    foreach my $version (@{$opts{versions}}) {
        push(
            @{$q{body}{query}{bool}{should}},
            {match => {version => $version}}
        );
    }

    #Paginate the query, TODO short-circuit when we stop getting results?
    my $hits = App::Prove::Elasticsearch::Utils::do_paginated_query(
        $e, $max_query_size,
        %q
    );
    return 0 unless scalar(@$hits);

    #Now, update w/ the defect.
    my $failures = 0;
    my $attempts = 0;
    foreach my $hit (@$hits) {
        $hit->{_source}->{platform} = [ $hit->{_source}->{platform} ]
          if ref($hit->{_source}->{platform}) ne 'ARRAY';
        next if (scalar(@{$opts{versions}}) && !$hit->{_source}->{version});
        next
          unless List::Util::any { $hit->{_source}->{version} eq $_ }
        @{$opts{versions}};
        next if (scalar(@{$opts{platforms}}) && !$hit->{_source}->{platform});
        next unless List::Util::all {
            my $p = $_;
            grep { $_ eq $p } @{$hit->{_source}->{platform}}
        }
        @{$opts{platforms}};
        next unless $hit->{_source}->{name} eq $opts{case};

        $attempts++;

        #Merge the existing defects with the ones we are adding in
        $hit->{defect} //= [];
        my @df_merged =
          List::Util::uniq((@{$hit->{defect}}, @{$opts{defects}}));

        my %update = (
            index => $index,
            id    => $hit->{_id},
            type  => 'result',
            body  => {
                doc => {
                    defect => \@df_merged,
                },
            }
        );
        $update{body}{doc}{status} = $opts{status} if $opts{status};

        my $res = $e->update(%update);

        print "Associated cases to document $hit->{_id}\n"
          if $res->{result} eq 'updated';
        if (!grep { $res->{result} eq $_ } qw{updated noop}) {
            print
              "Something went wrong associating cases to document $hit->{_id}!\n$res->{result}\n";
            $failures++;
        }
    }

    print "No cases matching your query could be found.  No action was taken.\n"
      unless $attempts;

    return $failures;
}

1;

__END__