이벤트 로그 서버 구축 (5) - 데이터 적재 파트 2

지난 글에서는 데이터 적재를 위한 AWS 서비스(S3, Firehose)를 설정하고, 이를 Serverless Framework를 활용해 효율적으로 관리하는 방법을 알아보았습니다. 이번 글에서는 이벤트 로그를 Firehose로 전송하는 코드를 작성하고, 이를 실제로 배포 및 테스트하는 과정을 살펴보겠습니다.

 

1. 이벤트 로그 전송 코드 추가

이벤트 로그 적재에 필요한 리소스인 S3와 Firehose의 배포가 완료되었다면, 이제 이전에 작성했던 코드에 Firehose로 데이터를 전송하는 로직을 추가합니다.

 

먼저 @aws-sdk/firehose-client 패키지를 설치합니다. 이 패키지는 Firehose에 데이터를 전송하기 위한 AWS SDK 모듈 입니다.

$ pnpm add @aws-sdk/client-firehose

 

그리고 이전에 작성했던 코드에 Firehose 클라이언트를 초기화하고 데이터를 Firehose를 전송하는 코드를 추가합니다. 전체 코드는 아래와 같습니다.

 

[src/handler.ts]

import { APIGatewayProxyEvent, APIGatewayProxyResult } from "aws-lambda";
import { z, Zoderror } from 'zod';
// +++
import { FirehoseClient, PutRecordCommand } from "@aws-sdk/client-firehose";

// +++
const firehoseClient = new FirehoseClient({ region: "ap-northeast-2" });
const firehoseDeliveryStreamName = process.env.FIREHOSE_DELIVERY_STREAM_NAME;

const eventLogSchema = z.preprocess(
  (input) => {
    if (typeof input === "string") {
      return JSON.parse(input);
    }
    throw new Error("Invalid JSON format");
  },
  z.object({
    event_name: z.string().nonempty(),
    event_date: z.string().regex(/^\d{4}-\d{2}-\d{2}$/),
    event_params: z.record(z.any()),
  })
);

export const recordEventLog = async (
  event: APIGatewayProxyEvent
): Promise<APIGatewayProxyResult> => {
  const { body } = event;

  try {
    const eventLog = eventLogSchema.parse(body);

		// +++    
    const eventLogString = JSON.stringify(eventLog);
    const putRecordCommand = new PutRecordCommand({
      DeliveryStreamName: firehoseDeliveryStreamName,
      Record: {
        Data: Buffer.from(eventLogString),
      },
    });
    await firehoseClient.send(putRecordCommand);

    return {
      statusCode: 200,
      body: JSON.stringify({
        message: "success",
      }),
    };
  } catch (e: unknown) {
    let statusCode: number = 500;
    let message: string = "Interval server error";
    if (e instanceof ZodError) {
      statusCode = 400;
      message = e.errors.map((e) => `[${e.path}: ${e.message}]`).join(",");
    }
    if (e instanceof Error) {
      message = e.message;
    }

    return {
      statusCode,
      body: JSON.stringify({
        message,
      }),
    };
  }
};

 

위에서 추가한 코드 중에 firehoseDeliveryStreamName 변수는 환경 변수에서 가져오고 있습니다. Node.js에서는 환경 변수를 주로 .env 파일에서 관리합니다. 이를 위해 프로젝트 루트에 .env 파일을 생성하고, FIREHOSE_DELIVERY_STREAM_NAME 변수를 추가합니다. 이 값은 serverless.yml에서 정의한 DeliveryStreamName에 설정한 값을 사용합니다.

 

[.env.dev]

FIREHOSE_DELIVERY_STREAM_NAME=event-log-stream-dev

 

이제 .env에 정의한 환경 변수를 실행 시 로드해야 합니다. 기존에는 serverless-dotenv-plugin을 사용했지만, v4부터는 더 이상 지원하지 않는 것으로 보입니다. 대신, v4에서는 .env 파일을 자동으로 로드하며, --stage 옵션을 사용하면 환경별로 다른 .env 파일을 구분하여 사용할 수 있습니다. 다만, 주의할 점은 .env 파일의 변수가 자동으로 process.env에 등록되는 것이 아니라, serverless.yml에서 ${env:}로 참조할 수 있게 등록된다는 점입니다. 

# .env.dev를 로드합니다.
$ sls offline --stage dev

# .env.prod를 로드합니다.
$ sls offline --stage prod

 

로드한 환경 변수를 코드에서 사용하려면 process.env에 등록해야 합니다. 이를 위해 serverless.yml 파일의 functions 섹션에 환경 변수를 추가해야 합니다. 사실, 이는 AWS Lambda의 환경 변수에 등록하는 것으로, 배포 후 AWS Console에서 해당 변수를 확인할 수 있습니다. 로컬에서는 serverless-offline이 API Gateway와 Lambda를 시뮬레이션하기 때문에 이 동작이 동일하게 적용됩니다.

 

[serverless.yml]

...
functions:
  recordEventLog:
    handler: src/handler.recordEventLog
    # +++
    environment:
	    FIREHOSE_DELIVERY_STREAM_NAME: ${env:FIREHOSE_DELIVERY_STREAM_NAME}
    events:
      - httpApi:
          path: /
          method: POST

 

로컬 테스트를 위해 serverless-offline을 실행한 후 요청을 보내보겠습니다. 요청이 성공적으로 처리되면 S3 버킷에 이벤트 로그 파일이 저장됩니다.

테스트를 마쳤으니, 이제 AWS에 배포를 진행해 실제 환경에서도 이벤트 로그 처리가 정상적으로 동작하는지 확인하겠습니다. 하지만, 배포 전에 serverless.yml에 Lambda 함수가 Firehose로 데이터를 전송할 수 있도록 필요한 IAM 권한을 추가해야 합니다. 아래와 같이 provider 부분에 iam을 추가합니다.

 

[serverless.yml]

...

provider:
  name: aws
  runtime: nodejs20.x
  region: ap-northeast-2
  # +++
  iam:
    role:
      statements:
        - Effect: Allow
          Action:
            - firehose:PutRecord
            - firehose:PutRecordBatch
          Resource:
            - !Sub "arn:aws:firehose:${AWS::Region}:${AWS::AccountId}:deliverystream/event-log-stream-${opt:stage}"
            
...

$ sls deploy --stage dev

 

배포가 완료 되었다면, 요청을 보내기 전에 Lambda에 환경 변수가 제대로 등록 되었는지 확인해보세요. 환경 변수가 잘 등록 되었다면 똑같이 요청을 보내 이벤트 로그가 S3에 잘 쌓이는지 확인해보세요.

 

지금까지 작성된 최종 코드는 아래와 같습니다.

[src/handler.ts]

import { APIGatewayProxyEvent, APIGatewayProxyResult } from "aws-lambda";
import { FirehoseClient, PutRecordCommand } from "@aws-sdk/client-firehose";
import { z, ZodError } from "zod";

const firehoseClient = new FirehoseClient({ region: "ap-northeast-2" });
const firehoseDeliveryStreamName = process.env.FIREHOSE_DELIVERY_STREAM_NAME;

const eventLogSchema = z.preprocess(
  (input) => {
    if (typeof input === "string") {
      return JSON.parse(input);
    }
    throw new Error("Invalid JSON format");
  },
  z.object({
    event_name: z.string().nonempty(),
    event_date: z.string().regex(/^\d{4}-\d{2}-\d{2}$/),
    event_params: z.record(z.any()),
  })
);

export const recordEventLog = async (
  event: APIGatewayProxyEvent
): Promise<APIGatewayProxyResult> => {
  const { body } = event;

  try {
    const eventLog = eventLogSchema.parse(body);

    const eventLogString = JSON.stringify(eventLog);

    const putRecordCommand = new PutRecordCommand({
      DeliveryStreamName: firehoseDeliveryStreamName,
      Record: {
        Data: Buffer.from(eventLogString),
      },
    });
    await firehoseClient.send(putRecordCommand);

    return {
      statusCode: 200,
      body: JSON.stringify({
        message: "success",
      }),
    };
  } catch (e: unknown) {
    let statusCode: number = 500;
    let message: string = "Interval server error";
    if (e instanceof ZodError) {
      statusCode = 400;
      message = e.errors.map((e) => `[${e.path}: ${e.message}]`).join(",");
    }
    if (e instanceof Error) {
      message = e.message;
    }

    return {
      statusCode,
      body: JSON.stringify({
        message,
      }),
    };
  }
};

 

[serverless.yml]

# "org" ensures this Service is used with the correct Serverless Framework Access Key.
org: tjdals12
# "service" is the name of this project. This will also be added to your AWS resource names.
service: event-log-server

provider:
  name: aws
  runtime: nodejs20.x
  region: ap-northeast-2

functions:
  recordEventLog:
    handler: src/handler.recordEventLog
    environment:
      FIREHOSE_DELIVERY_STREAM_NAME: ${env:FIREHOSE_DELIVERY_STREAM_NAME}
    events:
      - httpApi:
          path: /
          method: POST

plugins:
  - serverless-offline

resources:
  Resources:
    EventLogBucket:
      Type: AWS::S3::Bucket
      Properties:
        BucketName: tjdals12-event-log-${opt:stage}
        OwnershipControls:
          Rules:
            - ObjectOwnership: BucketOwnerEnforced
        PublicAccessBlockConfiguration:
          BlockPublicAcls: true
          BlockPublicPolicy: true
          IgnorePublicAcls: true
          RestrictPublicBuckets: true
        VersioningConfiguration:
          Status: "Suspended"
    EventLogFirehoseRole:
      Type: AWS::IAM::Role
      Properties:
        RoleName: event-log-firehose-role-${opt:stage}
        AssumeRolePolicyDocument:
          Version: "2012-10-17"
          Statement:
            - Effect: Allow
              Principal:
                Service: firehose.amazonaws.com
              Action: sts:AssumeRole
        Policies:
          - PolicyName: FirehoseS3Access
            PolicyDocument:
              Version: "2012-10-17"
              Statement:
                - Effect: Allow
                  Action:
                    - s3:PutObject
                    - s3:PutObjectAcl
                  Resource:
                    - !Sub "${EventLogBucket.Arn}/*"
    EventLogFirehoseStream:
      Type: AWS::KinesisFirehose::DeliveryStream
      Properties:
        DeliveryStreamName: event-log-stream-${opt:stage}
        DeliveryStreamType: DirectPut
        ExtendedS3DestinationConfiguration:
          Prefix: "events/year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/"
          ErrorOutputPrefix: "errors/!{firehose:error-output-type}/year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/"
          CustomTimeZone: UTC
          BucketARN: !GetAtt EventLogBucket.Arn
          RoleARN: !GetAtt EventLogFirehoseRole.Arn

 

[.env.dev]

FIREHOSE_DELIVERY_STREAM_NAME=event-log-stream-dev

 

[.env.prod]

FIREHOSE_DELIVERY_STREAM_NAME=event-log-stream-prod

 

다음 글에서는

이번 글에서는 이벤트 로그를 Firehose에 전송하는 코드를 작성하고, 이를 실제로 배포해 테스트하는 과정까지 진행했습니다. 다음 글에서는 이렇게 적재된 이벤트 로그를 분석하기 위한 인프라를 설정하고, 이를 활용하는 방법에 대해 다뤄보겠습니다.

 

2025.01.04 - [이벤트 로그] - 이벤트 로그 서버 구축 (1) - 개요

2025.01.11 - [이벤트 로그] - 이벤트 로그 서버 구축 (2) - 데이터 수집 파트 1

2025.01.18 - [이벤트 로그] - 이벤트 로그 서버 구축 (3) - 데이터 수집 파트 2

2025.01.25 - [이벤트 로그] - 이벤트 로그 서버 구축 (4) - 데이터 적재 파트 1

2025.02.02 - [이벤트 로그] - 이벤트 로그 서버 구축 (5) - 데이터 적재 파트 2

2025.02.09 - [이벤트 로그] - 이벤트 로그 서버 구축 (6) - 데이터 분석 파트