Amazon ElasticSearch 이모저모

AWS ElasticSearch는 간단하게 ElasticSearch를 AWS 내에서 제공하는 서비스

구성은 간단히 Web Console에서 Name(Master), Data Node 구성 가능, 자동 백업이 가능

사용 방법은 http://mykumi.tistory.com/entry/AWS-ElasticSearchKibanaFluentd-1-AWS-ElasticSearch-%EC%83%9D%EC%84%B1 에서 확인 가능

문제는 ElasticSearch는 VPC 내의 Private 서비스가 아닌 Public 구간의 서비스.

따라서 접근 제어는 Public IP 혹은 ARN과 Signature 구성을 통해 가능. 접근 제어 변경 시 약 5분에서 길게는 20분까지 반영되는데 시간이 걸림

Signature 구성 방법: https://github.com/awslabs/amazon-elasticsearch-lambda-samples/blob/master/src/s3_lambda_es.js

Lambda 샘플코드에서 인증 부분

// v1.1.2
var https = require('https');
var crypto = require('crypto');
 
var endpoint = 'search-elasticsearch-cloudfront-xgdpyfpmmhtk3g43iay2i75r34.ap-northeast-2.es.amazonaws.com';
 
function post(body, callback) {
    var requestParams = buildRequest(endpoint, body);
 
    var request = https.request(requestParams, function(response) {
        var responseBody = '';
        response.on('data', function(chunk) {
            responseBody += chunk;
        });
        response.on('end', function() {
            var info = JSON.parse(responseBody);
            var failedItems;
            var success;
            
            if (response.statusCode >= 200 && response.statusCode < 299) {
                failedItems = info.items.filter(function(x) {
                    return x.index.status >= 300;
                });
 
                success = { 
                    "attemptedItems": info.items.length,
                    "successfulItems": info.items.length - failedItems.length,
                    "failedItems": failedItems.length
                };
            }
 
            var error = response.statusCode !== 200 || info.errors === true ? {
                "statusCode": response.statusCode,
                "responseBody": responseBody
            } : null;
 
            callback(error, success, response.statusCode, failedItems);
        });
    }).on('error', function(e) {
        callback(e);
    });
    request.end(requestParams.body);
}
 
function buildRequest(endpoint, body) {
    var endpointParts = endpoint.match(/^([^\.]+)\.?([^\.]*)\.?([^\.]*)\.amazonaws\.com$/);
    var region = endpointParts[2];
    var service = endpointParts[3];
    var datetime = (new Date()).toISOString().replace(/[:\-]|\.\d{3}/g, '');
    var date = datetime.substr(0, 8);
    var kDate = hmac('AWS4' + process.env.AWS_SECRET_ACCESS_KEY, date);
    var kRegion = hmac(kDate, region);
    var kService = hmac(kRegion, service);
    var kSigning = hmac(kService, 'aws4_request');
    
    var request = {
        host: endpoint,
        method: 'POST',
        path: '/_bulk',
        body: body,
        headers: { 
            'Content-Type': 'application/json',
            'Host': endpoint,
            'Content-Length': Buffer.byteLength(body),
            'X-Amz-Security-Token': process.env.AWS_SESSION_TOKEN,
            'X-Amz-Date': datetime
        }
    };
 
    var canonicalHeaders = Object.keys(request.headers)
        .sort(function(a, b) { return a.toLowerCase() < b.toLowerCase() ? -1 : 1; })
        .map(function(k) { return k.toLowerCase() + ':' + request.headers[k]; })
        .join('\n');
 
    var signedHeaders = Object.keys(request.headers)
        .map(function(k) { return k.toLowerCase(); })
        .sort()
        .join(';');
 
    var canonicalString = [
        request.method,
        request.path, '',
        canonicalHeaders, '',
        signedHeaders,
        hash(request.body, 'hex'),
    ].join('\n');
 
    var credentialString = [ date, region, service, 'aws4_request' ].join('/');
 
    var stringToSign = [
        'AWS4-HMAC-SHA256',
        datetime,
        credentialString,
        hash(canonicalString, 'hex')
    ] .join('\n');
 
    request.headers.Authorization = [
        'AWS4-HMAC-SHA256 Credential=' + process.env.AWS_ACCESS_KEY_ID + '/' + credentialString,
        'SignedHeaders=' + signedHeaders,
        'Signature=' + hmac(kSigning, stringToSign, 'hex')
    ].join(', ');
 
    return request;
}
 
function hmac(key, str, encoding) {
    return crypto.createHmac('sha256', key).update(str, 'utf8').digest(encoding);
}
 
function hash(str, encoding) {
    return crypto.createHash('sha256').update(str, 'utf8').digest(encoding);
}
 

CPU 임계치는 70% 이하, Cluster 추가 시 리밸런싱 작업 시 IOPS가 모자를 수 있으므로 미리미리 스토리지 가용성을 확보해둬야 한다. 상황에 따라 며칠이 소요되기도 한다. 장애...

장애 발생 시 Snapshot을 이용해서 다시 노드를 만드는 수 밖에...