PerlでHarrietを使ってElasticsearchのテストをする & 並列でもテストできるように

　Elasticsearchを使った開発をしていると、Elasticsearchを使った機能でも簡単にテストしたいという気持ちになってくる。またproveの-jオプションを使って並列に動かしていても変にコンフリクトせずにいい感じになってほしい。

　この課題を解決するために、Harrietを用いてテスト実行前にElasticsearchを起動し、テストを行うというのをやってみた。

作戦

Harrietを用いて、テスト実行前にElasticsearchのnodeを一つ起動する
並列数分、Elasticsearchのインデックスを作成し、ついでにロック用のインデックスも作っておく
環境変数で情報をテストファイルに渡す
テストファイルでは環境変数から情報を読み込んで、ロックを取りつつ、立ち上がったportやインデックスにアクセスする

HarrietでElasticsearchの起動 & インデックスの作成

　Harrietについてはtokuhirom blog を参照。

　t/harriet/elasticsearch.plを用意して、以下のように書く。

use strict;
use warnings;
use utf8;

use JSON::XS qw(encode_json);
use Test::TCP;
use App::Prove;

use Search::Elasticsearch;

# ctrl-cした時とかにguardが開放されないのでここで明示的に開放してkillしている
# See also: http://perl-users.jp/articles/advent-calendar/2012/hacker/9
sub install_signal_handler ($) {
    my ($sig) = @_;

    my $original_handler = $SIG{$sig};
    $SIG{$sig} = sub {
        $HARRIET_GUARDS::ELASTICSEARCH = undef;
        $SIG{$sig} = $original_handler || 'DEFAULT';
        kill $sig => $$;
    };
}

# テスト用にElasticsearchを一台立ち上げる
my $now = time;
my $test_process_pid = $$;
$HARRIET_GUARDS::ELASTICSEARCH = Test::TCP->new(
    code => sub {
        my $port = shift;

        # elasticsearchのtcp側は大体100増やしたportを使うので、それに合わせる
        my $tcp_port = $port + 100;
        exec(
            'elasticsearch/bin/elasticsearch',
            # テストが同時に実行された時にそれらで勝手にクラスタを作られると困るので
            # ジョブのプロセスIDをクラスタ名に入れる。またpidだけではいつか
            # 被る恐れがあるので、現在時刻もクラスタ名に入れる
            sprintf("--cluster.name=sample-test-%s-%s", $now, $test_process_pid),
            "--http.port=$port",
            "--transport.tcp.port=$tcp_port",
        );
    },
);
my $http_port = $HARRIET_GUARDS::ELASTICSEARCH->port;

# インデックスの作成
my $es = Search::Elasticsearch->new(nodes => ["localhost:$http_port"]);
my $elasticsearch_index_names = [];

my $setup = sub {
    my $j = shift;

    my $index_name = "sample-$j";

    # テストで利用するインデックスを作成。これは別ファイルで
    # Mappingsを定義しておいて流しこむのが良いと思う
    $es->indices->create(
        index => $index_name,
        body => {
            mappings => {
                hoge => {
                    properties => {
                        foo => { type => 'string' },
                    },
                },
            },
        },
    );

    # 並列時にロックを取るためのMappingを用意
    $es->indices->put_mapping(
        index => $index_name,
        type  => 'lock',
        body  => {
            lock => {},
        },
    );

    push @$elasticsearch_index_names, $index_name;
};

# テストの並列数分インデックスを作る
my $prove = App::Prove->new;
$prove->process_args(@ARGV);
my $jobs_n = $prove->jobs || 1;
$setup->($_) for (1..$jobs_n);

# INTとTERMのときに正しくguardが開放されるようにする
install_signal_handler($_) for qw(INT TERM);

# 環境変数を経由して、portと作成したインデックス名をテストに渡す
$ENV{TEST_HARRIET_ELASTICSEARCH_JSON} = encode_json({
    http_port   => $http_port,
    index_names => $elasticsearch_index_names,
});

　すごく長いし難しいのだが、それぞれやっていることはコメントに書いた。ポイントとしては

並列に実行するとして、その台数分Elasticsearchを起動するのは重いので、インデックスを複数作るという方向で考える
Elasticsearchはクラスタを勝手に作るので、クラスタ名が被らないように工夫する
$HARRIET_GUARDSというところに入れておけば、テスト終了時にオブジェクトが破棄され、Elasticsearchを終了できる

テストファイルから起動したElasticsearch及び作成したインデックスにアクセスする

　環境変数で情報は渡ってきているので、それをテストファイルから利用したら良い。

　例えば以下のようにt/elasticsearch.tを作る。

use strict;
use warnings;
use utf8;

use lib 't/lib';

use Test::More;
use Search::Elasticsearch;
use JSON::XS qw(decode_json);
use Guard qw(guard);

my $http_port;
my $free_index_name;
my $elasticsearch_unlock;

if (my $elasticsearch = eval { decode_json($ENV{TEST_HARRIET_ELASTICSEARCH_JSON}) }) {
    $http_port = $elasticsearch->{http_port};

    # ロック用のドキュメントID。どんな値でも良いので1にしている。
    my $LOCK_ID = 1;

    # Harrietで作ったインデックスに対してロックが取れるまで待つ
    # 並列数分インデックスは作成しているので、一瞬で取れる想定
    my $es = Search::Elasticsearch->new(nodes => ["localhost:$http_port"]);
    GET_INDEX: for my $count (1..100) {
        for my $index_name (@{ $elasticsearch->{index_names} }) {
            eval {
                $es->create(
                    index => $index_name,
                    type  => 'lock',
                    id    => $LOCK_ID,
                    body  => {},
                );
            };
            unless ($@) {
                $free_index_name = $index_name;
                last GET_INDEX;
            }
        }

        # 取れなかったら1秒待つ
        sleep(1);
    }

    die 'No elasticsearch index acquired' unless $free_index_name;

    # オブジェクトが破棄されたらロックが解除されるようにしておく
    $elasticsearch_unlock = guard {
        my $es = Search::Elasticsearch->new(
            nodes  => ["localhost:$http_port"],
        );

        $es->delete(
            index => $free_index_name,
            type  => 'lock',
            id    => $LOCK_ID,
        );
    };
}

my $es = Search::Elasticsearch->new(nodes => ["localhost:$http_port"]);
ok $es->indices->exists(index => $free_index_name), 'index exists';

undef $elasticsearch_unlock;

done_testing();