【第2回】Amazon MSKのsyslogとauthlogをMSK ConnectでS3へRaw保存する

はじめに

先行シリーズで構築したオンプレKafka ConnectのS3 Sinkを、MSK ConnectとConfluent Amazon S3 Sink Connectorへ置き換える。MSK Connectは既存bucketのraw/配下へ、切り替え後の新規データだけを追加する。

1. 既存S3 rawを確認する

先行シリーズで使用した環境ファイルとhome-rawlog-backup profileをそのまま使う。新しいRaw bucketは作成しない。

source /etc/rawlog-s3-backup/common.env
export RAW_BUCKET="${S3_BUCKET_NAME}"
export AWS_PROFILE=home-rawlog-backup

aws sts get-caller-identity --profile "${AWS_PROFILE}"
aws s3api head-bucket \
  --bucket "${RAW_BUCKET}" \
  --profile "${AWS_PROFILE}"
aws s3 ls "s3://${RAW_BUCKET}/raw/syslog/" \
  --recursive --profile "${AWS_PROFILE}" | tail
aws s3 ls "s3://${RAW_BUCKET}/raw/authlog/" \
  --recursive --profile "${AWS_PROFILE}" | tail

既存objectを上書きしないよう、切り替え時刻を時間境界に合わせる。バックアップ処理とMSK Connectが同じ時間prefixへ同時に書く期間は、object keyの衝突と二重計上を避けるため作らない。

2. pluginを登録する

AWS公式例はConfluent Amazon S3 Sink Connectorをcustom pluginとしてMSK Connectへ登録する。配布元、license、checksum、Kafka Connect互換versionを確認し、ZIPをartifact bucketへ置く。

export AWS_REGION=ap-northeast-1
export AWS_ADMIN_PROFILE=home-aws-admin
export PLUGIN_ZIP="$PWD/confluent-s3-sink.zip"

test -s "${PLUGIN_ZIP}" || exit 1
sha256sum "${PLUGIN_ZIP}"
aws s3 cp "${PLUGIN_ZIP}" \
  "s3://${ARTIFACT_BUCKET}/plugins/confluent-s3-sink.zip" \
  --region "${AWS_REGION}" --profile "${AWS_ADMIN_PROFILE}"

CUSTOM_PLUGIN_ARN=$(aws kafkaconnect create-custom-plugin \
  --region "${AWS_REGION}" \
  --name confluent-s3-sink \
  --content-type ZIP \
  --location "s3Location={bucketArn=arn:aws:s3:::${ARTIFACT_BUCKET},fileKey=plugins/confluent-s3-sink.zip}" \
  --query CustomPluginArn --output text \
  --profile "${AWS_ADMIN_PROFILE}")

while [ "$(aws kafkaconnect describe-custom-plugin \
  --region "${AWS_REGION}" --custom-plugin-arn "${CUSTOM_PLUGIN_ARN}" \
  --query CustomPluginState --output text \
  --profile "${AWS_ADMIN_PROFILE}")" != ACTIVE ]; do sleep 15; done

CUSTOM_PLUGIN_REVISION=$(aws kafkaconnect describe-custom-plugin \
  --region "${AWS_REGION}" --custom-plugin-arn "${CUSTOM_PLUGIN_ARN}" \
  --query LatestRevision.Revision --output text \
  --profile "${AWS_ADMIN_PROFILE}")

3. execution roleを作成する

roleには既存bucketのraw/ prefixへのListBucket、GetObject、PutObject、AbortMultipartUploadと、MSK接続に必要な権限だけを与える。Rawバックアップ用IAMユーザーのaccess keyやhome-rawlog-backup profileをMSK Connectへ渡さない。SSE-KMSならKMS権限も追加する。trust principalはkafkaconnect.amazonaws.comとする。

cat > /tmp/msk-connect-trust.json <<'EOF'
{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":{"Service":"kafkaconnect.amazonaws.com"},"Action":"sts:AssumeRole"}]}
EOF

MSK_CONNECT_ROLE_ARN=$(aws iam create-role \
  --role-name home-log-msk-connect-role \
  --assume-role-policy-document file:///tmp/msk-connect-trust.json \
  --query Role.Arn --output text --profile "${AWS_ADMIN_PROFILE}")

jq -n --arg bucket "${RAW_BUCKET}" '{Version:"2012-10-17",Statement:[
  {Effect:"Allow",Action:"s3:ListBucket",Resource:("arn:aws:s3:::"+$bucket),Condition:{StringLike:{"s3:prefix":"raw/*"}}},
  {Effect:"Allow",Action:["s3:GetObject","s3:PutObject","s3:AbortMultipartUpload"],Resource:("arn:aws:s3:::"+$bucket+"/raw/*")},
  {Effect:"Allow",Action:["kafka-cluster:Connect","kafka-cluster:DescribeCluster","kafka-cluster:ReadData","kafka-cluster:DescribeTopic","kafka-cluster:AlterGroup","kafka-cluster:DescribeGroup"],Resource:"*"},
  {Effect:"Allow",Action:["logs:CreateLogStream","logs:PutLogEvents"],Resource:"*"}
]}' > /tmp/msk-connect-policy.json

aws iam put-role-policy \
  --role-name home-log-msk-connect-role \
  --policy-name home-log-msk-connect-policy \
  --policy-document file:///tmp/msk-connect-policy.json \
  --profile "${AWS_ADMIN_PROFILE}"

動作確認後はkafka-clusterとCloudWatch LogsのResourceを、実際のcluster、topic、consumer group、log group ARNへ絞る。SSE-KMSの場合は対象keyのkms:Decryptとkms:GenerateDataKeyも追加する。

4. connectorを作成する

topicごとにconnectorを分けるとprefixと障害範囲を分離できる。以下はsyslog用の中心設定である。

connector.class=io.confluent.connect.s3.S3SinkConnector
tasks.max=1
topics=syslog
s3.region=ap-northeast-1
s3.bucket.name=実際のRAW_BUCKET名
topics.dir=raw
storage.class=io.confluent.connect.s3.storage.S3Storage
format.class=io.confluent.connect.s3.format.json.JsonFormat
schema.compatibility=NONE
partitioner.class=io.confluent.connect.storage.partitioner.TimeBasedPartitioner
path.format=YYYY/MM/dd/HH
partition.duration.ms=3600000
timezone=Asia/Tokyo
locale=en
timestamp.extractor=Wallclock
flush.size=500
rotate.interval.ms=600000
rotate.schedule.interval.ms=600000

topics.dir=rawによりtopic名が続き、raw/syslog/...となる。authlog用はtopics=authlogへ変更する。

MSK ConnectはMSKと同じVPCのprivate subnet/Security Groupへ置く。cluster認証に合わせてTLSとSASL/SCRAMまたはIAMを設定し、execution role、custom plugin revision、worker capacity、CloudWatch log deliveryを指定する。

MSK_CONNECT_SECURITY_GROUP_ID=$(aws ec2 create-security-group \
  --region "${AWS_REGION}" --vpc-id "${VPC_ID}" \
  --group-name home-log-msk-connect-sg \
  --description 'MSK Connect workers' \
  --query GroupId --output text --profile "${AWS_ADMIN_PROFILE}")
aws ec2 authorize-security-group-ingress \
  --region "${AWS_REGION}" --group-id "${MSK_SECURITY_GROUP_ID}" \
  --protocol tcp --port 9096 \
  --source-group "${MSK_CONNECT_SECURITY_GROUP_ID}" \
  --profile "${AWS_ADMIN_PROFILE}"

MSK_PRIVATE_BOOTSTRAP=$(aws kafka get-bootstrap-brokers \
  --region "${AWS_REGION}" --cluster-arn "${MSK_CLUSTER_ARN}" \
  --query BootstrapBrokerStringSaslScram --output text \
  --profile "${AWS_ADMIN_PROFILE}")

jq -n --arg bucket "${RAW_BUCKET}" '{
  "connector.class":"io.confluent.connect.s3.S3SinkConnector","tasks.max":"1",
  "topics":"syslog","s3.region":"ap-northeast-1","s3.bucket.name":$bucket,
  "topics.dir":"raw","storage.class":"io.confluent.connect.s3.storage.S3Storage",
  "format.class":"io.confluent.connect.s3.format.json.JsonFormat","schema.compatibility":"NONE",
  "partitioner.class":"io.confluent.connect.storage.partitioner.TimeBasedPartitioner",
  "path.format":"YYYY/MM/dd/HH","partition.duration.ms":"3600000",
  "timezone":"Asia/Tokyo","locale":"en","timestamp.extractor":"Wallclock",
  "flush.size":"500","rotate.interval.ms":"600000","rotate.schedule.interval.ms":"600000"
}' > /tmp/syslog-s3-connector.json

export KAFKA_CONNECT_VERSION=2.7.1
SYSLOG_CONNECTOR_ARN=$(aws kafkaconnect create-connector \
  --region "${AWS_REGION}" --connector-name home-log-syslog-s3 \
  --capacity 'provisionedCapacity={mcuCount=1,workerCount=1}' \
  --connector-configuration file:///tmp/syslog-s3-connector.json \
  --kafka-cluster "apacheKafkaCluster={bootstrapServers='${MSK_PRIVATE_BOOTSTRAP}',vpc={subnets=['${SUBNET_ID_1}','${SUBNET_ID_2}','${SUBNET_ID_3}'],securityGroups=['${MSK_CONNECT_SECURITY_GROUP_ID}']}}" \
  --kafka-cluster-client-authentication authenticationType=SASL_SCRAM \
  --kafka-cluster-encryption-in-transit encryptionType=TLS \
  --kafka-connect-version "${KAFKA_CONNECT_VERSION}" \
  --plugins "customPlugin={customPluginArn='${CUSTOM_PLUGIN_ARN}',revision=${CUSTOM_PLUGIN_REVISION}}" \
  --service-execution-role-arn "${MSK_CONNECT_ROLE_ARN}" \
  --query ConnectorArn --output text --profile "${AWS_ADMIN_PROFILE}")

aws kafkaconnect describe-connector \
  --region "${AWS_REGION}" --connector-arn "${SYSLOG_CONNECTOR_ARN}" \
  --query '{State:ConnectorState,Error:StateDescription}' \
  --profile "${AWS_ADMIN_PROFILE}"

KAFKA_CONNECT_VERSIONはpluginが対応するversionへ合わせる。authlog用はJSONのtopicsをauthlog、connector名をhome-log-authlog-s3へ変えて同じcommandを実行する。

5. objectを確認する

aws kafkaconnect list-connectors --region "$AWS_REGION"
aws s3 ls "s3://$RAW_BUCKET/raw/syslog/" --recursive --profile "${AWS_PROFILE}" | tail
aws s3 ls "s3://$RAW_BUCKET/raw/authlog/" --recursive --profile "${AWS_PROFILE}" | tail
LATEST_OBJECT=$(aws s3api list-objects-v2 \
  --bucket "${RAW_BUCKET}" --prefix raw/syslog/ \
  --query 'reverse(sort_by(Contents,&LastModified))[0].Key' \
  --output text --profile "${AWS_PROFILE}")
aws s3 cp "s3://${RAW_BUCKET}/${LATEST_OBJECT}" - \
  --profile "${AWS_PROFILE}" | head

JSONにts、src_host、program、messageが残ることを確認する。既存S3 SinkはWallclock partitionなので、event時刻ではなくconnector処理時刻でdirectoryが決まる。この性質も維持する。

6. 並行稼働と切り替え

1. オンプレS3 Sinkの最終offset、object数、サイズを記録する
2. 時間境界で対象prefixへのオンプレS3 Sinkの書き込みを止める
3. MSK Connectを起動し、次の時間prefixから新規objectが追加されることを確認する
4. 切り替え前後のKafka offsetとS3 recordを比較し、欠損と二重保存がないことを確認する
5. S3 objectが10分ごとに確定することを確認してからオンプレS3 Sinkを停止する

consumer groupが異なるため二重保存は可能だが、同じS3 connectorを二重起動しない。

注意

• flush.size=1は疎通確認には便利だがsmall fileとrequest costが増える
• JSON converter/schema設定は実データに合わせる
• MSK ConnectはPublic MSK endpoint経由ではなくVPC内接続にする
• connector削除前にlagと未確定objectを確認する

参考資料

• Amazon S3 sink connectorを設定する
• MSK Connect custom plugin

関連記事

• 【導入編】オンプレのログ分析基盤をAWSマネージドサービスへ移行する全体設計
• 【第1回】オンプレFluentdからAmazon MSKへsyslogとauthlogを送信する
• 【第3回】S3 Rawログの日次Spark処理をEMR Serverlessへ移行する

実機検証について

本稿は未検証である。pluginのversion、設定名、licenseをダウンロード時に確認する。