소개

이전에 쓴 FluentBit × Kinesis × S3로 로그 보존 기반 만들기 ~ 개요 편 ~의 계속입니다.

이번에는 AWS측의 구현인 Kinesis Firehose Delivery Stream에서 S3로 데이터를 흘리는 기반을 만듭니다.

리소스 생성

이번에는 CloudFormation을 사용하여 리소스를 만들었습니다.

필요한 리소스

S3 버킷

Kinesis Firehose Delivery Stream용 IAM 역할

S3에 쓸 수 있어야합니다

필요한 IAM 정책은 공식 페이지을 참조하십시오

Kinesis Firehose Delivery Stream

※ Kinesis Firehose Delivery Stream이 길기 때문에 이후에는 Firehose라고합니다.

템플릿

위의 요구 사항을 충족하는 템플릿은 다음과 같습니다.
버킷 이름은 반드시 쓰여 버리므로, 적절히 바꾸도록 하는 것이 좋다고 생각합니다.
(샘플용으로 쓰고 있으므로 자원명의 적당성은 용서해 주세요...)

firehose_to_s3.yaml

AWSTemplateFormatVersion: 2010-09-09
Resources:
  LogBucket:
    Type: AWS::S3::Bucket
    Properties:
      BucketName: log-bucket-stream-sample0021
      AccessControl: Private
  LogToS3Stream:
    Type: AWS::KinesisFirehose::DeliveryStream
    Properties:
      DeliveryStreamName: log-to-s3-stream
      DeliveryStreamType: DirectPut
      ExtendedS3DestinationConfiguration:
        BucketARN: !GetAtt LogBucket.Arn
        RoleARN: !GetAtt FirehoseToS3IamRole.Arn
        BufferingHints:
          IntervalInSeconds: 60
          SizeInMBs: 50
        CompressionFormat: GZIP
  FirehoseToS3IamRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName: firehose-to-s3-iam-role
      AssumeRolePolicyDocument:
        Version: 2012-10-17
        Statement:
          - Effect: Allow
            Principal:
              Service:
                - firehose.amazonaws.com
            Action:
              - sts:AssumeRole
      ManagedPolicyArns:
        - !Ref FirehoseToS3DeliveryPolicy
  FirehoseToS3DeliveryPolicy:
    Type: AWS::IAM::ManagedPolicy
    Properties:
      PolicyDocument:
        Version: 2012-10-17
        Statement:
          -
            Sid: AllowFirehoseToS3Delivery
            Effect: Allow
            Action:
              - s3:AbortMultipartUpload
              - s3:GetBucketLocation
              - s3:GetObject
              - s3:ListBucket
              - s3:ListBucketMultipartUploads
              - s3:PutObject
              - kinesis:DescribeStream
              - kinesis:GetShardIterator
              - kinesis:GetRecords
              - kinesis:ListShards
              - kms:Decrypt
              - kms:GenerateDataKey
              - logs:PutLogEvents
              - lambda:InvokeFunction
              - lambda:GetFunctionConfiguration
            Resource: "*"

여기를 CloudFormation에서 흘려 보냅니다.
변수는 특별히 포함되지 않으므로 설정이 필요하지 않습니다.

리소스 확인

S3

버킷이 생성되었습니다.
여기가 Firehose에서 전송되는 데이터가 기록되는 곳입니다.

Firehose

이런 식으로 리소스가 만들어졌습니다.


캐치 한 데이터를 쓰는 S3도 제대로 지정됩니다.
덧붙여서 여기의 Buffer conditions는, Firehose가 데이터를 캐치하고 나서, S3에 보내기까지의 간격치등의 버퍼 설정입니다.

움직여 보자

데이터 보내기

Firehose의 콘솔 화면에는 테스트 데이터를 보내는 버튼이 설치되어 있으므로 이것을 사용하여 Firehose에 데이터를 보냅니다.


센딩 데모 데이터를 클릭하면 데이터가 전송되기 시작합니다.
1번 누르면 1건 데이터가 보내지는 것이 아니라, 몇건이나 데이터가 연속으로 보내지는 것 같습니다.
그 때문에, 「Stop sending demo data」를 클릭하지 않으면 데이터의 송신이 멈추지 않으므로 주의를!

그건 그렇고{"ticker_symbol":"QXZ", "sector":"HEALTHCARE", "change":-0.05, "price":84.51}가 여러 번 보내지는 형태입니다. JSON 문자열입니다.

Firehose 모니터링을 확인해 봅니다.

Monitoring을 클릭하면 Firehose가 수신한 데이터의 양, S3에 전송한 데이터의 양 등의 지표를 볼 수 있습니다.
잠시 기다려 화면 업데이트를 해보면 Firehose가 데이터를 스트리밍하는 것을 알 수 있는 지표 데이터가 나타납니다.


이번에는 60개 데이터를 Firehose에 흘리고, 그것을 잡은 Firehose가 60개의 데이터를 S3에 썼습니다.

버킷 확인

S3을 살펴보면 날짜와 시간으로 구분된 디렉터리가 자동으로 생성되고, 그 안에 Firehose가 스트리밍으로 쓴 데이터의 파일이 배치됩니다.


gz 파일을 다운로드하고 less 명령 등으로 확인.
그러면


Firehose에서 보낸 데이터를 확인할 수있었습니다.

마지막으로

이번에는 Firehose에서 S3로 데이터를 스트리밍하는 기반을 구축했습니다.
다음은 컨테이너 측의 구현이되는 FluentBit에서 Firehose로 데이터를 보내는 부분에 대해 정리합니다.