Logstash と DCR ベースの API を使用してログを Microsoft Sentinel にストリーミングする

Microsoft Sentinelの Logstash 出力プラグインでは、データ収集規則 (DCR) を使用したパイプライン変換と高度な構成がサポートされています。 プラグインは、外部データ ソースから Log Analytics または Microsoft Sentinelのカスタム テーブルまたは標準テーブルにログを転送します。

この記事では、出力スキーマを完全に制御して、Log Analytics または DCR を使用してMicrosoft Sentinelにデータをストリーミングするように Logstash プラグインを設定する方法について説明します。

プラグインを使用すると、次のことができます。

• 列名と型の構成を制御します。
• フィルター処理やエンリッチメントなどのインジェスト時間変換を実行します。
• カスタム ログをカスタム テーブルに取り込むか、Syslog 入力ストリームを Log Analytics Syslog テーブルに取り込みます。

標準テーブルへのインジェストは、 カスタム ログ インジェストでサポートされている標準テーブルにのみ制限されます。

Logstash データ収集エンジンの操作の詳細については、「 Logstash の概要」を参照してください。

アーキテクチャの概要

Logstash エンジンは、次の 3 つのコンポーネントで構成されます。

• 入力プラグイン: さまざまなソースからのデータのカスタマイズされたコレクション。
• フィルター プラグイン: 指定した条件に従ったデータの操作と正規化。
• 出力プラグイン: 収集および処理されたデータをさまざまな宛先にカスタマイズして送信します。

プラグインは、ログ インジェスト API を使用して JSON 形式のデータを Log Analytics ワークスペースに送信します。 データは、カスタム ログまたは標準テーブルに取り込まれます。

• ログ インジェスト API の詳細については、こちらをご覧ください。

Logstash でMicrosoft Sentinel出力プラグインをデプロイする

プラグインを設定するには、次の手順に従います。

• Logstashプラグインの前提条件を確認してください
• プラグインをインストールする
• サンプル ファイルを作成する
• 必要な DCR 関連リソースを作成する
• Logstash 構成ファイルを構成する
• Logstash を再起動する
• Microsoft Sentinelで受信ログを表示する
• 出力プラグインの監査ログを監視する

Logstash プラグインの前提条件

• サポートされているバージョンの Logstash をインストールします。 このプラグインでは、次の Logstash バージョンがサポートされています。

  • 7.0 - 7.17.13

  • 8.0 - 8.9(これらのバージョンはLogstashによると セキュリティアップデートが必要です)

  • 8.11 - 8.15(これらのバージョンはLogstashによると セキュリティアップデートが必要です)

  • 8.19.2(このバージョンは Logstashによるとセキュリティアップデートが必要です)

  • 9.0.8(このバージョンはLogstashによると セキュリティアップデートが必要)

  • 9.1.10(このバージョンは Logstashによるとセキュリティアップデートが必要です)

  • 9.2.4 - 9.2.5(これらのバージョンはLogstashによると セキュリティアップデートが必要です)

  • 9.3.3

  • 9.4.0

    注:

    Logstash 8 を使用する場合は、 パイプラインで ECS を無効にすることをお勧めします。

• 少なくとも共同作成者の権限を持つ Log Analytics ワークスペースがあることを確認します。

• ワークスペースに DCR オブジェクトを作成するアクセス許可があることを確認します。

プラグインをインストールする

Microsoft Sentinel出力プラグインは、RubyGems の Logstash コレクションで使用できます。

• Logstash の 「Working with plugins」ドキュメントの手順に従って、microsoft-sentinel-log-analytics-logstash-output-plugin プラグインをインストールします。 既存の Logstash インストールにインストールするには、次のコマンドを実行します。

  logstash-plugin install microsoft-sentinel-log-analytics-logstash-output-plugin
  

• Logstash システムにインターネットにアクセスできない場合は、Logstash オフライン プラグイン管理 ドキュメントの指示に従って、オフライン プラグイン パックを準備して使用します。 (これには、インターネット にアクセスできる別の Logstash システムを構築する必要があります)。

サンプル ファイルを作成する

このセクションでは、次のいずれかのシナリオでサンプル ファイルを作成します。

• カスタム ログのサンプル ファイルを作成する
• Syslog テーブルにログを取り込むサンプル ファイルを作成する

カスタム ログのサンプル ファイルを作成する

このシナリオでは、イベントをMicrosoft Sentinelに送信するように Logstash 入力プラグインを構成します。 この例では、ジェネレーター入力プラグインを使用してイベントをシミュレートします。 その他の入力プラグインを使用できます。

この例では、Logstash 構成ファイルは次のようになります。

input {
      generator {
            lines => [
                 "This is a test log message"
            ]
           count => 10
      }
}

サンプル ファイルを作成するには、次の手順に従います。

1. 以下の出力プラグイン構成を Logstash 構成ファイルにコピーします。

   output {
       microsoft-sentinel-log-analytics-logstash-output-plugin {
         create_sample_file => true
         sample_file_path => "<enter the path to the file in which the sample data will be written>" #for example: "c:\\temp" (for windows) or "/tmp" for Linux. 
       }
   }
   

2. 参照先のファイル パスが既に存在することを確認し、Logstash を開始します。

   このプラグインは、サンプリングするイベントが 10 個ある場合、または Logstash プロセスが正常に終了すると、構成されたパス内の sampleFile<epoch seconds>.json という名前のサンプル ファイルに 10 個のレコードを書き込みます。 例: c:\temp\sampleFile1648453501.json。 プラグインが作成するサンプル ファイルの一部を次に示します。

   [
           {
               "host": "logstashMachine",
               "sequence": 0,
               "message": "This is a test log message",
               "ls_timestamp": "2022-03-28T17:45:01.690Z",
               "ls_version": "1"
           },
           {
               "host": "logstashMachine",
               "sequence": 1
       ...
   
       ]    
   

   プラグインは、次のプロパティをすべてのレコードに自動的に追加します。

   • ls_timestamp: 入力プラグインからレコードを受信した時刻
   • ls_version: Logstash のパイプラインのバージョン。

   これらのフィールドは、DCR を作成するときに削除できます。

Syslog テーブルにログを取り込むサンプル ファイルを作成する

このシナリオでは、syslog イベントをMicrosoft Sentinelに送信するように Logstash 入力プラグインを構成します。

1. Syslog メッセージがまだ Logstash マシンに転送されていない場合は、logger コマンドを使用してメッセージを生成できます。 たとえば (Linuxの場合):

   logger -p local4.warn --rfc3164 --tcp -t CEF "0|Microsoft|Device|cef-test|example|data|1|here is some more data for the example" -P 514 -d -n 127.0.0.1
   

   Logstash 入力プラグインの例を次に示します。

   input {
        syslog {
            port => 514
       }
   }
   

2. 以下の出力プラグイン構成を Logstash 構成ファイルにコピーします。

   output {
       microsoft-sentinel-log-analytics-logstash-output-plugin {
         create_sample_file => true
         sample_file_path => "<enter the path to the file in which the sample data will be written>" #for example: "c:\\temp" (for windows) or "/tmp" for Linux. 
       }
   }
   

3. ファイル パスが既に存在することを確認し、Logstash を開始します。

   このプラグインは、サンプリングするイベントが 10 個ある場合、または Logstash プロセスが正常に終了すると、構成されたパス内の sampleFile<epoch seconds>.json という名前のサンプル ファイルに 10 個のレコードを書き込みます。 例: c:\temp\sampleFile1648453501.json。 プラグインが作成するサンプル ファイルの一部を次に示します。

   [
           {
               "logsource": "logstashMachine",
               "facility": 20,
               "severity_label": "Warning",
               "severity": 4,
               "timestamp": "Apr  7 08:26:04",
               "program": "CEF:",
               "host": "127.0.0.1",
               "facility_label": "local4",
               "priority": 164,
               "message": "0|Microsoft|Device|cef-test|example|data|1|here is some more data for the example",
               "ls_timestamp": "2022-04-07T08:26:04.000Z",
               "ls_version": "1"
           }
   ]    
   
   

   プラグインは、次のプロパティをすべてのレコードに自動的に追加します。

   • ls_timestamp: 入力プラグインからレコードを受信した時刻
   • ls_version: Logstash のパイプラインのバージョン。

   これらのフィールドは、DCR を作成するときに削除できます。

必要な DCR リソースを作成する

MICROSOFT SENTINEL DCR ベースの Logstash プラグインを構成するには、まず DCR 関連のリソースを作成します。

このセクションでは、次のいずれかのシナリオで、DCR に使用するリソースを作成します。

• カスタム テーブルへのインジェスト用の DCR リソースを作成する
• 標準テーブルへのインジェスト用の DCR リソースを作成する

カスタム テーブルへのインジェスト用の DCR リソースを作成する

データをカスタム テーブルに取り込むには、次の手順に従います（REST API を使用して Azure Monitor Logs にデータを送信する (Azure portal) チュートリアルに基づいています）。

1. 前提条件については、こちらをご覧ください。

2. アプリケーションを構成します。

3. カスタム ログ テーブルを追加します。

4. 前のセクションで作成したサンプル ファイルを使用して、サンプル データを解析してフィルター処理します。

5. DCR から情報を収集します。

6. DCR にアクセス許可を割り当てます。

   [サンプル データの送信] ステップをスキップします。

もし問題が発生した場合は、 Logs Ingestion APIのトラブルシューティング手順を参照してください。

標準テーブルへのインジェスト用の DCR リソースを作成する

Syslog や CommonSecurityLog などの標準テーブルにデータを取り込むには、REST API (Resource Manager テンプレート) を使用して Azure Monitor ログにデータを送信するチュートリアルに基づいたプロセスを使用します。 このチュートリアルでは、カスタム テーブルにデータを取り込む方法について説明しますが、データを標準テーブルに取り込むプロセスを簡単に調整できます。 次の手順は、手順の関連する変更を示しています。

1. 前提条件については、こちらをご覧ください。

2. ワークスペースの詳細を収集します。

3. アプリケーションを構成します。

   Log Analytics ワークスペースで新しいテーブルを作成する手順をスキップします。 テーブルは Log Analytics で既に定義されているため、データを標準テーブルに取り込む場合、この手順は関係ありません。

4. DCR を作成します。 この手順では、次の手順を実行します。

   • 作成したサンプルファイルを「 Create a sample file」で提供してください。
   • 作成したサンプル ファイルを使用して、 streamDeclarations プロパティを定義します。 サンプル ファイル内の各フィールドには、同じ名前と適切な型を持つ対応する列が必要です (下の例を参照)。
   • カスタム テーブルの代わりに標準テーブルの名前を使用して、 outputStream プロパティの値を構成します。 カスタム テーブルとは異なり、標準のテーブル名には _CL サフィックスがありません。
   • テーブル名のプレフィックスは、Microsoft-ではなくCustom-する必要があります。 この例では、 outputStream プロパティの値が Microsoft-Syslog。

5. DCR にアクセス許可を割り当てます。

   [サンプル データの送信] ステップをスキップします。

もし問題が発生した場合は、 Logs Ingestion APIのトラブルシューティング手順を参照してください。

例: Syslog テーブルにデータを取り込む DCR

次の点に留意してください。

• streamDeclarations列名と型は、サンプル ファイル フィールドと同じである必要がありますが、それらすべてを指定する必要はありません。 たとえば、次の DCR では、 PRI、 type 、 ls_version の各フィールドは、 streamDeclarations 列から省略されます。
• dataflows プロパティは、入力を Syslog テーブル形式に変換し、outputStreamを Microsoft-Syslog に設定します。

{
  "$schema": "https://schema.management.azure.com/schemas/2019-04-01/deploymentTemplate.json#",
  "contentVersion": "1.0.0.0",
  "parameters": {
    "dataCollectionRuleName": {
      "type": "String",
      "metadata": {
        "description": "Specifies the name of the Data Collection Rule to create."
      }
    },
    "location": {
      "defaultValue": "[resourceGroup().location]",
      "type": "String",
      "metadata": {
        "description": "Specifies the location in which to create the Data Collection Rule."
      }
    },
    "workspaceResourceId": {
      "type": "String",
      "metadata": {
        "description": "Specifies the Azure resource ID of the Log Analytics workspace to use."
      }
    }
  },
  "resources": [
    {
      "type": "Microsoft.Insights/dataCollectionRules",
      "apiVersion": "2021-09-01-preview",
      "name": "[parameters('dataCollectionRuleName')]",
      "location": "[parameters('location')]",
      "properties": {
        "streamDeclarations": {
          "Custom-SyslogStream": {
            "columns": [
              { "name": "ls_timestamp", "type": "datetime" },
              { "name": "timestamp", "type": "datetime" },
              { "name": "message", "type": "string" },
              { "name": "facility_label", "type": "string" },
              { "name": "severity_label", "type": "string" },
              { "name": "host", "type": "string" },
              { "name": "logsource", "type": "string" }
            ]
          }
        },
        "destinations": {
          "logAnalytics": [
            {
              "workspaceResourceId": "[parameters('workspaceResourceId')]",
              "name": "clv2ws1"
            }
          ]
        },
        "dataFlows": [
          {
            "streams": ["Custom-SyslogStream"],
            "destinations": ["clv2ws1"],
            "transformKql": "source | project TimeGenerated = ls_timestamp, EventTime = todatetime(timestamp), Computer = logsource, HostName = logsource, HostIP = host, SyslogMessage = message, Facility = facility_label, SeverityLevel = severity_label",
            "outputStream": "Microsoft-Syslog"
          }
        ]
      }
    }
  ],
  "outputs": {
    "dataCollectionRuleId": {
      "type": "String",
      "value": "[resourceId('Microsoft.Insights/dataCollectionRules', parameters('dataCollectionRuleName'))]"
    }
  }
}

Logstash 構成ファイルを構成する

このプラグインでは、 サービス プリンシパル (クライアント資格情報) と マネージド ID (パスワードレス) の 2 つの認証方法がサポートされています。 環境に合った方法を選択します。

サービス プリンシパル認証

Logstashの設定ファイルをサービスプリンシパル認証でログをカスタムテーブルに取り込むように設定するには、 client_id、 client_secret、 tenant_id、 data_collection_endpoint、 dcr_id、 stream_nameの値を取得します。

必要な値を取得した後:

1. 前の手順で作成した Logstash 構成ファイルの出力セクションを次の例に置き換えます。
2. 次の例のプレースホルダー文字列を、取得した値に置き換えます。
3. create_sample_file属性を必ず false に変更してください。

例: サービス プリンシパル出力プラグインの構成

output {
    microsoft-sentinel-log-analytics-logstash-output-plugin {
      client_id => "<enter your client_id value here>"
      client_secret => "<enter your client_secret value here>"
      tenant_id => "<enter your tenant id here>"
      data_collection_endpoint => "<enter your logsIngestion URI here>"
      dcr_id => "<enter your DCR immutableId here>"
      stream_name => "<enter your stream name here>"
      create_sample_file=> false
      sample_file_path => "c:\\temp"
    }
}

マネージド ID 認証 (パスワードレス)

サービスプリンシパル認証情報(client_id、client_secret、tenant_id)を提供しない場合、プラグインはAzure SDKのDefaultAzureCredentialを使って認証します。 DefaultAzureCredential 一連の認証方法を試し、最初に成功したものを使用します。 サーバー環境では、関連するメソッドは以下の順序で試みられます。

1. 環境変数: AZURE_CLIENT_ID、 AZURE_TENANT_ID、 AZURE_CLIENT_SECRET などの環境変数から認証情報を読み込み、サービスプリンシパルとして認証します。
2. ワークロード識別:プラグインがワークロード識別が有効のホストAzure(例:AZURE_FEDERATED_TOKEN_FILE環境変数を設定したAKS)で動作する場合、プラグインはOIDCトークン交換を行います。
3. マネージドID:ホストがマネージドIDを有効にしている場合、プラグインはそのアイデンティティを使って認証します。 この方法はAzure VM、Virtual Machine Scale Sets、Azure Arc対応サーバーをカバーします。

DefaultAzureCredential が試行する資格情報チェーン全体については、Java 用 Azure Identity ライブラリの資格情報チェーンを参照してください。

マネージド ID に必要な構成:

例:マネージド・アイデンティティ

output {
    microsoft-sentinel-log-analytics-logstash-output-plugin {
      data_collection_endpoint => "<enter your DCE logsIngestion URI here>"
      dcr_id => "<enter your DCR immutableId here>"
      stream_name => "<enter your stream name here>"
    }
}

オプションの構成

Logstash を再起動する

更新された出力プラグイン構成で Logstash を再起動します。 DCR 構成に従って、データが正しいテーブルに取り込まれたことを確認します。

Microsoft Sentinelで受信ログを表示する

ログ データがワークスペースに到達することを確認するには、次の手順に従います。

1. メッセージが出力プラグインに送信されていることを確認します。

2. Microsoft Sentinel ナビゲーション メニューで、[ログ] を選択します。 [ テーブル ] 見出しで、[ カスタム ログ] カテゴリを展開します。 構成で指定したテーブルの名前 ( _CL サフィックス付き) を見つけて選択します。

   

3. テーブル内のレコードを表示するには、テーブル名をスキーマとして使用してテーブルにクエリを実行します。

   

出力プラグインの監査ログを監視する

Microsoft Sentinel出力プラグインの接続とアクティビティを監視するには、適切な Logstash ログ ファイルを有効にします。 ログ ファイルの場所については、 Logstash ディレクトリ レイアウト に関するドキュメントを参照してください。

このログ ファイルにデータが表示されない場合は、入力プラグインとフィルター プラグインを介してローカルでいくつかのイベントを生成して送信し、出力プラグインがデータを受信していることを確認します。 Microsoft Sentinelでは、出力プラグインに関連する問題のみがサポートされます。

ネットワーク セキュリティ

ネットワーク設定を定義し、Microsoft Sentinel Logstash 出力プラグインのネットワーク分離を有効にします。

仮想ネットワーク サービス タグ

Microsoft Sentinel出力プラグインでは、Azure仮想ネットワーク サービス タグがサポートされています。 AzureMonitor タグと AzureActiveDirectory タグの両方が必要です。

Azure Virtual Networkサービス タグを使用して、ネットワーク セキュリティ グループ、Azure Firewall、およびユーザー定義ルートに対するネットワーク アクセス制御を定義できます。 セキュリティ 規則とルートを作成するときは、特定の IP アドレスの代わりにサービス タグを使用します。 サービス タグAzure Virtual Network使用できないシナリオでは、ファイアウォールの要件を次に示します。

ファイアウォールの要件

次の表に、Azure仮想ネットワーク サービス タグを使用できないシナリオのファイアウォール要件を示します。

プラグインのバージョン履歴

2.5.0

• proxy、proxy_aad、proxy_endpointを用いた認証およびインジェストラフィック用のプラグインごとのオプションプロキシ設定を追加しました。
• Nettyハンドラー、HTTP、HTTP/2、DNSコンポーネントを4.1.133.Finalから4.1.136.Finalに更新しました。
• Jackson DatabindとJackson Coreを2.18.6から2.18.8に更新しました。

2.4.0

• ワーカースレッドは現在、バウンドされたエグゼキュータースケジューズパスとして実行されます。回復可能な例外はログに記録され、ワーカーは次のサイクルで再開されます。致命的なJVMエラーはログされ、再スローされます。
• ワーカーが停止する前に処理中のバッチがすべて完了するよう、正常終了処理を修正しました。停止順序はバッチャー、ユニファイアー、センダーで、処理時間の上限は max_graceful_shutdown_time_seconds で指定します。
• 設定可能なアップロードタイムアウト connect_timeout_seconds (デフォルト15)と write_timeout_seconds (デフォルト60)を追加しました。接続と書き込みタイムアウトは再試行されます。
• バッチ失敗ログにスレッドID、例外タイプ、バッチサイズ、DCRストリームを追加しました。

2.3.3

• 数値およびブール型の忠実度の固定損失:Logstashの内部JRuby型(例:ポート数やバイト数)に裏付けられたフィールドが、文字列に変換されるのではなく、ネイティブJSON数とブール値として保持され、型付きカラムでDCRへの信頼性の高い取り込みが保証されています。

2.3.2

• ワーカー処理ループの未検出例外によって引き起こされるサイレントワーカースレッドの死を修正しました。
• Azure が null の HTTP レスポンスを伴う LogsUploadException を返した場合に SenderWorker で発生する NullPointerException を修正しました。
• 連続的なエラートラッキングによるレジリエントなエラー処理を追加し、作業者の恒久的な失敗を減らしました。
• テレメトリ用のオプションの id 設定値が追加されました。
• 送信バッチログにDCRストリームを追加しました。

2.3.0

• Logstash 9.4で機能を有効にしました。
• 外部ライブラリ（azure-sdk-bom、logback、slf4j、Netty）の依存関係のバージョンを更新しました。

2.2.1

• バッチが正常に送信されると、情報レベルのログラインを追加します。

2.2.0

• 新旧の設定値のいずれかを使える機能が追加されました。

2.1.2

• ドキュメントの更新。

2.1.0

• イベント正規化を修正しました。

2.0.0

• Ruby から Java へのプラグインをリファクタリングしました。
• ManagedIdentity 認証を追加しました。
• GitHub から DevOps Azureにコードベースを移動しました。
• 閉じられたコードベース。

1.2.0

• Azure VM/VMSS (IMDS 経由でシステム割り当てとユーザー割り当て) のマネージド ID 認証のサポートを追加します。
• OIDC トークン交換を介して AKS ワークロード ID のサポートを追加します。
• ハイブリッド サーバーとオンプレミス サーバー Azure Arc マネージド ID のサポートを追加します。
• 環境 (ワークロード ID env vars、Arc エージェント、または IMDS フォールバック) に基づいて、実行時に認証方法を自動検出します。
• HTTP クライアントを excon から rest-client に移行し、JRuby プラグインと Logstash プラグインのエコシステムの互換性を向上させます。
• Azure Active Directory への参照を Microsoft Entra ID に変更します。

1.1.4

• プロキシ excon 使用するときにポートが常に使用されるように、ライブラリ バージョンを 1.0.0 未満に制限します。

1.1.3

• Azureへの接続に使用するrest-client ライブラリをexcon ライブラリに置き換えます。

1.1.1

• 21Vianet が中国で運用Azure米国政府クラウドと Microsoft Azureのサポートを追加します。

1.1.0

• API 接続に異なるプロキシ値を設定できます。
• ログ インジェスト API のバージョンを 2023-01-01 にアップグレードします。
• プラグインの名前を microsoft-sentinel-log-analytics-logstash-output-plugin に変更します。

1.0.0

• Microsoft Sentinelの Logstash 出力プラグインの最初のリリース。 このプラグインは、Azure Monitor のログ インジェスト API でデータ収集規則 (DCR) を使用します。

既知の問題

Lite Ubuntu の Docker イメージにインストールされている Logstash を使用すると、次の警告が表示されることがあります。

java.lang.RuntimeException: getprotobyname_r failed

このエラーを解決するには、 Dockerfile に netbase パッケージをインストールします。

USER root
RUN apt install netbase -y

詳細については、「 Logstash 7.17.0 (Docker)の JNR 回帰」を参照してください。

環境のイベント率が低い場合は、 max_waiting_time_for_batch_seconds と max_waiting_for_unifier_time_seconds の値を60以上に上げましょう。 インジェスト ペイロードは 、DCR メトリックを使用して監視できます。 待機時間変数の詳細については、 オプション設定 表をご覧ください。

制限事項

• 標準テーブルへのインジェストは、 カスタム ログ インジェストでサポートされている標準テーブルにのみ制限されます。

• streamDeclarations プロパティの入力ストリームの列は、文字で始まる必要があります。 他の文字 ( @ や _など) で列を開始すると、操作は失敗します。

• TimeGenerated datetime フィールドが必要です。 KQL 変換には、このフィールドを含める必要があります。

• 追加の問題については、 Logs Ingestion APIのトラブルシューティング手順を確認してください。