@Bean
fun objectMapperSnakeCase(): ObjectMapper {
    return jacksonObjectMapper().also {
        it.propertyNamingStrategy = PropertyNamingStrategies.SNAKE_CASE
    }
}

<appender name="jsonFormattedAppender" class="ch.qos.logback.core.ConsoleAppender">
    <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
        <pattern>%msg %n</pattern>
    </encoder>
</appender>

<logger name="opensearch_efk" level="INFO" additivity="false">
    <appender-ref ref="jsonFormattedAppender" />
</logger>

class OpenSearchDocument(
    val loggerName: String,
    val document: Any
)

@Service
class OpenSearchService(
    val objectMapperSnakeCase: ObjectMapper
) {
    companion object {
        private const val OPENSEARCH_LOGGER_NAME = "opensearch_efk"
    }
 
    private val openSearchLogger by LoggerDelegator(OPENSEARCH_LOGGER_NAME)

    fun sendByFluentBit(document: Any) {
       openSearchLogger.info(
          objectMapperSnakeCase.writeValueAsString(
             OpensearchDocumentDto(openSearchLogger.name, document)
          )
       )
    }
}

class OsDocument(
    val eventType: String,
    val eventCode: String,
    // 생략
) {
    class Products(
        val productId: String,
        val productName: String,
        // 생략 
    )
}

openSearchService.sendByFluentBit(OsDocument(..))

{"logger_name":"opensearch_efk","document":{"event_type":"테스트","event_code":"테스트","event_date":"2024-04-15T10:09:53+09:00","event_id":"테스트","store_id":"테스트","store_name":"테스트","products":{"product_id":"테스트","product_name":"테스트","base_quantity":0,"remain_quantity":0,"stocked_in_quantity":0,"category1":"테스트","category2":"테스트","category3":"테스트","brand":"테스트","partner":"테스트","price":17000,"change_remain_quantity":0,"change_stocked_in_quantity":0,"oy_pick":false}}}

[SERVICE]
    Flush  1
    Log_Level  info
    parsers_file  /fluent-bit/etc/extra-parsers.conf
 
[FILTER]
    Name  rewrite_tag
    Match_regex  ^((?!opensearch).)*$
    Rule $log .*logger_name\":\"opensearch_efk.* opensearch false
 
[FILTER]
    Name  parser
    Match  opensearch
    Parser  stock_live_parser
    Key_Name  log

[FILTER]
    Name record_modifier
    Match opensearch
    Allowlist_key document

[FILTER]
    Name nest
    Match opensearch
    Operation lift
    Nested_under document
 
[OUTPUT]
    Name  opensearch
    Match  opensearch
    Host  vpc-....amazonaws.com
    Port  443
    Index  store-stock-%Y%m%d
    Type  _doc
    tls  On
    tls.verify  Off
    Suppress_Type_Name  On
    AWS_Auth  Off
    AWS_Region  ap-northeast-2
    Http_User  ${FLUENTBIT_OS_USER}
    Http_Passwd  ${FLUENTBIT_OS_PW}
    Retry_Limit  2
 
[OUTPUT]
    Name  stdout
    Match  opensearch
    Log_Level  info

[PARSER]
    Name  stock_live_parser
    Format  json

{"log"=>"{"data1":"test1", "data2":"test2"}"}

FROM public.ecr.aws/aws-observability/aws-for-fluent-bit:latest
 
COPY extra.conf /fluent-bit/etc/extra.conf
COPY extra-parsers.conf /fluent-bit/etc/extra-parsers.conf

# aws sso login
aws sso login --profile ECS_Developer-40**

# ecr 접속
aws --profile ECS_Developer-40** ecr get-login-password --region ap-northeast-2 | docker login --username AWS --password-stdin 40**.dkr.ecr.ap-northeast-2.amazonaws.com

# docker build
# mac os 를 사용할 경우 --platform linux/amd64 옵션을 주어 이미지 호환성 오류 방지
docker build . --platform linux/amd64 -t 40**.dkr.ecr.ap-northeast-2.amazonaws.com/catalog-inventory-fluentbit:latest

# aws ecr에 푸시
docker push 40**.dkr.ecr.ap-northeast-2.amazonaws.com/catalog-inventory-fluentbit:latest

{
    "containerDefinitions": [
        {
            "name": "catalog-api-dev-container",  # app container
            "image": "40**.dkr.ecr.ap-northeast-2.amazonaws.com/catalog-api-dev-repo",
            "logConfiguration": {
                "logDriver": "awsfirelens",   # log driver 로 firelens 사용
                "secretOptions": [
                    {
                        "valueFrom": "arn:aws:ssm:ap-northeast-2:**:parameter/datadog/apikey/aws/ecs/fargate",
                        "name": "apikey"
                    }
                ],
                "options": { # datadog 로그 전송을 위한 설정으로 본 포스팅에선 다루지 않습니다.
                    "dd_message_key": "log",
                    "compress": "gzip",
                    "provider": "ecs",
                    "dd_service": "catalog-api-dev",
                    "Host": "http-intake.logs.datadoghq.com",
                    "TLS": "on",
                    "dd_source": "fargate",
                    "dd_tags": "env:dev,env_type:fargate",
                    "Name": "datadog"
                }
            },
//         ... 생략
        },
        {
            "name": "oymall-datadog",  # datadog agent (jvm metric 전송용)
            "image": "public.ecr.aws/datadog/agent:latest",
//           ... 생략
        },
        {
            "name": "fluentbit-log-router",  # fluentbit container 
            "image": "40**.dkr.ecr.ap-northeast-2.amazonaws.com/catalog-inventory-fluentbit:latest",  # custom docker image 사용
            "environment": [
                {
                    "name": "FLUENTBIT_OS_USER",
                    "value": "fluentbit"    # fluentbit container 에서 사용할 opensearch user 정보
                },
                {
                    "name": "FLUENTBIT_OS_PW",
                    "value": ""
                }
            ],
            "logConfiguration": {
                "logDriver": "awslogs", # fluentbit container 로그를 cloudwatch 에서 확인 필요할 때 사용. fluentbit 설정에 output stdout 필요. 
                "options": {
                    "awslogs-create-group": "true",
                    "awslogs-group": "/ecs/catalog-api-dev-task",
                    "awslogs-region": "ap-northeast-2",
                    "awslogs-stream-prefix": "ecs"
                }
            },
            "firelensConfiguration": {  # firelens 설정
                "type": "fluentbit",
                "options": {
                    "config-file-type": "file",
                    "config-file-value": "/fluent-bit/etc/extra.conf",  # 앞서 설정한 fluentbit conf 파일 지정
                    "enable-ecs-log-metadata": "true"   #ecs 로그 수집을 위해 true 선언 필수
                }
            },
//           ... 생략
        }
    ],
    "requiresCompatibilities": [
        "FARGATE"
    ]
//    ... 생략
}

implementation("org.opensearch.client:opensearch-rest-client:2.12.0")
implementation("org.opensearch.client:opensearch-java:2.6.0")

@Configuration
class OpenSearchClientProvider(
   val credentialEncoder: CredentialEncoder
) {
   @Autowired
   lateinit var opensearchProperty: OpensearchProperty

   @Bean
   fun osClientCredentials() : BasicCredentialsProvider { 
      val basicCredentialProvider = BasicCredentialsProvider()
      basicCredentialProvider.setCredentials(
         AuthScope.ANY, 
         UsernamePasswordCredentials(
            opensearchProperty.username,
            credentialEncoder.decryptInternalByAwsKms(opensearchProperty.password, KmsKey.OPENSEARCH)
         )
      )

      return basicCredentialProvider
   }

   @Bean 
   fun openSearchClient(
      @Qualifier("osClientCredentials") 
      credentialsProvider: CredentialsProvider
   ): OpenSearchClient { 
      val httpHost = HttpHost(opensearchProperty.host, opensearchProperty.port, opensearchProperty.scheme)
      val restClient = RestClient.builder(httpHost)
          .setHttpClientConfigCallback { httpClientBuilder ->
             httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider)
          }.build()
      val transport = RestClientTransport(restClient, JacksonJsonpMapper())

      return OpenSearchClient(transport)
   }
}

@Service
class OpensearchService(
    @Qualifier("openSearchClient")
    val openSearchClient: OpenSearchClient,
) {
   fun <TDocument> sendByBulkApi(
      docs: List<TDocument>,
      index: OpensearchIndex
   ): BulkResponse {
      val bulkRequest = BulkRequest.Builder()
      docs.forEach {
         bulkRequest.operations { operation: BulkOperation.Builder ->
            operation.index { idx: IndexOperation.Builder<TDocument> ->
               idx.index(OpensearchIndex.dateFormattedIndex(index))
                  .id(null)
                  .document(it)
            }
         }
      }

      return openSearchClient.bulk(bulkRequest.build())
   }
}