0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@naritomo08(nari tomo)

【第1回】オンプレFluentdからAmazon MSKへsyslogとauthlogを送信する

0
Last updated at Posted at 2026-09-18

はじめに

オンプレKafka 3.7.1のsyslogとauthlogをAmazon MSKへ移す。個人PoCとしてPublic accessを扱うが、本番ではVPN等のprivate connectivityを優先する。

AWS側の管理操作には、Iceberg構築記事で作成したhome-aws-admin profileを使用する。以降のAWS CLI例を実行するshellで次を設定し、対象accountを必ず確認する。

管理端末にはAWS CLI v2、jq、curlを導入しておく。

export AWS_PROFILE=home-aws-admin
aws sts get-caller-identity --profile "${AWS_PROFILE}"
aws --version
jq --version
curl --version | head -n 1

このprofileはMSKを操作する端末用であり、brokerや後続のAWSサービスへaccess keyを配布するものではない。

構成

Public accessはcluster作成後に有効化する。認証なしでは利用せず、client-broker encryptionをTLSにする。

1. networkを準備する

3 AZのpublic subnet、Internet Gatewayへのroute、MSK用Security GroupをAWS CLIで作成する。以下では新規VPCを作るため、既存VPCへ追加する場合はCIDRの重複に注意する。

export AWS_REGION=ap-northeast-1
export VPC_CIDR=10.42.0.0/16
export MSK_CLUSTER_NAME=home-log-msk

# 利用可能なAZを3つ取得する。3 AZないregionでは別regionを選ぶ。
AZS=($(aws ec2 describe-availability-zones \
  --region "${AWS_REGION}" \
  --filters Name=state,Values=available \
  --query 'AvailabilityZones[].ZoneName' \
  --output text \
  --profile "${AWS_PROFILE}"))

if [ "${#AZS[@]}" -lt 3 ]; then
  echo "3つ以上のAZを利用できません" >&2
  exit 1
fi

VPC_ID=$(aws ec2 create-vpc \
  --region "${AWS_REGION}" \
  --cidr-block "${VPC_CIDR}" \
  --tag-specifications \
    'ResourceType=vpc,Tags=[{Key=Name,Value=home-log-msk-vpc}]' \
  --query 'Vpc.VpcId' \
  --output text \
  --profile "${AWS_PROFILE}")

aws ec2 wait vpc-available \
  --region "${AWS_REGION}" \
  --vpc-ids "${VPC_ID}" \
  --profile "${AWS_PROFILE}"

# MSK brokerの名前解決に必要なVPC DNSを有効化する。
aws ec2 modify-vpc-attribute \
  --region "${AWS_REGION}" \
  --vpc-id "${VPC_ID}" \
  --enable-dns-support '{"Value":true}' \
  --profile "${AWS_PROFILE}"
aws ec2 modify-vpc-attribute \
  --region "${AWS_REGION}" \
  --vpc-id "${VPC_ID}" \
  --enable-dns-hostnames '{"Value":true}' \
  --profile "${AWS_PROFILE}"

IGW_ID=$(aws ec2 create-internet-gateway \
  --region "${AWS_REGION}" \
  --tag-specifications \
    'ResourceType=internet-gateway,Tags=[{Key=Name,Value=home-log-msk-igw}]' \
  --query 'InternetGateway.InternetGatewayId' \
  --output text \
  --profile "${AWS_PROFILE}")
aws ec2 attach-internet-gateway \
  --region "${AWS_REGION}" \
  --internet-gateway-id "${IGW_ID}" \
  --vpc-id "${VPC_ID}" \
  --profile "${AWS_PROFILE}"

ROUTE_TABLE_ID=$(aws ec2 create-route-table \
  --region "${AWS_REGION}" \
  --vpc-id "${VPC_ID}" \
  --tag-specifications \
    'ResourceType=route-table,Tags=[{Key=Name,Value=home-log-msk-public-rt}]' \
  --query 'RouteTable.RouteTableId' \
  --output text \
  --profile "${AWS_PROFILE}")
aws ec2 create-route \
  --region "${AWS_REGION}" \
  --route-table-id "${ROUTE_TABLE_ID}" \
  --destination-cidr-block 0.0.0.0/0 \
  --gateway-id "${IGW_ID}" \
  --profile "${AWS_PROFILE}"

SUBNET_ID_1=$(aws ec2 create-subnet \
  --region "${AWS_REGION}" --vpc-id "${VPC_ID}" \
  --availability-zone "${AZS[0]}" --cidr-block 10.42.1.0/24 \
  --tag-specifications \
    'ResourceType=subnet,Tags=[{Key=Name,Value=home-log-msk-public-1}]' \
  --query 'Subnet.SubnetId' --output text --profile "${AWS_PROFILE}")
SUBNET_ID_2=$(aws ec2 create-subnet \
  --region "${AWS_REGION}" --vpc-id "${VPC_ID}" \
  --availability-zone "${AZS[1]}" --cidr-block 10.42.2.0/24 \
  --tag-specifications \
    'ResourceType=subnet,Tags=[{Key=Name,Value=home-log-msk-public-2}]' \
  --query 'Subnet.SubnetId' --output text --profile "${AWS_PROFILE}")
SUBNET_ID_3=$(aws ec2 create-subnet \
  --region "${AWS_REGION}" --vpc-id "${VPC_ID}" \
  --availability-zone "${AZS[2]}" --cidr-block 10.42.3.0/24 \
  --tag-specifications \
    'ResourceType=subnet,Tags=[{Key=Name,Value=home-log-msk-public-3}]' \
  --query 'Subnet.SubnetId' --output text --profile "${AWS_PROFILE}")

for SUBNET_ID in "${SUBNET_ID_1}" "${SUBNET_ID_2}" "${SUBNET_ID_3}"; do
  aws ec2 associate-route-table \
    --region "${AWS_REGION}" \
    --route-table-id "${ROUTE_TABLE_ID}" \
    --subnet-id "${SUBNET_ID}" \
    --profile "${AWS_PROFILE}"
done

MSK_SECURITY_GROUP_ID=$(aws ec2 create-security-group \
  --region "${AWS_REGION}" \
  --group-name home-log-msk-sg \
  --description 'Allow SASL/SCRAM access from home' \
  --vpc-id "${VPC_ID}" \
  --tag-specifications \
    'ResourceType=security-group,Tags=[{Key=Name,Value=home-log-msk-sg}]' \
  --query 'GroupId' \
  --output text \
  --profile "${AWS_PROFILE}")

MY_IP=$(curl -fsS -4 https://checkip.amazonaws.com | tr -d '[:space:]')
aws ec2 authorize-security-group-ingress \
  --region "${AWS_REGION}" \
  --group-id "${MSK_SECURITY_GROUP_ID}" \
  --protocol tcp \
  --port 9196 \
  --cidr "${MY_IP}/32" \
  --profile "${AWS_PROFILE}"

printf 'VPC_ID=%s\nSUBNETS=%s,%s,%s\nMSK_SECURITY_GROUP_ID=%s\nMY_IP=%s\n' \
  "${VPC_ID}" "${SUBNET_ID_1}" "${SUBNET_ID_2}" "${SUBNET_ID_3}" \
  "${MSK_SECURITY_GROUP_ID}" "${MY_IP}"

Public accessのSASL/SCRAM portは9196である。inboundは自宅global IPの/32だけを許可し、0.0.0.0/0は使わない。IPが変わった場合は古いCIDRをrevoke-security-group-ingressで削除してから、新しいCIDRを追加する。

新規VPCのdefault NACLはinbound/outboundを全許可するため、この手順では追加設定しない。custom NACLを使う場合は、9196/tcpのinboundに加えてclient側ephemeral portへの戻りtrafficを許可する。

2. secretを作成する

MSKのSASL/SCRAMはAWS Secrets Managerのsecretを関連付ける。secret名はAmazonMSK_で始め、MSK clusterと同じaccount/regionに作成する。また、AWS managed keyではなくcustomer managed KMS keyを使う。

KMS_KEY_ID=$(aws kms create-key \
  --region "${AWS_REGION}" \
  --description 'KMS key for Amazon MSK SCRAM secrets' \
  --query 'KeyMetadata.KeyId' \
  --output text \
  --profile "${AWS_PROFILE}")
aws kms create-alias \
  --region "${AWS_REGION}" \
  --alias-name alias/home-log-msk-scram \
  --target-key-id "${KMS_KEY_ID}" \
  --profile "${AWS_PROFILE}"

# shell historyや記事へpasswordを残さない。
read -r -s -p 'MSK password: ' MSK_PASSWORD
echo
SECRET_STRING=$(jq -cn \
  --arg username fluentd \
  --arg password "${MSK_PASSWORD}" \
  '{username:$username,password:$password}')

MSK_SECRET_ARN=$(aws secretsmanager create-secret \
  --region "${AWS_REGION}" \
  --name AmazonMSK_home_fluentd \
  --kms-key-id "${KMS_KEY_ID}" \
  --secret-string "${SECRET_STRING}" \
  --query ARN \
  --output text \
  --profile "${AWS_PROFILE}")
unset MSK_PASSWORD SECRET_STRING

3. MSK clusterを作成する

MSK Provisioned clusterをAWS CLIで作成する。まず、利用可能なKafka versionを確認し、オンプレ3.7.1 clientと互換性のあるversionを選ぶ。以下のKAFKA_VERSIONは確認結果に置き換える。

aws kafka list-kafka-versions \
  --region "${AWS_REGION}" \
  --query 'KafkaVersions[].Version' \
  --output table \
  --profile "${AWS_PROFILE}"

export KAFKA_VERSION='<利用するversion>'

cat > /tmp/msk-server.properties <<'EOF'
auto.create.topics.enable=false
EOF

MSK_CONFIGURATION_ARN=$(aws kafka create-configuration \
  --region "${AWS_REGION}" \
  --name home-log-msk-config \
  --kafka-versions "${KAFKA_VERSION}" \
  --server-properties fileb:///tmp/msk-server.properties \
  --query Arn \
  --output text \
  --profile "${AWS_PROFILE}")

LOG_GROUP_NAME=/aws/msk/home-log-msk
aws logs create-log-group \
  --region "${AWS_REGION}" \
  --log-group-name "${LOG_GROUP_NAME}" \
  --profile "${AWS_PROFILE}"
aws logs put-retention-policy \
  --region "${AWS_REGION}" \
  --log-group-name "${LOG_GROUP_NAME}" \
  --retention-in-days 30 \
  --profile "${AWS_PROFILE}"

BROKER_NODE_GROUP_INFO=$(jq -cn \
  --arg subnet1 "${SUBNET_ID_1}" \
  --arg subnet2 "${SUBNET_ID_2}" \
  --arg subnet3 "${SUBNET_ID_3}" \
  --arg sg "${MSK_SECURITY_GROUP_ID}" \
  '{InstanceType:"kafka.m5.large",
    ClientSubnets:[$subnet1,$subnet2,$subnet3],
    SecurityGroups:[$sg],
    StorageInfo:{EbsStorageInfo:{VolumeSize:100}}}')

MSK_CLUSTER_ARN=$(aws kafka create-cluster \
  --region "${AWS_REGION}" \
  --cluster-name "${MSK_CLUSTER_NAME}" \
  --kafka-version "${KAFKA_VERSION}" \
  --number-of-broker-nodes 3 \
  --broker-node-group-info "${BROKER_NODE_GROUP_INFO}" \
  --encryption-info \
    'EncryptionInTransit={ClientBroker=TLS,InCluster=true}' \
  --client-authentication \
    'Sasl={Scram={Enabled=true}},Unauthenticated={Enabled=false}' \
  --configuration-info \
    "Arn=${MSK_CONFIGURATION_ARN},Revision=1" \
  --logging-info \
    "BrokerLogs={CloudWatchLogs={Enabled=true,LogGroup=${LOG_GROUP_NAME}}}" \
  --query ClusterArn \
  --output text \
  --profile "${AWS_PROFILE}")

aws kafka wait cluster-active \
  --region "${AWS_REGION}" \
  --cluster-arn "${MSK_CLUSTER_ARN}" \
  --profile "${AWS_PROFILE}"

clusterがACTIVEになったらsecretを関連付ける。関連付け時にAmazon MSKがsecretへ必要なresource policyを追加するため、そのpolicyを後から変更しない。

aws kafka batch-associate-scram-secret \
  --region "${AWS_REGION}" \
  --cluster-arn "${MSK_CLUSTER_ARN}" \
  --secret-arn-list "${MSK_SECRET_ARN}" \
  --profile "${AWS_PROFILE}"

aws kafka wait cluster-active \
  --region "${AWS_REGION}" \
  --cluster-arn "${MSK_CLUSTER_ARN}" \
  --profile "${AWS_PROFILE}"

Public accessはcluster作成時には指定できない。secret関連付け後、clusterのcurrent versionを取得して有効化する。

MSK_CURRENT_VERSION=$(aws kafka describe-cluster-v2 \
  --region "${AWS_REGION}" \
  --cluster-arn "${MSK_CLUSTER_ARN}" \
  --query 'ClusterInfo.CurrentVersion' \
  --output text \
  --profile "${AWS_PROFILE}")

aws kafka update-connectivity \
  --region "${AWS_REGION}" \
  --cluster-arn "${MSK_CLUSTER_ARN}" \
  --current-version "${MSK_CURRENT_VERSION}" \
  --connectivity-info 'PublicAccess={Type=SERVICE_PROVIDED_EIPS}' \
  --profile "${AWS_PROFILE}"

aws kafka wait cluster-active \
  --region "${AWS_REGION}" \
  --cluster-arn "${MSK_CLUSTER_ARN}" \
  --profile "${AWS_PROFILE}"

MSK_PUBLIC_BOOTSTRAP=$(aws kafka get-bootstrap-brokers \
  --region "${AWS_REGION}" \
  --cluster-arn "${MSK_CLUSTER_ARN}" \
  --query 'BootstrapBrokerStringPublicSaslScram' \
  --output text \
  --profile "${AWS_PROFILE}")

test -n "${MSK_PUBLIC_BOOTSTRAP}" && \
  test "${MSK_PUBLIC_BOOTSTRAP}" != 'None' || {
    echo 'Public SASL/SCRAM bootstrap brokerを取得できません' >&2
    exit 1
  }
printf '%s\n' "${MSK_PUBLIC_BOOTSTRAP}"

出力のBootstrapBrokerStringPublicSaslScramを使用する。private用bootstrap stringと取り違えない。

4. topicを作成する

Kafka 3.7.1 clientをオンプレに残して使う。

read -r -s -p 'MSK password: ' MSK_PASSWORD
echo
case "${MSK_PASSWORD}" in
  *\"*|*\\*) echo 'この例ではdouble quoteとbackslashを含まないpasswordを使用する' >&2; exit 1 ;;
esac

umask 077
cat > /tmp/msk-client.properties <<EOF
security.protocol=SASL_SSL
sasl.mechanism=SCRAM-SHA-512
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="fluentd" password="${MSK_PASSWORD}";
ssl.endpoint.identification.algorithm=https
EOF
sudo install -o root -g root -m 600 \
  /tmp/msk-client.properties /etc/kafka/msk-client.properties
rm /tmp/msk-client.properties

/opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server "$MSK_PUBLIC_BOOTSTRAP" \
  --command-config /etc/kafka/msk-client.properties \
  --create --topic syslog --partitions 3 --replication-factor 3

/opt/kafka/bin/kafka-topics.sh \
  --bootstrap-server "$MSK_PUBLIC_BOOTSTRAP" \
  --command-config /etc/kafka/msk-client.properties \
  --create --topic authlog --partitions 3 --replication-factor 3

5. Fluentdを接続する

fluent-plugin-kafkaがSASL_SSL/SCRAMを利用できるversionであることを確認する。passwordはrootのみ読めるEnvironmentFile等から渡し、設定例のplaceholderを実値へ展開して配置する。

<match syslog.relay>
  @type kafka2
  brokers "#{ENV['MSK_PUBLIC_BOOTSTRAP']}"
  default_topic syslog
  use_event_time true
  ssl_ca_certs_from_system true
  sasl_over_ssl true
  username fluentd
  password "#{ENV['MSK_PASSWORD']}"
  sasl_mechanism scram-sha-512
  required_acks -1
  compression_codec gzip
  <format>
    @type json
  </format>
</match>

authlog側も既存のtag/parserを変えず、default_topic authlogとする。いきなり旧Kafkaを止めず、短時間の試験送信後に切り替える。

Fluentd serviceへ値を渡すEnvironmentFileとsystemd overrideを作成する。passwordに改行、space、quoteを含める場合はsystemdのEnvironmentFile構文に合わせてescapeする。

printf 'MSK_PUBLIC_BOOTSTRAP=%s\nMSK_PASSWORD=%s\n' \
  "${MSK_PUBLIC_BOOTSTRAP}" "${MSK_PASSWORD}" |
  sudo tee /etc/fluent/msk.env >/dev/null
sudo chmod 600 /etc/fluent/msk.env

sudo mkdir -p /etc/systemd/system/fluentd.service.d
printf '%s\n' '[Service]' 'EnvironmentFile=/etc/fluent/msk.env' |
  sudo tee /etc/systemd/system/fluentd.service.d/msk.conf >/dev/null
sudo systemctl daemon-reload
unset MSK_PASSWORD

設定反映前に構文を確認し、サービスを再起動する。package名やunit名は導入方法に合わせて読み替える。

sudo fluentd --dry-run -c /etc/fluent/fluent.conf
sudo systemctl restart fluentd
sudo systemctl --no-pager --full status fluentd

6. 確認する

logger -p user.info "msk syslog test $(hostname) $(date -Is)"

/opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server "$MSK_PUBLIC_BOOTSTRAP" \
  --consumer.config /etc/kafka/msk-client.properties \
  --topic syslog --from-beginning --max-messages 10

authlogは別端末でSSH loginを発生させ、同様にconsumerで確認する。CloudWatchのBytesInPerSec、MessagesInPerSec、UnderReplicatedPartitionsも見る。

/opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server "${MSK_PUBLIC_BOOTSTRAP}" \
  --consumer.config /etc/kafka/msk-client.properties \
  --topic authlog --from-beginning --max-messages 10

broker単位の主要metricは次で確認できる。例ではbroker ID 1を対象にするため、2、3についても同じように実行する。

aws cloudwatch get-metric-data \
  --region "${AWS_REGION}" \
  --start-time "$(date -u -v-15M +%Y-%m-%dT%H:%M:%SZ)" \
  --end-time "$(date -u +%Y-%m-%dT%H:%M:%SZ)" \
  --metric-data-queries "[\
    {\"Id\":\"bytes_in\",\"MetricStat\":{\"Metric\":{\"Namespace\":\"AWS/Kafka\",\"MetricName\":\"BytesInPerSec\",\"Dimensions\":[{\"Name\":\"Cluster Name\",\"Value\":\"${MSK_CLUSTER_NAME}\"},{\"Name\":\"Broker ID\",\"Value\":\"1\"}]},\"Period\":60,\"Stat\":\"Sum\"}},\
    {\"Id\":\"messages_in\",\"MetricStat\":{\"Metric\":{\"Namespace\":\"AWS/Kafka\",\"MetricName\":\"MessagesInPerSec\",\"Dimensions\":[{\"Name\":\"Cluster Name\",\"Value\":\"${MSK_CLUSTER_NAME}\"},{\"Name\":\"Broker ID\",\"Value\":\"1\"}]},\"Period\":60,\"Stat\":\"Sum\"}},\
    {\"Id\":\"under_replicated\",\"MetricStat\":{\"Metric\":{\"Namespace\":\"AWS/Kafka\",\"MetricName\":\"UnderReplicatedPartitions\",\"Dimensions\":[{\"Name\":\"Cluster Name\",\"Value\":\"${MSK_CLUSTER_NAME}\"},{\"Name\":\"Broker ID\",\"Value\":\"1\"}]},\"Period\":60,\"Stat\":\"Maximum\"}}\
  ]" \
  --profile "${AWS_PROFILE}"

上のdate -vはmacOS用である。Linuxではstart timeをdate -u -d '15 minutes ago' +%Y-%m-%dT%H:%M:%SZで作る。

rollback

Fluentdの送信先を旧kafka1:9092,kafka2:9092,kafka3:9092へ戻して再起動する。MSK topicにだけ到着したoffset範囲を記録し、raw保存開始後に欠損・重複を確認する。

注意

Public accessはPoCの例外である。TLSでもbroker endpointはInternetから到達可能になるため、認証、送信元IP制限、secret rotation、監視が必要である。検証後はPublic accessを無効化し、VPN、Direct Connect、Transit Gateway等へ移す。

参考資料

関連記事

実機検証について

本稿は未検証の構築手順である。実行前にAWS CLIのlist-kafka-versionsと公式documentで、Public access対応cluster type、利用可能Kafka version、portを再確認する。

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?