Logstash プラグインによる SLS ログの Kafka と HDFS へのストリーミング - SLS

Simple Log Service を使用すると、Logstash を介してログデータを消費できます。この機能を利用するには、Logstash 入力プラグイン Simple Log Service を設定し、Simple Log Service からデータをプルして、Kafka や HDFS などの他のシステムに送信します。

主な機能

分散型消費：単一の Logstore から同時にデータを消費するように、複数のサーバーを設定できます。
高性能：Java コンシューマーグループ実装に基づくため、プラグインはシングルコアで最大 20 MB/s（圧縮前）のデータ処理が可能です。
高信頼性：消費進捗はサーバー上に保存されます。コンシューマーが中断された場合、最後のチェックポイントから自動的に再開されます。
自動負荷分散：システムはコンシューマー数に応じてシャードを自動的に割り当て、コンシューマーの追加または削除時に負荷を再バランス化します。

操作手順

お使いのオペレーティングシステムに対応した Logstash インストールパッケージを、以下のサイトからダウンロードします。Logstash。

以下の例では Linux 環境を使用します：

Logstash をインストールします。詳細については、公式の Logstash ドキュメントをご参照ください。
1. 公開署名キーをダウンロードしてインストールします。
```
sudo rpm --import https://artifacts.elastic.co/GPG-KEY-elasticsearch
```
2. /etc/yum.repos.d/ ディレクトリ内に、.repo 拡張子を持つリポジトリファイル（例： logstash.repo）を作成し、以下の内容を追加します：
```
[logstash-9.x]
name=Elastic repository for 9.x packages
baseurl=https://artifacts.elastic.co/packages/9.x/yum
gpgcheck=1
gpgkey=https://artifacts.elastic.co/GPG-KEY-elasticsearch
enabled=1
autorefresh=1
type=rpm-md
```
3. Logstash をダウンロードしてインストールします。
```
sudo yum install logstash
```
入力プラグインをインストールします。
1. logstash-input-sls を GitHub からダウンロードします。
2. 入力プラグインをインストールします。
```
/usr/share/logstash/bin/logstash-plugin install logstash-input-sls.zip
```
  説明
  インストール失敗の原因と対処方法については、「プラグインのインストールと構成」をご参照ください。
logstash ユーザーを作成します。Logstash は root 権限以外のユーザーとして実行する必要があります。
1. logstash ユーザーを作成します。
```
sudo adduser --system --no-create-home --group logstash
```
2. logstash ユーザーの権限を設定します。/usr/share/logstash、/etc/logstash、/var/log/logstash など、Logstash 関連のディレクトリはすべて logstash ユーザーが所有している必要があります。
```
sudo chown -R logstash:logstash /usr/share/logstash /etc/logstash /var/log/logstash
```
3. logstash ユーザーが正しく作成されたか確認します。
```
id logstash
```
  出力には logstash ユーザーのユーザー ID（UID）およびグループ ID（GID）が表示され、ユーザーの作成が正常に完了したことが確認できます。

logstash ユーザーとして Logstash を起動します。

/etc/logstash ディレクトリに、拡張子が .conf の構成ファイルを作成します。このトピックでは logstash-sample.conf を例に説明します。

サンプルコードを logstash-sample.conf ファイルに追加し、logstash ユーザーとして Logstash を起動します。

sudo -u logstash /usr/share/logstash/bin/logstash -f /etc/logstash/logstash-sample.conf

以下の例では、Logstash を Logstore からのデータ消費および標準出力への出力用に構成しています。例で使用される各パラメーターの説明は、以下に記載します。

パラメーター	サンプルコード `input { logservice{ endpoint => "your project endpoint" access_id => "your_accesskey_id" access_key => "your_accesskey_secret" project => "your project name" logstore => "your logstore name" consumer_group => "consumer group name" consumer_name => "consumer name" position => "end" checkpoint_second => 30 include_meta => true consumer_name_with_ip => true } } output { stdout {} }`
endpoint`string`（必須） SLS プロジェクトのサービスエンドポイントです。詳細については、「サービスエンドポイント」をご参照ください。
access_id`string`（必須） Alibaba Cloud アカウントの AccessKey ID です。この AccessKey ID には、コンシューマーグループを管理するための必要な権限が付与されている必要があります。詳細については、「Logstore からのデータ消費に必要な権限の付与」をご参照ください。
access_key`string`（必須） Alibaba Cloud アカウントの AccessKey シークレットです。この AccessKey シークレットには、コンシューマーグループを管理するための必要な権限が付与されている必要があります。詳細については、「Logstore からのデータ消費に必要な権限の付与」をご参照ください。
project`string`（必須） SLS プロジェクトの名称です。
logstore`string`（必須） SLS の Logstore 名です。
consumer_group`string`（必須）コンシューマーグループの名称です。
consumer_name`string`（必須）コンシューマーの名称です。この名称は、コンシューマーグループ内で一意である必要があります。
position`string`（必須）データ消費を開始する位置です。 begin：Logstore 内の最初のログエントリから消費を開始します。 end：現在時刻から消費を開始します。 yyyy-MM-dd HH:mm:ss：指定した時刻から消費を開始します。
checkpoint_second`number`（任意）チェックポイントを保存する間隔（秒単位）です。値は 10～60 の整数で指定します。デフォルト値は 30 です。
include_meta`boolean`（任意）出力にメタデータを含めるかどうかを指定します。メタデータには、ソース、時間、タグ、トピックなどのフィールドが含まれます。デフォルト値は true です。
consumer_name_with_ip`boolean`（任意）コンシューマーの IP アドレスをコンシューマー名に追加するかどうかを指定します。分散型消費を有効にするには、このパラメーターを `true` に設定する必要があります。デフォルト値は true です。