엘라스틱서치 RestHighLevelClient를 이용한 로그백 커스텀 appender 만들기

AWS에서 제공하는 ES를 상용애플리케이션 로그시스템으로 활용하고 있는데 최근 코로나 덕분에 재택근무를 하다보니 인프라팀에서 키바나 접속을 제한하면서 재택근무 하는 날 상용서비스에 대한 로그확인이 불가능해졌다. 그래서 키바나에 Cognito서비스를 적용하여 인증을 통해서 키바나 접근이 가능하도록 설정을 하긴 했는데.....로그가 기록이 안되는 문제가 발생했다. 기존에 사용하던 라이브러리는 internetitem/logback-elasticsearch-appender 였는데 기본적인 rest api를 통한 인덱싱만 가능했고 ES의 access policy를 role 기반으로 cognito를 설정하면서 인증문제로 인해 ES로깅을 못하게 된 것이다. 

그럼 이 문제를 어떻게 해결해야 할까 하다가 내린 결론은 "Elasticsearch에서 제공하는 RestHighLevelClient를 이용해서 logback appender를 직접 만들어서 써보지 뭐~" 였다.

우선 RestHighLevelClient를 사용할 수 있으려면 아래 라이브러리들을 추가해야한다. 

compile 'org.elasticsearch:elasticsearch:7.7.1' 
compile 'org.elasticsearch.client:elasticsearch-rest-high-level-client:7.7.1' 
compile 'com.github.awslabs:aws-request-signing-apache-interceptor:deb7941e85' 
compile 'org.elasticsearch:elasticsearch-x-content:7.7.1'

이것도 공식 엘라스틱서치 문서가 제대로 업데이트 되어있지 않거나 불명확해서 런타임시에 자꾸 에러나서 구글링 열심히 해서 찾아냈다

자, 이제 AWS Elasticsearch에 access policy를 어떻게 설정했는지 한번 보자.

{
   "Version":"2016-10-17",
   "Statement":[
       {"Effect":"Allow",
         "Principal":{
         "AWS":"arn:aws:iam::77345249:role/my-iam-role"},
         "Action":"es:*",
         "Resource":"arn:aws:es:ap-northeast-2:77345249:domain/my-application/*"},
      {"Effect":"Allow",
         "Principal":{
            "AWS":"arn:aws:sts::77345249:assumed-role/Cognito_Role/CognitoIdentityCredentials"},
         "Action":"es:*",
         "Resource":"arn:aws:es:ap-northeast-2:77345249:domain/my-application/*"  }
    ]
}

위 처럼 설정을 하면 기존에 잘 동작하던 ElasticsearchAppender가 로그를 제대로 남기지 못하게 된다. (my-iam-role을 기반으로 인증을 해야 정상적으로 로깅을 할 수 있다) 그래서 internetitem/logback-elasticsearch-appender 소스를 좀 까 보았더니 나름 AWSAuthentication이라는 인증 관련 클래스를 가지고 있었고 logback 설정으로 BasicAuthentication 대신에 해당 인증 방식을 사용할 수 있을 것 같아서 logback 설정에 AWSAuthentication을 설정해주었지만 region을 제대로 가져오지 못하는 버그와 함께 CredentialChain을 Default어쩌구 체인을 사용하고있었고 role기반 인증을 하도록 설정을 할 수가 없었다. 어떻게든 기존 라이브러리를 사용해보려고 해당 인증클래스를 상속해서 별도의 인증 클래스를 작성한 뒤 해당 인증서비스를 사용할 수 있도록 해보려고 시도해보았으나 결국 signed request를 날리지 못하고 인증오류로 ES에서 400 에러만 주구장창 받았다.

그래서 RestHighLevelClient를 이용해서 인덱싱부터 해보고 그것을 기반으로 logback custom appender를 만들어 보기로 했다. 우선 인덱싱하는 것은 엘라스틱서치 문서에 친절하게 나와있긴 하였으나 예제가 최신화가 안되어있어 라이브러리 추가하는데만도 시간을 낭비했다. 그리고 코드를 작성하여 ES에 요청하는 순간!!! cognito가 적용되어있어 인증 실패!! 그래서 위 ES access policy statement의 마지막 항목으로 아래와 같이 정책을 추가해서 임시로 허용해준다.

,
    {
      "Effect": "Allow",
      "Principal": {
        "AWS": "*"
      },
      "Action": "es:*",
      "Resource": "arn:aws:es:ap-northeast-2:72345249:domain/my-application/*"
    }

해당 도메인에서 ES로 들어오는 모든 요청을 허용해준다는 의미다. 

여기까지 성공했는데 인증을 도대체 어떤 credential을 이용해야할지 너무 난감했다. 기존에 경력 20년가까이 된 개발자가 작성해놓은 코드가 정상적으로 돌아가지 않았기 때문에 처음부터 하는 심정으로 구글링을 하기 시작했다. 동시에 aws sdk의 인증관련 클래스들을 하나하나 보기 시작했다. 그러다가 눈에 들어온 녀석. 바로 STSAssumeRoleSessionCredentialsProvider 이다. 이 녀석을 이용해서 아래와 같이 credential을 받아오는 코드를 작성하고 해당 credential을 이용해서 인증에 성공했다. 

  private AWSCredentialsProvider awsS3CredentialsProvider() {
      List<AWSCredentialsProvider> credentialsProviders = new ArrayList<>();

      String roleSessionName = "role-session-name-" + System.currentTimeMillis();

      credentialsProviders.add(new STSAssumeRoleSessionCredentialsProvider.Builder(roleArn, roleSessionName)
              .withStsClient(AWSSecurityTokenServiceClientBuilder.defaultClient())
              .build());

      return new AWSCredentialsProviderChain(credentialsProviders);
  }

여기서 roleArn 정보만 있으면 나머지는 aws sdk에서 알아서 정보를 가지고 와서 잘 처리해준다. 유후~~ 인증에 성공했으니 위에서 access policy에 마지막에 추가했던 항목을 다시 지워도 된다.

하지만 위 코드를 얻기까지 아래 오류메시지를 수도 없이 보았다.

Failed to send events to Elasticsearch: Got response code [403] from server with data {"message":"The request signature we calculated does not match the signature you provided. Check your AWS Secret Access Key and signing method.

Failed to send events to Elasticsearch: Got response code [403] from server with data {"message":"The security token included in the request is expired"}

그리고 ES에 max 샤드 개수가 꽉차서 로깅시 신규 인덱스를 만들 수 없는 상황도 있었는데 이때는 아래와 같은 오류 메시지가 떴다.

Suppressed: org.elasticsearch.client.ResponseException: method [PUT], host [https://vpc-my-application-ig5szafi26tsfulustdplosdu3f4.ap-northeast-2.es.amazonaws.com], URI [/index-2020-06-20/_doc/4ba14a8b96b242d0?timeout=1m], status line [HTTP/1.1 400 Bad Request] {"error":{"root_cause":[{"type":"index_creation_exception","reason":"failed to create index [api-2020-06-20]"}],"type":"validation_exception","reason":"Validation Failed: 1: this action would add [10] total shards, but this cluster currently has [1002]/[1000] maximum shards open;"},"status":400}

키바나에서 직접 PUT myindex-2020-06-20 명령으로 시도해도 실패했고, 위 에러메시지와 동일한 내용을 확인할 수 있었다. 위와 같은 에러메시지가 발생하면 max 샤드개수를 올려주던지 아니면 기존에 있는 인덱스들 중에서 불필요한 것들을 좀 날려주면 된다. 일부 인덱스를 삭제하니 로깅이 잘 되는 것을 확인할 수 있었다.

하지만 뭔가 이상했다. 테스트 API를 호출하면 로그가 3개가 찍혀야되는데 1개만 찍히고 있었다. 도대체 뭘까 어디가 문제일까 한참을 고민하고 이것저것 수정하면서 테스트를 했다.

처음 로그메시지 인덱싱을 위한 테스트코드는 아래처럼 작성했었다.

  IndexRequest request = new IndexRequest(indexPrefix + LocalDate.now().toString())
          .id(UuidUtil.getTrackingId())
          .source(document);

  Runnable r = () -> {
      try (RestHighLevelClient esClient = esClient()) {
          esClient.index(request, RequestOptions.DEFAULT);
      } catch (Exception e) {
          log.error("ES logging ERROR!!! e-msg:{}", e.getMessage(), e);
      }
  };
  r.run();

저렇게 작성하니 아무 문제없이 로깅을 했으나 로그가 3개중 1개만 기록이 되고있었다. 그래서 IndexRequest의 create설정(default:false)을 true로 추가해보았다. 왠지 덮어쓰는 것 같은 느낌이 들었기 때문이다. 그랬더니 아래 "오류 1" 메시지가 출력이 되었고, version 설정을 했더니 오류 2가 출력이 되었다. 

// 오류 1 : create(true) 추가했더니
status:CONFLICT, e-msg:Elasticsearch exception [type=version_conflict_engine_exception, reason=[bf0af4cfa5b64945]: version conflict, document already exists (current version [1])]

// 오류 2 : version 설정을 해줬더니
create operations do not support explicit versions. use index instead

// 오류 3 : create 설정을 다시 뺐더니
internal versioning can not be used for optimistic concurrency control. Please use `if_seq_no` and `if_primary_term` instead

// 오류 4 : ifSeqNo만 설정했더니
ifSeqNo is set, but primary term is [0];

// 오류 5 : ifPrimaryTerm도 설정했더니
version conflict, required seqNo [329563504204790], primary term [1592662218337]. but no document was found]

하나 둘 오류 메시지를 확인하면서 코드를 수정하고 다시 테스트 하기를 몇 번 하고나니 동일한 버전으로 여러 도큐먼트를 인덱싱하려다가 오류가 발생(versioning문제)하는 것이라 추측할 수 있었고, IndexRequest에 세팅하는 값에 version이 있어서 그것 때문일 거라 생각했었으나 문제는 id였다. -_-;; 우선, 나의 목적은 모든 로그를 남기려는 거였고 도큐먼트를 버저닝해가면서 업데이트 할 일은 없는데 왜 자꾸 버저닝 관련 오류가 나는 것일까? 고민을 좀 하다가 발견한 사실! id값을 특정 프로세스를 트래킹하기위한 똑같은 uuid값을 넣고있었던 것이다. 그래서 id에 세팅하는 값을 아래처럼 나노시간으로 바꾸어주었다.

IndexRequest request = new IndexRequest(indexPrefix + LocalDate.now().toString())
                .id(String.valueOf(System.nanoTime()))
                .source(document);

그랬더니 더이상 오류도 나지 않고 인증도 잘하고 정상적으로 cognito 설정을 완료할 수 있었다.

하지만 실제 상용서비스에서 사용함에 있어서 버그가 없는지 또는 성능에 문제가 없을지는 부하테스트를 좀 해봐야겠다.

황금같은 주말에 아가랑 놀아주지도 못하고 새벽까지 이러고 있다니 ㅜㅜ

그래도 해결의 실마리를 찾아서 기쁘다. 씻고 자야겠다...

[6월 24일 업데이트]

부하테스트 결과 try with resource 구문형태로 index 메서드를 사용할 경우 BAD_REQUEST Unable to parse response body 와 같은 오류나 아니면 "You have exceeded the number of permissible concurrent requests with unique IAM Identities. Please retry." 와 같은 오류를 보게될 수 있다. 따라서 상용로그처럼 1초내에 수백~수천 건의 로그를 쌓아야 한다면 IndexRequest대신 BulkRequest를 bulk메서드로 호출하도록 하고, try with resource 구문이 아닌 client를 한번만 생성한 다음 계속 재사용하도록 해야한다. 관련된 내용은 아래 문서에서 확인할 수 있다.

docs.aws.amazon.com/IAM/latest/UserGuide/reference_iam-limits.html

www.elastic.co/guide/en/elasticsearch/client/java-rest/current/java-rest-high-getting-started-initialization.html

[6월 30일 업데이트]

BulkRequest도 bulk 메서드를 사용할 경우 synchronous하게 동작하기 때문에 로깅하는 것이 기존 애플리케이션 로직의 소요시간에 영향을 주게 된다. 따라서 로깅을 매우 빈번하게 해야하는 경우에는 bulkAsync 메서드를 사용해야 한다.

테스트 결과 bulk 메서드는 bulkAsync 메서드보다 2배의 성능 저하 현상을 가져온다. 테스트 시 ES는 AWS ES 7.1 버전으로 테스트했다. 라이브러리도 페이지 첫 부분에 명시했듯이 7.7.1 버전을 사용했다.

[7월 7일 업데이트]

세 가지 이슈가 추가로 발견되었다.

첫 번째, 6월 24일 업데이트에 나왔던 문제가 또 나왔다. 

method [POST], host [https://vpc-elasticsearch.ap-northeast-2.es.amazonaws.com], URI [/_bulk?timeout=1m], status line [HTTP/1.1 400 Bad Request]
{"Message":"You have exceeded the number of permissible concurrent requests with unique IAM Identities. Please retry."}

이번에는 소스레벨에서 많은 클라이언트를 사용한게 아니었지만, AWS 콘솔에 접속하여 사용하는 사용자가 많을 경우 종종 발생한다. 이 부분을 어떻게 해결할 수 있을까 고민을 해보아도 이 부분까지 컨트롤 할 수는 없을 것 같다는 생각이 든다.

두 번째, 의도치 않게 클라이언트가 종료되는 현상이 개발서버에서 발견되었다.

IllegalStateExceptionRequest cannot be executed; I/O reactor status: STOPPED

이건 예외를 잡아서 새로운 클라이언트를 만들어 사용하도록 수정하긴 했으나....왜 클라이언트가 종료되었는지 알 수가 없어 답답하다.

세 번째, 부하를 많이 준 인스턴스에서 아래와 같은 오류 발생 (thread: I/O dispatcher 2)

Connection lease request time out:
at org.apache.http.nio.pool.AbstractNIOConnPool.processPendingRequest(AbstractNIOConnPool.java:411)
at org.apache.http.nio.pool.AbstractNIOConnPool.processNextPendingRequest(AbstractNIOConnPool.java:391)
at org.apache.http.nio.pool.AbstractNIOConnPool.release(AbstractNIOConnPool.java:355)
at org.apache.http.impl.nio.conn.PoolingNHttpClientConnectionManager.releaseConnection(PoolingNHttpClientConnectionManager.java:364)
at org.apache.http.impl.nio.client.AbstractClientExchangeHandler.releaseConnection(AbstractClientExchangeHandler.java:245)
at org.apache.http.impl.nio.client.MainClientExec.responseCompleted(MainClientExec.java:387)
at org.apache.http.impl.nio.client.DefaultClientExchangeHandlerImpl.responseCompleted(DefaultClientExchangeHandlerImpl.java:172)
at org.apache.http.nio.protocol.HttpAsyncRequestExecutor.processResponse(HttpAsyncRequestExecutor.java:448)
at org.apache.http.nio.protocol.HttpAsyncRequestExecutor.inputReady(HttpAsyncRequestExecutor.java:338)
at org.apache.http.impl.nio.client.InternalRequestExecutor.inputReady(InternalRequestExecutor.java:83)
at org.apache.http.impl.nio.DefaultNHttpClientConnection.consumeInput(DefaultNHttpClientConnection.java:265)
at org.apache.http.impl.nio.client.InternalIODispatch.onInputReady(InternalIODispatch.java:81)
at org.apache.http.impl.nio.client.InternalIODispatch.onInputReady(InternalIODispatch.java:39)
at org.apache.http.impl.nio.reactor.AbstractIODispatch.inputReady(AbstractIODispatch.java:121)
at org.apache.http.impl.nio.reactor.BaseIOReactor.readable(BaseIOReactor.java:162)
at org.apache.http.impl.nio.reactor.AbstractIOReactor.processEvent(AbstractIOReactor.java:337)
at org.apache.http.impl.nio.reactor.AbstractIOReactor.processEvents(AbstractIOReactor.java:315)
at org.apache.http.impl.nio.reactor.AbstractIOReactor.execute(AbstractIOReactor.java:276)
at org.apache.http.impl.nio.reactor.BaseIOReactor.execute(BaseIOReactor.java:104)
at org.apache.http.impl.nio.reactor.AbstractMultiworkerIOReactor$Worker.run(AbstractMultiworkerIOReactor.java:591)
at java.lang.Thread.run(Thread.java:748)

참고자료:

https://www.elastic.co/guide/en/elasticsearch/client/java-rest/master/java-rest-high-document-index.html http://logback.qos.ch/manual/appenders.html