Elasticsearch SDK と API による Log Service への接続 - SLS

このトピックでは、Elasticsearch SDK および Elasticsearch 互換 API を使用して Simple Log Service でデータを分析する方法について説明します。

重要

本ドキュメントは Alibaba Cloud が独自に作成したものであり、著作権は Alibaba Cloud に帰属します。本ドキュメントでは、Alibaba Cloud サービスとサードパーティ製品との連携方法について記述しています。サードパーティの名称および製品名の記載は、あくまで参考目的のみです。

前提条件

プロジェクトと標準ログストアが作成され、ログが収集されます。詳細については、「プロジェクトの管理」、「基本的なログストアの作成」、および「データ収集」をご参照ください。
ログのクエリを実行する前に、インデックスが作成済みであること。詳細については、「インデックスの作成」をご参照ください。
RAM ユーザーに対して AccessKey ペアが作成され、ログストアでログをクエリするために必要な権限が付与されます。詳細については、「RAM 認可」をご参照ください。

注意事項

Elasticsearch 互換 API は、Elasticsearch SDK のバージョン 7.x のみをサポートします。
クエリ内で @timestamp を指定しない場合、デフォルトで過去 24 時間分のデータが取得されます。

パラメーター

パラメーター	説明
`hosts`	データアクセスアドレス。フォーマットは `https://${プロジェクト名}.${プロジェクトアクセスドメイン名}/es/` です。詳細については、「エンドポイント」をご参照ください。重要 HTTPS プロトコルのみがサポートされます。
`Username`	`[ユーザー名]` と `[パスワード]` を入力します。 RAM ユーザーの AccessKey ペアを使用することをお勧めします。 RAM ユーザーには、LogStore のデータをクエリする権限が必要です。この権限は、権限アシスタントを使用して設定できます。詳細については、「権限アシスタントを設定する」をご参照ください。 AccessKey ペアを取得するには、「AccessKey ペア」をご参照ください。
`Password`
`index`	`${ProjectName}.${LogstoreName}`

サンプル

以下のサンプルでは、Elasticsearch SDK および Elasticsearch 互換 API を使用して Simple Log Service でデータを分析する方法を示します。これらのサンプルでは、プロジェクト名は etl-dev、Logstore 名は accesslog、Simple Log Service のエンドポイントは cn-huhehaote.log.aliyuncs.com です。

cURL サンプル

curl -u ${ALIYUN_ACCESS_KEY_ID}:${ALIYUN_ACCESS_KEY_SECRET} "https://etl-dev.cn-huhehaote.log.aliyuncs.com/es/etl-dev.accesslog/_search?q=status:200"

Python SDK サンプル

依存関係をインストールします。
```
pip install elasticsearch==7.17
```

サンプルコード

#!/bin/env python3
import os
import json
import time
from elasticsearch import Elasticsearch, helpers

slsProject = "etl-dev"
slsEndpoint = "cn-huhehaote.log.aliyuncs.com"
slsLogstore = "accesslog"

esHost = "https://%s.%s/es/" % (slsProject, slsEndpoint)
esIndex = "%s.%s" % (slsProject, slsLogstore)

# 環境変数から AccessKey ペアを取得します。
accessKeyId = os.environ['ALIYUN_ACCESS_KEY_ID']
accessKeySecret = os.environ['ALIYUN_ACCESS_KEY_SECRET']

esClient = Elasticsearch(hosts=esHost,
                http_auth=(accessKeyId, accessKeySecret),
                   verify_certs=True, timeout=300)

endTime = int(time.time()*1000)
startTime = endTime - 3600*1000

r = esClient.search(
    index=esIndex,
    body=   {
        "query": {
            "bool": {
                "filter": [
                    {
                        "range": {
                            "@timestamp": {
                                "gte": startTime,
                                "lte": endTime,
                                "format": "epoch_millis"
                            }
                        }
                    }
                ]
            }
        }
     }
)

print(json.dumps(r, indent=4))

Elasticsearch DSL サンプル

Elasticsearch DSL（ドメイン固有言語）は、Elasticsearch におけるデータ検索のためのクエリ言語です。DSL クエリを手動で構築せずに済むよう、以下のクライアントライブラリを使用してください。

依存関係をインストールします。
```
pip install elasticsearch-dsl==7.4.1
```

Elasticsearch DSL サンプル

#!/bin/env python3
import os
import json
import time
from elasticsearch import Elasticsearch, helpers
from elasticsearch_dsl import Search, Q

slsProject = "etl-dev"
slsEndpoint = "cn-huhehaote.log.aliyuncs.com"
slsLogstore = "accesslog"

esHost = "https://%s.%s/es/" % (slsProject, slsEndpoint)
esIndex = "%s.%s" % (slsProject, slsLogstore)

# 環境変数から AccessKey ペアを取得します。
accessKeyId = os.environ['ALIYUN_ACCESS_KEY_ID']
accessKeySecret = os.environ['ALIYUN_ACCESS_KEY_SECRET']

esClient = Elasticsearch(hosts=esHost,
                http_auth=(accessKeyId, accessKeySecret),
                   verify_certs=True, timeout=300)

endTime = int(time.time()*1000)
startTime = endTime - 3600*1000

s = Search(using=esClient, index=esIndex) \
        .filter(Q("range", **{"@timestamp": {"gte": startTime, "lt": endTime, "format": "epoch_millis"}}))  \
        .query("match", request_method="GET") \

response = s.execute()

for hit in response:
    # request_method、host、client_ip は Simple Log Service のログ内のフィールドです。
    print(hit.request_method, hit.host, hit.client_ip)

Go SDK サンプル

package main

import (
	"context"
	"fmt"
	"os"
	"time"

	"github.com/olivere/elastic/v7"
)

func main() {
	// 以下のコードは、Elasticsearch SDK を使用して Simple Log Service の Elasticsearch 互換 API にアクセスする方法を示します。
	slsProject := "etl-dev"
	slsLogstore := "accesslog"
	slsEndpoint := "cn-huhehaote.log.aliyuncs.com"

	accessKeyID := os.Getenv("ALIYUN_ACCESS_KEY_ID")
	accessKeySecret := os.Getenv("ALIYUN_ACCESS_KEY_SECRET")
	esHost := fmt.Sprintf("https://%s.%s:443/es", slsProject, slsEndpoint)
	esIndex := fmt.Sprintf("%s.%s", slsProject, slsLogstore)

	esClient, err := elastic.NewClient(
		elastic.SetURL(esHost),
		elastic.SetSniff(false),
		elastic.SetBasicAuth(accessKeyID, accessKeySecret), // 基本認証のユーザー名およびパスワードを設定します。
		elastic.SetHealthcheck(false),                      // ヘルスチェックを無効化します。
	)
	if err != nil {
		panic(err)
	}

	termQuery := elastic.NewTermQuery("request_method", "GET")
	endTime := time.Now().Unix()
	startTime := endTime - 3600
	timeRangeQuery := elastic.NewRangeQuery("@timestamp").Gte(startTime).Lte(endTime)

	boolQuery := elastic.NewBoolQuery()
	boolQuery = boolQuery.Must(timeRangeQuery, termQuery)

	searchResult, err := esClient.Search().
		Index(esIndex).
		Query(boolQuery).
		From(0).Size(10).
		Pretty(true).
		Do(context.Background())
	if err != nil {
		panic(err)
	}

	// 結果を出力します。
	for _, hit := range searchResult.Hits.Hits {
		fmt.Println(string(hit.Source))
	}
}

Java SDK サンプル

依存関係を pom.xml ファイルに追加します。

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>org.example</groupId>
    <artifactId>estest</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.elasticsearch.client</groupId>
            <artifactId>elasticsearch-rest-high-level-client</artifactId>
            <version>7.10.1</version>
        </dependency>
        <dependency>
            <groupId>org.elasticsearch.client</groupId>
            <artifactId>elasticsearch-rest-client</artifactId>
            <version>7.10.1</version>
        </dependency>
    </dependencies>
</project>

サンプルコード

package org.example;

import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.MatchQueryBuilder;
import org.elasticsearch.index.query.RangeQueryBuilder;
import org.elasticsearch.search.builder.SearchSourceBuilder;

import java.io.IOException;

public class Main {
    public static void main(String[] args) throws IOException {

        String slsProject = "etl-dev";
        String slsLogstore = "accesslog";
        String slsEndpoint = "cn-huhehaote.log.aliyuncs.com";

        String schema = "https";
        String esHost = slsProject + "." +  slsEndpoint; // ${project}.${endpoint}
        int port = 443;
        String esIndex = slsProject + "." + slsLogstore; // ${project}.${logstore}
        String esPrefix = "/es/";
        String accessKeyId = System.getenv("ALIYUN_ACCESS_KEY_ID");
        String accessKeySecret = System.getenv("ALIYUN_ACCESS_KEY_SECRET");

        final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
        credentialsProvider.setCredentials(AuthScope.ANY,
                new UsernamePasswordCredentials(accessKeyId, accessKeySecret));

        RestClientBuilder builder = RestClient.builder(new HttpHost(esHost, port, schema)).setHttpClientConfigCallback(
                    httpClientBuilder -> httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider));

        // /es/ パスプレフィックスを設定します。
        builder.setPathPrefix(esPrefix);
        RestHighLevelClient client = new RestHighLevelClient(builder);

        // クエリを構築します。
        BoolQueryBuilder boolExpr= new BoolQueryBuilder();

        long endTime = System.currentTimeMillis();
        long startTime = endTime - 3600 * 1000;
        boolExpr.filter().add(new MatchQueryBuilder("request_method", "GET"));
        boolExpr.filter().add(new RangeQueryBuilder("@timestamp").gte(startTime).lte(endTime).format("epoch_millis"));

        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();

        searchSourceBuilder.query(boolExpr);
        SearchRequest searchRequest = new SearchRequest(esIndex);
        searchRequest.source(searchSourceBuilder);
        SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
        System.out.println(searchResponse.toString());

        client.close();
    }
}

PHP SDK サンプル

Composer を使用して Elasticsearch PHP クライアントをインストールします。
```
composer require elasticsearch/elasticsearch
```

サンプルコード

<?php

require 'vendor/autoload.php';

use Elasticsearch\ClientBuilder;

$slsProject = 'etl-dev';
$slsLogstore = 'accesslog';
$slsEndpoint = 'cn-huhehaote.log.aliyuncs.com';

$esHost = $slsProject . '.' . $slsEndpoint;
$esIndex = $slsProject . '.' . $slsLogstore;

$accessKeyId = getenv('ALIYUN_ACCESS_KEY_ID');
$accessKeySecret = getenv('ALIYUN_ACCESS_KEY_SECRET');

$hosts = [
    [
        'host' => $esHost,
        'port' => '443',
        'scheme' => 'https',
        'path' => '/es',
        'user' => $accessKeyId,
        'pass' => $accessKeySecret,
    ]
];

$client = ClientBuilder::create()
    ->setHosts($hosts)
    ->build();

$endTime = round(microtime(true) * 1000); // 単位：ミリ秒
$startTime = $endTime - (3600 * 1000);


$params = [
    'index' => $esIndex,
    'body'  => [
        'query' => [
            'bool' => [
                'must' => [
                    'match' => [
                        'request_method' => 'GET'
                    ]
                ],
                'filter' => [
                    'range' => [
                        '@timestamp' => [
                            'gte' => $startTime,
                            'lte' => $endTime
                        ]
                    ]
                ]
            ]
        ]
    ]
];

$response = $client->search($params);

print_r($response);