2020/01/07 추가:

Python 3.8/Elasticsearch Service 6.8 지원 버전으로 업데이트된 기사는 여기입니다.

CloudFront 액세스 로그를 Elasticsearch Service 6.8에 포함하는 메모(Lambda Python 3.8)

【주의! 】(2019/04/17 추가)

내 관측 범위에서 2019/04/17 아침부터 Lambda 환경의 AWS SDK for Python 버전이 증가했으며 Pyhthon 2.7 (및 3.6) 환경의 botocore가 1.10에서 1.12로 변경되었습니다.

이로 인해 "BotocoreHTTPSession 없음"이라는 오류가 발생합니다.

그래서 일부 내용을 다시 써 botocore-1.12 를 직접 사용하지 않게 하고 있습니다.

이 문서는 ALB/CLB 액세스 로그를 Elasticsearch Service(6.0/6.2)로 가져오는 메모의 CloudFront 액세스 로그 버전입니다.

기사에서 참조하는 전 기사, 클래스 메소드의

Ingest Node를 사용하여 AWS Lambda를 통해 Elasticsearch에 ELB 액세스 로그를보다 쉽게 ​​가져옵니다. (Developers.IO)

의 끝에,

Ingest Pipeline 설정을 변경하면 동일한 Lambda 함수에서 CloudFront 액세스 로그를 시각화할 수 있습니다.

라고 적혀 있지만 적어도 Elasticsearch Service 6.0 (이후)에서는 작동하지 않았기 때문에 그 팔로우로 쓰고 있습니다.

※타이틀에 있는 대로, 하는 방법은 완전히 손 빼고입니다…급하게 준비할 필요가 있었으므로…

대응 포인트

베이스는 마지막 기사 대로입니다만, 단순히 Ingest Pipeline 의 설정을 바꾸는 것만으로는, 이하의 점으로 문제가 생깁니다.

로그가 탭 구분(TSV?)이므로, 그대로 Elasticsearch Service의 Ingest Pipeline에 보내면 에러가 나와 화난다.

로그의 선두에 코멘트행이 2행 정도 들어가므로, 거기의 혼잡에서도 에러가 나온다(무시해도 좋지만, 기분 나쁘다…).

날짜의 형식이, Grok 가 디폴트로 가지고 있는 패턴 (DATE_US·DATE_EU 등)에서는 받아들일 수 없다 (정의하면 됩니다만, 정의한 다음에 추가로 시각과 결합해, 타임 존을… 라고 생각한다 가 귀찮은).

그래서 이번에는 거의 Python 코드 측에서 대응해 버립니다 (실제는 Ingest Pipeline 측에서 어떻게든 하는 것이 근육이겠지만…).

Ingest Pipeline 설정

CloudFront의 로그 형식은 여기입니다.

액세스 로그 로그 파일 형식 (AWS Documentation » Amazon CloudFront » 개발자 가이드)

또, Grok 의 설정에 대해서는, 이쪽을 참고로 했습니다.

CloudFront 액세스 로그를 Kibana로 시각화 (Developers.IO)

※Logstash 설정(Advanced)의 로그 메시지의 퍼스 부분입니다.

설정(CloudFront용)

$ curl -H "Content-Type: application/json" -XPUT 'https://XXXXX-XXXXXXXXXXXXXXXXXXXXXXXXXX.ap-northeast-1.es.amazonaws.com/_ingest/pipeline/cflog' -d '{
  "processors": [{
    "grok": {
      "field": "message",
      "patterns":[ "%{TIMESTAMP_ISO8601:timestamp} %{NOTSPACE:x_edge_location} (?:%{INT:sc_bytes:int}|-) %{IPORHOST:c_ip} %{WORD:cs_method} %{HOSTNAME:cs_host} %{NOTSPACE:cs_uri_stem} %{INT:sc_status:int} %{GREEDYDATA:referrer} %{GREEDYDATA:User_Agent} %{GREEDYDATA:cs_uri_query} %{GREEDYDATA:cookies} %{WORD:x_edge_result_type} %{NOTSPACE:x_edge_request_id} %{HOSTNAME:x_host_header} %{URIPROTO:cs_protocol} (?:%{INT:cs_bytes:int}|-) %{NUMBER:time_taken:float} %{NOTSPACE:x_forwarded_for} %{NOTSPACE:ssl_protocol} %{NOTSPACE:ssl_cipher} %{WORD:x_edge_response_result_type}" ],
      "ignore_missing": true
    }
  },{
    "remove":{
      "field": "message"
    }
  }, {
    "user_agent": {
      "field": "User_Agent",
      "target_field": "user_agent",
      "ignore_failure": true
    }
  }, {
    "remove": {
      "field": "User_Agent"
    }
  }]
}'

코드 변경점(ALB/CLB 버전에서)

25-31 줄을 다시 씁니다.

코드 변경 부분

            if not line.strip().startswith('#'):
                data += '{"index":{"_index":"%s","_type":"log"}}\n' % es_index
                data += '{"message":"%s"}\n' % line.strip().replace('"', '\\"').replace('\t', ' ').replace(' ', 'T', 1).replace(' ', 'Z ', 1)

                if len(data) > 3000000:
                    _bulk(data, awsauth)
                    data = ""

코멘트행은 스킵 한다.

탭 구분 기호를 반각 공백 구분 기호로 변환합니다.

날짜와 시간을 TIMESTAMP_ISO8601로 캡처할 수 있는 형식으로 변환합니다(시간대는 UTC).

하고 있습니다.

Lambda 변경점

로그를 가져오는 버킷 외에도 환경 변수 Elasticsearch Service로 가져올 때 INDEX 접두사와 Ingest Pipeline의 이름(아래 그림)을 변경합니다.


이제 CloudFront 로그도 캡처됩니다.

※아마, Elsaticsearch Service 6.2 에서도 같은 순서로 괜찮다고 생각합니다(미검증.근일중에 검증 예정). →04/08 괜찮았습니다.

Elasticsearch Service(6.0/6.2)에 포함된 로그(INDEX)를 Curator에서 삭제하는 메모