Log Shipper 用 Microsoft Azure Event Hubs プラグイン

このドキュメントでは、 Netskope Cloud Exchangeプラットフォームの Log Shipper モジュールを使用して Microsoft Azure Event Hubs v2.0.0 プラグインを構成する方法について説明します。 このプラグインは、アラート (DLP、マルウェア、ポリシー、侵害されたクレデンシャル、マルサイト、検疫、修復、セキュリティ評価、ウォッチリスト、UBA、CTEP、デバイス、コンテンツ)、イベント (ページ、アプリケーション、監査、インフラストラクチャ、ネットワーク、インシデント、エンドポイント、クライアント ステータス)、および WebTx [ Netskope LogStreaming 経由] データの取り込みをサポートします。 データはMicrosoft Azure Event Hubsに送信されます。このプラグインは、CEF形式とJSON形式の両方のデータ取り込みをサポートしています。

前提条件

この設定を完了するには、以下が必要です。

• Netskopeテナント（または、例えば本番環境と開発／テスト環境など、複数のテナント）。
• Netskope Cloud Exchangeテナントには、テナントプラグインとログシッパープラグインが既に設定済みです。
• AWS Netskope Log StreamingまたはAzure Netskope Log Streamingプラグインが既に設定されているNetskope Cloud Exchangeテナント。
• イベントハブサービスが含まれる標準プランのサブスクリプションです。
• Event Hubsサービスの詳細については、こちらをご覧ください。
• Azure Event Hubs for Apache Kafka に関する情報については、こちらをご覧ください。
• 以下のホストへの接続: https://portal.azure.com/ 。

Microsoft Azure Event Hubs プラグインのサポート

このプラグインは、アラート (侵害されたクレデンシャル、ポリシー、マルサイト、マルウェア、DLP、セキュリティ評価、ウォッチリスト、検疫、修復、UBA、CTEP、デバイス、コンテンツ)、イベント (ページ、アプリケーション、監査、インフラストラクチャ、ネットワーク、インシデント、エンドポイント、クライアント ステータス)、および WebTx ( Netskopeログ ストリーミング経由) データの取り込みをサポートします。データは Microsoft Azure Event Hubs に送信されます。このプラグインは、CEF 形式と JSON 形式の両方での取り込みをサポートしています。

権限 

SAS ポリシーを追加する際は、以下の権限がチェックされていることを確認してください。

• 送信 
• Listen

bootstrap.servers=NAMESPACENAME.servicebus.windows.net:9093
security.protocol=SASL_SSL
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="$ConnectionString" password="{YOUR.EVENTHUBS.CONNECTION.STRING}";

consumer = KafkaConsumer(
                event_hub_name,
                bootstrap_servers=bootstrap_server,
                security_protocol=”SASL_SSL,
                sasl_mechanism="PLAIN",
                sasl_plain_username="$ConnectionString",
                sasl_plain_password=connection_string,
                request_timeout_ms=TIMEOUT_MS,
                reconnect_backoff_ms=TIMEOUT_MS,
                client_id=self._add_user_agent(),
)

available_event_hubs = consumer.topics()

producer = KafkaProducer(
                bootstrap_servers=bootstrap_server,
                security_protocol=”SASL_SSL”,
                sasl_mechanism="PLAIN",
                sasl_plain_username="$ConnectionString",
                sasl_plain_password=connection_string,
                batch_size=BATCH_SIZE,
                acks=ACKS,
                retries=retries,
                linger_ms=LINGER_MS,
                request_timeout_ms=TIMEOUT_MS,
                reconnect_backoff_ms=TIMEOUT_MS,
                value_serializer=lambda x: x.encode("utf-8"),
                client_id=self._add_user_agent(),
            )

producer.send(
    topic=event_hub_name,
           value=(
                data
                     if not isinstance(data, dict)
                     else json.dumps(data)
                ),
        )

パフォーマンスマトリックス

このパフォーマンス測定は、以下のVM仕様を持つ大規模Cloud Exchangeスタック上で実施されました。 これらの測定値は、約2000万件のログを1万件ずつのバッチでAzure Event Hubに取り込んだ結果です。

ユーザーエージェント

netskope-ce-6.0.0-cls-microsoft-azure-event-hubs-v2.0.0

ワークフロー

1. Microsoft Azure Event Hubs を構成します。
2. Microsoft Azure Event Hubs プラグインを設定します。
3. Microsoft Azure Event のログシッパービジネスルールを構成します。
4. Microsoft Azure Event の Log Shipper ログ配信を構成します。
5. プラグインを検証してください。

動画を見る

再生ボタンをクリックして動画をご覧ください。

Event Hub の名前空間を作成します

1. Microsoft Azure プラットフォームにログインします。

   

2. Event Hubsを検索します。

   

   

3. 名前空間を作成するには、 Createをクリックしてください。

   

4. 詳細をすべて入力して、 Review + createをクリックしてください。（料金プランについては、前提条件を参照してください。）

名前空間にイベントハブを作成する

1. 最近作成した名前空間に移動し、 + Event Hubをクリックして、その名前空間に新しいイベントハブを作成します。

   

2. 必要な情報をすべて入力し、 Review + createをクリックしてください。

   

イベントハブ用の接続文字列を生成する

1. Event Hubs 名前空間の接続文字列については、 Namespace > Settings > Shared access policiesを参照してください。

   

2. Addをクリックして新しい接続文字列を作成します。ポリシー名を入力し、以下の権限を確認してください。
   • 送信
   • Listen

   

3. ポリシーが作成されたら、そのポリシーをクリックしてプライマリ接続文字列をコピーします。

   

Microsoft Azure Event Hubs プラグインを構成する

1. Settings > Plugin Storeへ移動してください。Microsoft Azure Event Hubs v2.0.0 (CLS)プラグインを検索して選択してください。

   

2. プラグイン構成名を入力し、Microsoft Azure Event Hubs の既定のマッピング ファイルが選択されていることを確認してください。データをJSON形式で共有したい場合は、「フォーマット」設定をJSONに変更してください。

   

3. Nextをクリックし、構成パラメータを入力してください。
   • Namesspace Name: Microsoft Azure Event Hubs の名前空間名。
   • Port: Microsoft Azure Event Hubs ポート。Event HubsのTLSハンドシェイクのデフォルトポートは9093ですが、設定に基づいてカスタムポートを指定することもできます。
   • Event Hubs Namespace Connection String: Microsoft Azure Event Hubs 名前空間の接続文字列。例:エンドポイント=sb://<namespace_name>.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=<shared_access_key> 。 接続文字列を取得するには、 Microsoft Azure Event Hubs Namespace > Shared Access Policies > Primary connection stringにアクセスしてください。
   • Event Hub Name: Microsoft Azure Event Hub のユーザー名。 
   • Log Source Identifier: これはすべてのログの接頭辞として追加されます。（ログソース識別子には空白文字を含めないでください。）
   • タイムスタンプフィールドを除外する： タイムスタンプフィールドなしでデータを取り込むには、 Yes Select 。 このオプションはJSON形式のデータにのみ適用されます。
   • ログソース識別子フィールドを除外する: ログソース識別子フィールドなしでデータを取り込む場合は、「はい」 Select 。 このオプションはJSON形式のデータにのみ適用されます。
   • Producer Batch Size (in KB): プロデューサーが単一のリクエストでイベントハブに送信しようとするメッセージのバッチの最大サイズ（KB単位）。バッチサイズを大きくするとスループットは向上するが、メモリ使用量が増加する可能性がある。許容範囲：16～1024KB。
   • Buffer Memory (in MB): 未送信メッセージのバッファリング用に割り当てられたメモリの合計容量（MB）。バッファがいっぱいになると、空き容量ができるまで、それ以上の送信要求はブロックされるか、失敗する可能性があります。この値を上げることで、ログデータのバースト処理に役立ちます。許容範囲：32～128MB。
   • Max Block Time (in seconds): バッファがいっぱいになった場合に、送信操作が例外を発生させるまでの最大ブロック時間（秒単位）。これにより、無期限のブロックを防ぎ、高負荷時におけるアプリケーションの応答性を制御するのに役立ちます。許容範囲：10～60秒。
   • Batch Linger Time (in milliseconds): バッチサイズに達していない場合、プロデューサーがバッチを送信する前に追加のメッセージを待つ時間（ミリ秒単位）。値を大きくするとバッチ処理の効率は向上しますが、レイテンシが増加する可能性があります。許容範囲：50～1000ミリ秒。
   • Data Chunk Size: Event Hubsに送信する前に、単一のデータチャンクにグループ化するログの数。この設定を調整することで、データ送信の粒度を制御し、ワークロードのパフォーマンスを最適化できます。許容範囲：1000～10000。
   • Flush Timeout (in seconds): タイムアウトする前に、バッファリングされたすべてのレコードをイベントハブにフラッシュするために許容される最大時間（秒）。シャットダウンや手動フラッシュ操作中にバッファ内でデータが無期限に遅延しないようにします。許容範囲：10～300秒。

   注記

   • Exclude Timestamp Field and Exclude Log Source Identifier Field are mandatory fields for JSON formatted data.
   • メモリ割り当てに関連するエラーが発生した場合は、Event Hubsの設定に従ってバッファメモリを増やし、チャンクサイズを小さくしてください。
   • タイムアウトエラーが発生した場合は、ユーザーはEvent Hubsの設定に従って、フラッシュタイムアウトを長くし、チャンクサイズを小さくすることができます。
   • データ取り込みに時間がかかりすぎる場合は、Event Hubsの設定を増やす必要があります。
   • 大規模なスタックの場合、10個以上のパーティションと20個以上のスループットユニットを用意することが推奨されます。
   • 中規模スタックの場合、7つ以上のパーティションと10以上のスループットユニットを用意することが推奨されます。
   • 小規模なスタックの場合、パーティション数は5つ以上、スループットユニット数は5つ以上が推奨されます。

   

   

4. Saveをクリックしてください。新しいプラグインの設定は、 Log Shipper > プラグインのページで確認できます。

Microsoft Azure Event Hubs のビジネスルールを構成する

1. Log ShipperでBusiness Rulesに移動します。
2. デフォルトでは、すべてのアラートとイベントをフィルタリングするビジネスルールが設定されています。特定の種類のアラートやイベントを除外したい場合は、 Create New Ruleをクリックして、ルール名とフィルターを追加して新しいビジネスルールを設定します。
3. Saveをクリックしてください。

   

Microsoft Azure Event Hubs のログシッパーによるログ配信を構成する

1. Log Shipperで、 Log Deliveryに移動してAdd Log Delivery Configurationをクリックします。
2. Sourceプラグイン(AWS Log StreamingまたはAzure Log Streaming)、Destinationプラグイン(Microsoft Azure Event Hubs)、ビジネスルールをSelect、Saveをクリックします。
3. ログ配信構成が追加されると、 Netskopeテナントからのデータの取得、変換、および Microsoft Azure Event Hubs プラットフォームへの取り込みが開始されます。

Microsoft Azure Event Hubs プラグインを検証する

プルを検証する

1. Cloud Exchangeで、 Loggingに移動して、プルしたログを検索します。

   

   

   

   

プッシュ通知を検証する

Cloud Exchange でプラグインのワークフローを検証するには：

Loggingに移動して、フィルターメッセージに「取り込み済み」が含まれるイベントとアラートを検索します。取り込まれたログはフィルタリングされます。

Microsoft Azure Event Hubs へのプッシュを検証するには：

1. Microsoft Azure プラットフォームにログインします。
2. 行く Event Hubs > {Namespace} > Data Explorer.
3. イベントハブ、パーティションID、イベント位置Select 、 View Eventsをクリックします。 新しいポジションを選択すると、プラグインによって取り込まれたライブイベントを取得できます。 「最古の位置」を選択すると、メッセージの保持期間に基づいて取り込まれた最も古いイベントを取得できます。また、「カスタム位置」を選択すると、カスタムフィルターからイベントのセットを検索できます。

   

   

   

   

   

   

   

   

   

   

   

   

   

   

   

   

   

   

   

   

   

   

JSON形式で取り込んだ場合、データは以下のようになります。

タイムスタンプとログソース識別子フィールドを含まないJSON形式で取り込んだ場合、データは以下のようになります。

Microsoft Azure Event Hubs プラグインのトラブルシューティング

Microsoft Azure Event Hubs プラグインの保存に関する問題

すべてのパラメータを入力して「保存」ボタンをクリックしても、設定が指定された設定と異なる場合など、エラーが発生する可能性があります。

What to do:

構成パラメーターが正しくないことが原因かもしれません。Microsoft Azure Event Hub の構成セクションの手順に従ってください。

Microsoft Azure Event Hubs でイベントを確認できません

イベントの取り込みが成功した後でも、プラグインから取り込まれたイベントを確認できません。これには以下の理由が考えられます。

• イベントハブでイベントを表示するための設定が間違っています。
• イベント表示待機時間中は、イベントは取り込まれません。
• または、お探しのデータが保存期間外にある可能性があります。

What to do:

1. リアルタイムで取り込まれたイベントを表示するには、 Newest Positionが選択されているかどうかを確認してください。
2. 取り込まれたイベントについては、 Cloud Exchange上のログを確認してください。
3. イベントハブの保持期間を確認するには、名前空間の概要ページに移動し、その中のメッセージの保持期間を確認してください。イベントハブを編集することで、イベントハブの保持期間を変更できます。「名前空間にイベントハブを作成する」セクションを参照してください。

プラグインのアップグレード後にプラグインを有効にできません

JSON形式のデータを取り込むプラグインを追加してアップグレードした場合、 「スキップ」ボタンをクリックすると、プラグインのアップグレード後にプラグイン設定を有効にする際に、 「除外タイムスタンプフィールド」と「除外ログソース識別子フィールド」でエラーが発生します。

何をするか：

「除外タイムスタンプフィールド」と「除外ログソース識別子フィールド」はJSON形式のデータでは必須フィールドであるため、要件に応じて両方のフィールドで「はい」または「いいえ」を選択する必要があります。

プラグインの設定/キューへの追加/プラグインの無効化ができません

プラグインの保存、キューの設定、またはMicrosoft Azure Event Hubsプラグインの無効化ができない場合は、SSL証明書の検証エラーが原因である可能性があります。

What to do:

オンプレミスのデバイスでCloud Exchange実行していて、 「[SSL: CERTIFICATE_VERIFY_FAILED] 証明書の検証に失敗しました: ローカル発行者の証明書を取得できません」エラーを受け取った場合は、組織 IT チームに連絡してください。 この問題は、EC2 や Azure などのクラウド プラットフォームを使用することで解決できます。

Microsoft Azure Event Hubsへのデータ取り込み中にタイムアウトエラーが発生しました

プラグイン設定で設定されたフラッシュタイムアウトとチャンクサイズが原因で、ユーザーは以下のようなタイムアウトエラーに遭遇する可能性があります。

What to do:

• Event Hubsの設定に応じて、フラッシュタイムアウトを長くし、チャンクサイズを小さくしてください。

Microsoft Azure Event Hubs プラグインの既知の動作

• このプラグインは BWAN イベント、クライアント ステータスタイプのイベント、デバイスおよびコンテンツタイプのアラートの取り込みをサポートしていないため、ユーザーは BWAN イベント、クライアント ステータスタイプのイベント、デバイスおよびコンテンツタイプのアラートを取り込むことができない場合があります。  
• ユーザーは、アクセント付き文字（英語）、英語以外の言語の文字、改行なしスペース、改行文字、その他の特殊な書式設定記号など、いくつかの要因により、取り込まれたデータにエスケープ文字が含まれている場合があります。
  例：
  

  <14>Apr 07 09:32:56 alltypes CEF:0|Netskope|Mock Netskope Tenant|NULL|application|NULL|Unknown|act=Download appcategory=Cloud Storage applicationType=nspolicy browser=unknown \\u4ed5\\u4e8b cci=89 ccl=high device=Other dst=ef82::1a12:1234:1b12 os=unknown requestClientApplication=Box sourceServiceName=Box src=ef82::1a12:1234:1b12 suser=support@netskope.com timestamp=1743736484

  
  

  上記のログでは「 \\u4ed5\\u4e8b 」のように見えた、摂取された2つの日本語文字を以下に示します。

• プラグインのアップグレード後、設定パラメータが空になっている場合があります。