Firehose부터 OpenSearch까지: AWS 서비스를 활용한 로그 분리 전략

안녕하세요 펫프렌즈 백엔드 개발자 팀버(Timbre/박기오) 입니다 :)

애플리케이션을 개발하고 운영하면서 로그는 장애 발생 시 신속한 대응을 가능하게 해주는 중요한 요소 중 하나 입니다. 하지만 여러 서비스에서 로그를 무제한으로 저장하는 것은 비용적으로 부담이 크기 때문에, 일반적으로 ISM(Index State Management) 정책을 적용해 일정 기간이 지난 로그를 자동으로 삭제하는 방식을 사용합니다.

펫프렌즈 에서는 AWS OpenSearch 를 활용하여 로그를 저장하고 있으며, 정책상 2주 이상 지난 인덱스 데이터는 자동으로 삭제 됩니다. (State Management Policy : delete_old_indexes_14d)

그런데 운영을 하다보니 동일한 애플리케이션 내에서도 일부 특정 케이스에 대해서는 더 긴 기간 동안 로그를 보관할 필요가 있는 상황이 발생했습니다. 요구사항에 따라 해당 로그를 기존 인덱스에서 분리하여 별도의 인덱스로 저장하는 작업을 진행하게 되었습니다.

1. 애플리케이션에서 로그 분리

1️⃣ MDC(Mapped Diagnostic Context)를 활용한 로그 필터링

🔹 Dependency 추가

로그백(Logback)의 조건부처리(if 등) 기능을 사용하기 위해서는 Janino 라이브러리를 추가 해야합니다. build.gradle 에 다음 의존성을 추가합니다.

implementation 'org.codehaus.janino:janino:3.0.12'

🔹 Filter 적용

특정 케이스의 로그를 분리하기 위해서, API 호출 시 Header에 isSeparation 값을 포함하여 전달합니다.

이 값이 존재할 경우 MDC 에 저장합니다.

@Override
protected void doFilterInternal(HttpServletRequest request, HttpServletResponse response, FilterChain filterChain) throws ServletException, IOException {
  if (!StringUtils.isEmpty(request.getHeader("isSeparation"))) {
            MDC.put("isSeparation", request.getHeader("isSeparation"));
        }

    filterChain.doFilter(request, response);
    MDC.clear();
}

MDC 란?

• 현재 실행중인 Thread 에 메타정보를 넣고 관리하는 공간입니다.
• Map(Key, Value) 형태로 값을 저장할 수 있습니다.
• 메타정보를 Thread 별로 관리하기 위해 내부적으로 ThreadLocal을 사용하고 있습니다.

MDC 사용 시 주의할 점

MDC는 Thread 별로 저장되는 ThreadLocal을 사용하므로, 요청이 완료될때 clear() 를 호출 해야 합니다.

그렇지 않으면 동일한 Thread가 재사용될 때 이전 데이터가 남아 있을 수 있으므로, doFilter()가 끝나면 MDC.clear() 를 호출해야 합니다.

🔹 Logback 설정

0. logback.xml에서 log Appender 적용

<springProfile name="develop | staging">
    <root level="info">
        <appender-ref ref="logstash" />
        <appender-ref ref="separation_logstash" />
    </root>
</springProfile>

1. logstash Appender

<appender name="logstash" class="ch.qos.logback.core.ConsoleAppender">
    <filter class="ch.qos.logback.core.filter.EvaluatorFilter">
        <evaluator>
            <expression>
                return mdc.containsKey("isSeparation");
            </expression>
        </evaluator>
        <OnMatch>DENY</OnMatch>
        <OnMismatch>ACCEPT</OnMismatch>
    </filter>

    <encoder class="net.logstash.logback.encoder.LogstashEncoder">
        <timeZone>UTC</timeZone>
    </encoder>
</appender>

📌 동작 방식

• logstash Appender는 기본 로그를 처리합니다.
• EvaluatorFilter를 사용하여 mdc.containsKey("isSeparation") 조건을 검사합니다.
• MDC에 isSeparation 키가 존재하면(OnMatch), 로그를 거부(DENY)
• MDC에 isSeparation 키가 없으면(OnMismatch), 로그를 허용(ACCEPT)
• isSeparation 값이 설정되지 않은 일반 로그들은 logstash Appender로 출력됩니다.

2. separation_logstash Appender

<appender name="separation_logstash" class="ch.qos.logback.core.ConsoleAppender">
    <filter class="ch.qos.logback.core.filter.EvaluatorFilter">
        <evaluator>
            <expression>
                return mdc.containsKey("isSeparation");
            </expression>
        </evaluator>
        <OnMatch>ACCEPT</OnMatch>
        <OnMismatch>DENY</OnMismatch>
    </filter>

    <encoder class="net.logstash.logback.encoder.LogstashEncoder">
        <timeZone>UTC</timeZone>
    </encoder>
</appender>

📌 동작 방식

• separation_logstash Appender는 특정 조건을 만족하는 로그만 처리합니다.
• EvaluatorFilter를 사용하여 mdc.containsKey("isSeparation") 조건을 검사합니다.
• MDC에 isSeparation 키가 존재하면(OnMatch), 로그를 허용(ACCEPT)
• MDC에 isSeparation 키가 없으면(OnMismatch), 로그를 거부(DENY)
• isSeparation 값이 설정된 특정 로그만 separation_logstash Appender로 출력됩니다.

💡 JaninoEventEvaluator

<evaluator> 의 기본값입니다.

Java 언어의 Boolean 표현식을 사용하여 true/false 값을 반환합니다.

2️⃣ Service Layer에서 별도 Logger 사용 ✅

이번 방식에서는 Service Layer에서 특정 Logger를 직접 호출하여 로그를 분리하는 방법을 사용합니다.

🔹 Logback 설정

📌 logback.xml에서 새로운 Logger 추가

<logger name="separation" additivity="false" level="info">
      <springProfile name="develop | staging">
          <appender-ref ref="logstash" />
      </springProfile>

    <springProfile name="production">
          <appender-ref ref="logstash" />
          <appender-ref ref="sentry" />
      </springProfile>
</logger>

📌 동작 방식

1)separation이라는 새로운 Logger를 정의 합니다.

2)additivity="false"

• 부모 Logger(기본 root Logger)로 로그가 전파되지 않도록 설정합니다.

🔹 Service Layer에서 Logger 호출

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class Service {
   private final Logger separationLogger = LoggerFactory.getLogger("separation");

   public void separationLog() {
    ...(로직)
    separationLogger.info("{}", objectMapper.convertValue(requestDto, Map.class));
   }
}

1)separation이라는 별도의 Logger 인스턴스를 생성 합니다.

2)separationLogger.info(...)를 호출하면 logback 설정에서 정의한 separation Logger 설정을 따릅니다.

3)특정 요청(requestDto)을 JSON 형태로 변환하여 로그로 저장 합니다.

2. 데이터 파이프라인을 활용한 로그 분리

1️⃣ OpenSearch Transform Jobs 활용

가장 먼저 OpenSearch Transform Jobs 에서 제공하는 기능을 사용해보기로 하였습니다. Transform Jobs 을 사용하면 기존 인덱스의 데이터를 특정 필드를 기준으로 그룹화하여 새로운 인덱스로 변환할 수 있습니다.

🔹 설정 및 생성 과정

1. OpenSearch 접속 및 설정

• OpenSearch에 접속한 후, 좌측 탭에서 OpenSearch Plugins > Index Management 로 이동

• Transform Jobs > Create Transform Job을 선택

2. Transform Job 생성

Step 1: Job Name과 Description 입력

새로운 Transform Job을 생성할 때, Job 이름과 설명을 입력합니다.

Step 2: Source Index 및 필터 설정

• Indices에서 변환할 원본 인덱스 선택
• Source index filter를 설정하여 특정 조건의 데이터만 변환 (Query DSL 사용)

{
    "match" : {
        "url" : {
            "query" : "bookmark"
        }
    }
}

• Target index 를 지정 (새로운 인덱스를 생성하는 목적이기 때문에 기존 인덱스가 아닌 새 인덱스명을 입력)

3. Transform 방식 설정

Transform Jobs의 기본 동작 방식은 마이그레이션이 아닌, 특정 필드를 기준으로 그룹화해서 새로운 집계를 생성하는 기능입니다.

🚨 문제점 — 기존 데이터가 집계되는 방식

• traceId 같은 고유한 값이 포함된 데이터라도, 여러 개의 로그가 있으면 count 기반 집계가 수행됩니다.
• 개별 로그가 새로운 인덱스로 마이그레이션되지 않고, 집계된 데이터 형태로 변환됩니다.

✅ 해결 방법 — 다중 그룹핑 적용

하지만 다중 그룹핑이 가능하기 때문에 traceId와 timestamp를 함께 집계하면, 마치 기존 로그 데이터를 마이그레이션한 것처럼 변환할 수 있습니다.

4. Schedule Interval 설정

Transform Job이 일정 주기로 실행되도록 설정합니다. (예: 30일마다 실행)

🔹 Transform Jobs의 한계

기존 데이터가 새로운 인덱스로 누적되지 않음

• Transform Jobs는 기존 데이터를 새로운 인덱스로 이동하는 방식이 아니라, 전체 데이터를 덮어쓰는 방식으로 동작합니다.
• 이전에 이동한 데이터 이후의 offset부터 가져와서 append하지 않습니다.

기존 2주 보관 정책을 해결할 수 없음

• 결국, Transform Jobs를 사용해도 기존과 동일하게 2주의 데이터만 보관 가능합니다.
• 로그 데이터를 장기간 보관하는 목적보다는, 기존 데이터를 기반으로 새로운 집계를 생성하는 용도로 더 적합하다고 판단하였습니다.

2️⃣ Fluent Bit 설정을 활용한 로그 분리

🔹 Fluent Bit 설정을 커스텀하여 라우팅

기존 Fluent Bit 설정을 변경하여 로그를 원하는 목적지로 분리하였습니다.

[FILTER]
    Name parser
    Match *
    Key_Name log
    Parser json
    Reserve_Data True

[FILTER]
    Name grep
    Match *
    Tag backend-connect
    Exclude isSeparation true
[FILTER]
    Name grep
    Match *
    Tag separation
    regex isSeparation: ^true$

[OUTPUT]
    Name firehose
   Match backend-connect
   region = ap-northeast-2
   delivery_stream = backend-connect-log
[OUTPUT]
   Name firehose
   Match separation
   region = ap-northeast-2
   delivery_stream = backend-connect-separation-log

커스텀 Fluent Bit 설정 설명

1. JSON 로그 파싱 (Parser Filter)

[FILTER]
    Name parser
    Match *
    Key_Name log
    Parser json
    Reserve_Data True

1) 로그 데이터를 JSON 형태로 파싱합니다.

2)log 필드를 JSON으로 변환하여 개별 필드로 사용 가능하게 합니다.

3)Reserve_Data True를 설정하면 원본 로그 데이터를 유지합니다.

2. 특정 조건에 따라 로그 필터링 (Grep Filter)

“backend-connect” 태그를 가진 로그 필터링

# Filter out logs to Firehose main
[FILTER]
    Name grep
    Match *
    Tag backend-connect
    Exclude isSeparation true

1)isSeparation 필드가 존재하면 해당 로그는 제외

2)결과적으로 isSeparation이 없는 로그만 Firehose로 전송됩니다.

“separation” 태그를 가진 로그 필터링

# Filter out logs to Firehose sub
[FILTER]
    Name grep
    Match *
    Tag separation
    regex isSeparation: ^true$

1)isSeparation 필드가 true인 경우에만 로그를 포함합니다.

• 정규표현식을 이용해 해당 필드를 포함하는 로그만 전달

2)isSeparation: true 값을 가진 로그만 Firehose로 전송됩니다.

3. Firehose로 로그 전송 (Output 설정)

backend-connect 로그 전송

# terraform logConfiguration 설정을 OUTPUT 설정으로 대체가능
[OUTPUT]
    Name firehose
    Match backend-connect
    region ap-northeast-2
    delivery_stream backend-connect-log

1)backend-connect 태그를 가진 로그를 AWS Kinesis Firehose의 backend-connect-log 스트림으로 전송합니다.

2)region ap-northeast-2는 서울 리전을 의미합니다.

separation 로그 전송

[OUTPUT]
    Name firehose
    Match separation
    region ap-northeast-2
    delivery_stream backend-connect-separation-log

1)separation 태그를 가진 로그를 AWS Kinesis Firehose의 backend-connect-separation-log 스트림으로 전송합니다.

2)isSeparation: true 필드가 있는 로그들만 이 스트림으로 전달됩니다.

🔹 Terraform 설정: S3에 업로드된 Fluent Bit 설정을 활용

Fluent Bit 설정을 S3에 업로드하고, 필요한 컨테이너가 해당 설정을 자동으로 가져와 사용할 수 있도록 구성하여 컨테이너 재배포 없이도 설정 변경이 가능하도록 개선하였습니다.

container_definitions_with_firelens = [
    {    
      name      = "log_router"
      image     = "public.ecr.aws/aws-observability/aws-for-fluent-bit:stable"
      essential = true
      firelensConfiguration = {
        type = "fluentbit"
        options = {
          config-file-type  = "s3",
          config-file-value = "arn:aws:s3:::{directory}/parse-json.conf"
        }
      }
    }
  ]

변경된 앱 컨테이너 설정 (log router 를 통해 데이터가 전달될 firehose 설정)

• fluent bit 설정을 따르기위해 컨테이너에서는 별도의 로그 설정을 하지 않고, Fluent Bit가 로그 라우팅을 전적으로 담당하도록 변경했습니다.

app_container_definition = {
    logConfiguration = null
}

3️⃣ Lambda 함수를 이용해 분리된 Firehose 로 할당 (Terraform을 이용한) ✅

🔹 Process

1. Lambda 함수를 생성하고 ZIP 파일로 압축하여 S3에 업로드
2. S3에 업로드된 함수를 이용해 AWS Lambda 생성

• Lambda 실행 시 필요한 2개의 Firehose Delivery Stream Name을 환경변수로 설정

3. Main Firehose 및 Sub Firehose 생성

Main Firehose

• 데이터를 OpenSearch로 전송하는 Configuration 설정 시 processing_configuration 옵션을 추가
• 해당 옵션에서 AWS Lambda 함수의 ARN을 지정하여 데이터 처리 적용

Sub Firehose

• processing_configuration 없이 생성 (Lambda를 통한 전처리 없이 바로 저장)

🔹 Terraform 설정: S3에 업로드된 함수를 가져와 Lambda 를 생성

1. Firehose 생성 및 Lambda 함수 연결

Terraform을 사용하여 Lambda 함수와 Firehose 를 자동으로 생성하는 모듈을 구성합니다.

module "logging" {
  source                = "../../logging/sub_firelens_essearch"
  count                 = var.use_cloudwatch_log ? 0 : 1
  stream_name           = local.firehose_stream_name
  sub_stream_name       = local.sub_firehose_stream_name
  essearch_domain       = var.essearch_domain
  index_name            = "${var.app_name}-service"
  sub_index_name        = "${var.index_name}-service"
  env                   = terraform.workspace
  index_rotation_period = var.index_rotation_period
  # 람다 함수가 업로드되어있는 S3 버킷명
  lambda_s3_bucket      = var.lambda_s3_bucket
 # ${var.index_name} = zip 파일이 있는 폴더명 (separation)
 # ${var.lambda_function_name} = S3 업로드된 zip 파일명
  lambda_s3_key         = "${var.index_name}/${var.lambda_function_name}"
}

2. IAM Role 설정

Lambda와 Firehose가 서로간의 통신에 필요한 적절한 IAM Role을 지정합니다.

variable "firehose_lambda_role" {
  default     = "connectServiceFirehoseRole"
  description = "Firehose lambda service 역할"
}

3. Lambda 함수 생성 (S3에서 ZIP 파일 가져오기)

• S3에 업로드된 Lambda ZIP 파일을 가져와 Lambda 함수를 생성합니다.
• 환경 변수로 Main/Sub Firehose 스트림 이름을 설정하여 Firehose 를 분리 합니다.

resource "aws_lambda_function" "firehose_lambda" {
  function_name     = var.lambda_function
  role              = data.aws_iam_role.firehose_lambda.arn
  # 람다 함수 생성에 사용된 파일명과 메소드명이 일치해야 합니다.
  # index = index.py, lambda_handler = def lambda_hanlder():
  handler           = "index.lambda_handler"
  # 가장 최신 버전 사용
  runtime           = "python3.9"
 # 위 S3 버킷정보를 가져오는 data 의 output 값을 이용합니다
  # aws_s3_bucket_object 에서 version_id 를 지정하지 않으면 항상 최신 버전을 가지고 옵니다
  s3_bucket         = data.aws_s3_bucket_object.lambda_function_zip.bucket
  s3_key            = data.aws_s3_bucket_object.lambda_function_zip.key
  s3_object_version = data.aws_s3_bucket_object.lambda_function_zip.version_id

  environment {
    variables = {
   # main, sub firehose delivery stream name 을 람다함수에서 사용하기 위한 환경변수 설정입니다
      main_firehose_name = var.stream_name
      sub_firehose_name  = var.sub_stream_name
    }
  }
}

4. Main Firehose 생성과 Lambda 연결

• Firehose가 데이터를 OpenSearch로 전송하기 전에 Lambda를 활용하여 데이터를 필터링 하도록 설정합니다.

resource "aws_kinesis_firehose_delivery_stream" "log_stream" {
  name        = var.stream_name
  destination = "elasticsearch"

  s3_configuration {
    role_arn        = data.aws_iam_role.log_stream.arn
    bucket_arn      = "arn:aws:s3:::${local.firehose_s3_bucket}"
  }

  elasticsearch_configuration {
    domain_arn            = data.aws_elasticsearch_domain.log_cluster.arn
    role_arn              = data.aws_iam_role.log_stream.arn                                                                                                                                                                                                                                                                   
    index_name            = var.index_name
    index_rotation_period = var.index_rotation_period
  # 해당 옵션을 사용하여 firehose 에서 opensearch 로 로그를 전달하기전 분리를위한 lambda 함수를 연결합니다
    processing_configuration {
      enabled = "true"
   # firehose 와 연결할 프로세서 타입(Lambda)을 지정하고
   # parameter 값으로 firehose 가 lambda 를 사용할 수 있는 권한을 부여합니다
      processors {
        type = "Lambda"
        parameters {
          parameter_name  = "LambdaArn"
          parameter_value = "${aws_lambda_function.firehose_lambda.arn}:$LATEST"
        }
      }
    }
  }
}

5. Sub Firehose 생성

• Main Firehose에서 걸러진 데이터가 별다른 처리 없이 OpenSearch로 전달됩니다.

resource "aws_kinesis_firehose_delivery_stream" "sub_log_stream" {
 # main firehose 에서 전달받은 로그는 모두 sub firehose 에서 처리해야하기 때문에,
 # 별다른 설정없이 opensearch로 보냅니다
  name        = var.sub_stream_name
  destination = "elasticsearch"

  s3_configuration {
    role_arn        = data.aws_iam_role.log_stream.arn
    bucket_arn      = "arn:aws:s3:::${local.firehose_s3_bucket}"
  }
  elasticsearch_configuration {
    domain_arn            = data.aws_elasticsearch_domain.log_cluster.arn
    role_arn              = data.aws_iam_role.log_stream.arn                                                                                                                                                                                                                                                                   
    index_name            = var.sub_index_name
    index_rotation_period = var.index_rotation_period
}

• S3에 업로드된 Lambda ZIP 파일을 가져와 Lambda 함수를 생성합니다.
• 환경 변수로 Main/Sub Firehose 스트림 이름을 설정하여 Firehose 데이터 분리를 수행합니다.

🔹 Lambda 를 생성할 함수 (로그 필터링)

아래 Lambda 함수는 Firehose에서 전달된 로그를 검사하여 isSeparation 여부에 따라 특정 Firehose로 데이터를 분리하는 역할을 합니다.

import os
import json
import base64
import boto3

firehose_client = boto3.client('firehose')
def lambda_handler(event, context):
  # delivery stream 생성시 환경변수로 저장했던 main, sub stream name 추출
    main_firehose_name = os.environ.get('main_firehose_name')
    sub_firehose_name = os.environ.get('sub_firehose_name')
    output_records = []
    for record in event['records']:
        payload = base64.b64decode(record['data']).decode('utf-8')
        data = json.loads(payload)
        if 'isSeparation' not in data and event['deliveryStreamArn'].endswith(main_firehose_name):
            # isSeparation 컬럼이 존재하지 않고 main Firehose 일 경우 -> main index (opensearch)로 전송
            output_records.append({
                'recordId': record['recordId'],
                'result': 'Ok',
                'data': record['data']
            })
        elif 'isSeparation' in data and event['deliveryStreamArn'].endswith(main_firehose_name):
            # isSeparation 컬럼이 존재하지만 main Firehose 일 경우 -> sub Firehose 스트림으로 데이터 전송
            firehose_client.put_record(
                DeliveryStreamName=sub_firehose_name,
                Record={
                    'Data': json.dumps(data).encode('utf-8')
                }
            )
        else:
            output_records.append({
                'recordId': record['recordId'],
                'result': 'Dropped',
                'data': record['data']
            })

    return {'records': output_records}

3. Infra (AWS)

1️⃣ IAM

Lambda와 Firehose 간의 상호 작용을 위해 필요한 IAM 역할 및 정책을 설정합니다.

🔹 connectServiceFirehoseRole (Lambda 실행 역할 및 권한 설정)

1. Lambda 함수 실행 역할 신뢰 정책 (AssumeRole)

{
  "Version": "2012-10-17",
  "Statement": [
    {
       "Effect": "Allow"
       "Principals": {
           "Type": "Service"
           "Identifiers": ["lambda.amazonaws.com"]
        }
       "Actions": ["sts:AssumeRole"]
    }
   ]
}

2. Lambda가 Firehose로 레코드 전송 가능하도록 권한 부여

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": "firehose:PutRecord",
            "Resource": "{arn-prefix}:deliverystream/backend-connect-*"
        }
    ]
}

3. 람다 함수 로그를 확인하기 위한 관리형 정책

• AWSLambdaBasicExecutionRole

🔹 KinesisFirehoseServiceRole-common-ap-northeast-2 (Firehose에서 Lambda 호출 권한 설정)

1. Firehose가 Lambda 함수를 호출할 수 있도록 권한 부여

{
    "Version": "2012-10-17",
    "Statement": [
        {
      "Effect": "Allow", 
      "Action": [ 
            "lambda:InvokeFunction", 
            "lambda:GetFunctionConfiguration" 
       ], 
       "Resource": "{arn-prefix}:function:*"
        }
    ]
}

2️⃣ Architecture

🔹 데이터 흐름

1. 데이터 수집

Main Firehose가 데이터를 수집합니다.

2. Processing Configuration (Lambda 사용)

Firehose의 processing_configuration에 설정된 AWS Lambda가 호출됩니다.

Lambda 함수에서 데이터를 변환 및 전달 할 수 있습니다.

3. Lambda가 isSeparation 값을 확인하여 분기 처리

isSeparation == true

• Lambda에서 처리된 데이터가 다시 Main Firehose로 반환됩니다.

isSeparation == false

• Sub Firehose로 전송합니다.

4. Elasticsearch (OpenSearch)로 전송

Firehose가 최종적으로 Amazon OpenSearch로 데이터를 전송합니다.

마무리

이번 글에서는 로그 데이터의 별도 인덱스 분리를 위한 다양한 방식을 정리해보았습니다.

라우팅 방식

• MDC 활용: Java Application에서 MDC를 이용한 로그 라우팅
• 별도 Logger 활용: 특정 로그를 별도 Logger를 통해 분리

인덱스 분리 방식

• OpenSearch Transform Jobs: 데이터 마이그레이션을 위한 활용 가능성 검토
• Custom Fluent-bit: 커스텀 로그 처리를 통한 데이터 이동 및 관리
• AWS Firehose + Lambda: Firehose와 Lambda를 이용한 로그 데이터 전송 및 저장

위와 같이 다양한 방식을 통해 장기간 보관이 필요한 특정 로그를 분리하여 관리할 수 있는 방법을 확인했습니다.

실제 운영 환경에서 적절한 방식들을 선택하고 조합하면, 더욱 효과적인 로그 관리가 가능하다고 생각합니다.

이 글을 읽는 분들에게 효과적인 로그 분리 방식을 도입하는 데 도움이 되었으면 좋겠습니다!