Apex Go에서 ELB 액세스 로그를 AWS Lambda에서 Elasticsearch로 가져오기

벌써 몇몇 분이 기사를 쓰고 있습니다만, Go 언어의 정보가 적기 때문에 써 보았습니다.
기본적으로는 Developers.IO의 후지모토 씨의 기사 을 바탕으로 Apex Go 로 구현해 보았습니다.

AWS Lambda Function 생성

완성판의 Lambda 코드는 Gist 를 봐 주세요. 이하, 요점만 설명합니다.

전체 흐름

ELB가 액세스 로그를 S3에 기록하면 메타 정보가 S3 이벤트로 Lambda에 전달되어 해당 Lambda 함수를 실행합니다.
이하, 처리 상세하게 됩니다.

Lambda 핸들러가 S3 이벤트를 인수로 수신합니다.

S3 이벤트에는 슬라이스로 복수의 액세스 로그 파일명이 포함되어 있으므로 루프로 처리한다.

S3에 접속해, 액세스 로그 파일의 컨텐츠를 취득한다.

취득한 파일을 1 행마다 루프로 처리한다. (1 행이 1 HTTP 요청)

로그를 JSON 형식으로 변환합니다. (Go struct 태그를 사용하여.)

Elasticsearch에 POST합니다. (Indexing. 문서 추가.)

Apex Lambda 핸들러

콜백에서 apexS3.Event.Records 에 S3 버킷에 업로드된 파일 이름 목록이 전달됩니다.

main.go

func main() {
    apexS3.HandleFunc(func(event *apexS3.Event, ctx *apex.Context) error {
        for _, rec := range event.Records {
                // ファイルの処理。Recordsには、ファイルではなくファイル名のリストが入っている。

참고: 파일이 전달되지 않습니다.

S3 버킷에서 파일 가져오기

rec.S3.Object.Key 에 S3의 파일명이 세트 되고 있습니다. 키로 S3에서 파일을 다운로드합니다.

main.go

            svc := s3.New(session.New(&aws.Config{Region: aws.String(rec.AWSRegion)}))

            // S3バケットからオブジェクトを取得する。
            s3out, err := svc.GetObject(&s3.GetObjectInput{
                Bucket: aws.String(rec.S3.Bucket.Name),
                Key:    aws.String(rec.S3.Object.Key),
            })
                        // バイト配列に一気に読み込む。
            bytes, err := ioutil.ReadAll(s3out.Body)
                        // 改行で分割する。
            for _, line := range strings.Split(string(bytes), "\n") {
                        // JSON形式に変換。
                        // Elasticsearchへ登録。

JSON 형식으로 변환

Elasticsearch에 등록할 문서의 스키마입니다. Struct 태그를 사용하여 JSON으로 만들고 있습니다. 데이터 유형은 Elasticsearch 측에서 잘 변환합니다. timestamp 는 date 형으로 변환되어 Kibana 에서 타임프레임으로 그래프를 draw 하는데 사용하는 필드가 됩니다.

main.go

type ElbAccessLog struct {
    Timestamp              string `json:"timestamp"`
    Elb                    string `json:"elb"`
    ClientIpAddress        string `json:"client_ip_address"`
    BackendIpAddress       string `json:"backend_ip_address"`
    RequestProcessingTime  string `json:"request_processing_time"`
    BackendProcessingTime  string `json:"backend_processing_time"`
    ResponseProcessingTime string `json:"response_processing_time"`
    ElbStatusCode          string `json:"elb_status_code"`
    BackendStatusCode      string `json:"backend_status_code"`
    ReceivedBytes          string `json:"received_bytes"`
    SentBytes              string `json:"sent_bytes"`
    Request                string `json:"request"`
}

Elasticsearch로 색인 생성

AWS 버전의 Elasticsearch 서비스에 연결하는 것은 조금 번거롭습니다.
olivere/elastic 과 edoardo89 의 RoundTripper 를 사용합니다. 시도하지 않지만 디버그 로그를 출력 할 수있는 것 같습니다.

라이브러리를 가져옵니다.

$ go get gopkg.in/olivere/elastic.v3
$ go get github.com/edoardo849/apex-aws-signer

Elasticsearch에 연결합니다.

main.go

            transport := signer.NewTransport(session.New(&aws.Config{Region: aws.String(rec.AWSRegion)}), elasticsearchservice.ServiceName)

            httpClient := &http.Client{
                Transport: transport,
            }

            // Elasticsearchに接続
            client, err := elastic.NewClient(
                elastic.SetSniff(false),
                elastic.SetURL(esUrl), // AWS Elasticsearchのエンドポイント
                elastic.SetScheme("https"),
                elastic.SetHttpClient(httpClient),
            )

Document를 등록(Indexing)합니다.

main.go

    put1, err := client.Index().
        Index(indexName). // インデックス名
        Type(esType).     // タイプ名
        BodyJson(doc).    // ドキュメントのコンテンツ
        Do()

완성된 코드를 Apex를 사용하여 Lambda에 등록합니다. dora63 님의 기사가 상세합니다.

AWS Lambda 콘솔에서 트리거 설정

AWS Lambda 콘솔에 로그인하고 Lambda 함수의 트리거에 S3 버킷(ELB 액세스 로그가 있는 곳)을 끈다.


적당히 ELB에 액세스해 봅니다. Lambda가 실행되고 Elasticsearch에 액세스 로그가 인덱싱됩니다. Chrome의 Sense 플러그인에서 검색합니다.

로컬에서 실행해보기

Apex의 장점은 Lambda 프로그램을 그대로 로컬로 실행할 수 있다는 것입니다. 이벤트는 표준 입력으로 전달됩니다.

go run main.go < s3event.json

s3event.json 는 실제로 AWS에서 Lambda가 받은 S3 이벤트 apexS3.Event입니다.

s3event.json

{
  "event": {
    "Records": [
      {
        "eventVersion": "2.0",
        "eventSource": "aws:s3",
        "awsRegion": "ap-northeast-1",
        "eventTime": "2016-09-15T23:56:16.822Z",
        "eventName": "ObjectCreated:CompleteMultipartUpload",
        "userIdentity": {
          "principalId": "AWS:Axxxx"
        },
        "requestParameters": {
          "sourceIPAddress": "52.x.x.x"
        },
        "S3": {
          "s3SchemaVersion": "1.0",
          "configurationId": "xxx",
          "bucket": {
            "name": "xxx",
            "ownerIdentity": {
              "principalId": "xxx"
            },
            "arn": "arn:aws:s3:::xxx"
          },
          "object": {
            "key": "AWSLogs/xxx/elasticloadbalancing/ap-northeast-1/2016/09/15/xxx_elasticloadbalancing_ap-northeast-1_elb1_20160915T2355Z_xxx_19usc64f.log",
            "size": 1212,
            "eTag": "8b0eb26c9b3b0fb86b5b7c9cafbc7331-1",
            "versionId": "",
            "sequencer": "0057DB3520885F756B"
          }
        }
      }
    ]
  }
}

독립 실행형으로 움직일 수 있는 것은 꽤 디버깅이 편해집니다. 사실,이 프로그램은 거의이 더미 이벤트를 사용하여 개발할 수있었습니다.

빠진 곳

AWS와 함께 제공되는 Kibana이지만 소수점은 3자리로 반올림됩니다. 응답 시간 등 밀리초로 반올림됩니다. Number 형식의 정확도를 Settings의 format:numberPrecision에서 변경할 수 있지만 AWS ES에서 사용하는 Kibana 4.1.2는 버그처럼입니다.

AWS의 역할 정책 설정은 좀처럼 이루어지지 않습니다. . . Lambda 프로그램에서 S3 버킷, Elasticsearch 및 CloudWatch에 대한 액세스 권한이 필요합니다.