Azure Monitor でのデータ収集の変換

[アーティクル]
04/11/2024

Azure Monitor の変換を使用すると、Log Analytics ワークスペースに送信される前に、受信データをフィルター処理または変更できます。この記事では、変換とその実装方法についての基本的な説明を示します。変換を作成する方法について説明した、他のコンテンツへのリンクも記載しています。

変換は、データソースからデータが配信された後、データが宛先に送信されるまでの間に、Azure Monitor のデータインジェストパイプライン内で実行されます。データソースでは、データを送信する前に独自のフィルター処理を実行できますが、宛先に送信されるまでのさらなる操作については、変換に依存することになります。

変換は、データ収集ルール (DCR) で定義されます。変換には、受信データ内の各エントリに個別に適用される、Kusto 照会言語 (KQL) ステートメントが使用されます。受信データの形式を理解し、宛先で想定されている構造で出力を作成する必要があります。

次の図は、受信データの変換プロセスと、使用される可能性があるサンプルクエリを示しています。変換クエリ構築の詳細については、「Azure Monitor での変換の構造」を参照してください。

変換を使用する理由

次の表では、変換を使用して実現できるさまざまな目標について説明します。

カテゴリ	詳細
機密データを削除する	データソースによっては、プライバシーやコンプライアンス上の理由から、保存するのが望ましくない情報が送信される場合があります。機密情報をフィルター処理する。機密情報を含む行全体、または特定の列を除外できます。機密情報を難読化する。 IP アドレスや電話番号の数字を共通の文字に置き換えます。別のテーブルに送信する。ロールベースのアクセス制御構成が異なる別のテーブルに機密レコードを送信します。
詳細情報や計算された情報を使用してデータを強化する	変換を使用してデータに情報を追加することで、ビジネスコンテキストを提供したり、その後のデータクエリを簡略化したりすることができます。詳細情報を含む列を追加する。たとえば、別の列内の IP アドレスが内部のものか外部のものかを識別する列を追加することもできます。ビジネス固有の情報を追加する。たとえば、他の列内の位置情報に基づいて会社の部門を示す列を追加することもできます。
データコストを削減する	Log Analytics ワークスペースに送信されるデータについてはインジェストコストが発生するため、コストを削減するには、不要なデータを除外する必要があります。行全体を削除する。たとえば、特定のリソースからリソースログを収集する診断設定があるが、生成されるすべてのログエントリが必要であるわけではない場合があります。特定の条件に一致するレコードを除外する変換を作成します。各行から列を削除する。たとえば、データには、冗長なデータや最小値データを含んだ列が含まれている場合があります。そのような場合は、不要な列を除外する変換を作成できます。列内の重要なデータを解析する。貴重なデータが特定の列に埋もれているテーブルがある場合があります。変換を使用すると、重要なデータを解析して新しい列を生成し、元の列を削除できます。特定の行を基本ログに送信する。基本的なクエリ機能を必要とするデータ内の行を基本ログテーブルに送信することで、インジェストコストを削減できます。
変換先に合わせてデータをフォーマットする	変換先テーブルの構造と一致しない形式でデータがデータソースから送信されることがあります。変換を利用し、必須のスキーマにデータを再フォーマットします。

サポートするテーブル

変換で使用できるテーブルの一覧については、「Azure Monitor ログでの変換をサポートするテーブル」を参照してください。 Azure Monitor データ参照を使用することもできます。変換がサポートされるかなど、各テーブルの属性が一覧表示されています。これらのテーブルに加え、カスタムテーブル (サフィックス _CL) もサポートされています。

「Azure Monitor ログでの変換をサポートするテーブル」に記載されている任意の Azure テーブル。 Azure Monitor データ参照を使用することもできます。変換がサポートされるかなど、各テーブルの属性が一覧表示されています。
Azure Monitor エージェント用に作成された任意のカスタムテーブル (MMA カスタムテーブルでは変換を使用できません)。

変換を作成する

変換を作成する方法は、データ収集の方法に応じて複数あります。次の表に、変換を作成するためのさまざまな方法に関するガイダンスを示します。

データコレクション	リファレンス
ログインジェスト API	REST API (Azure portal) を使用して Azure Monitor ログにデータを送信する REST API (Azure Resource Manager テンプレート) を使用して Azure Monitor ログにデータを送信する
Azure Monitor エージェントを使用した仮想マシン	Azure Monitor ログに変換を追加する
Container Insights を使用した Kubernetes クラスター	Container Insights でのデータ変換
Azure Event Hubs	トリアル: Azure Event Hubs から Azure Monitor ログにイベントを取り込む (パブリックプレビュー)

複数の宛先

変換では、1 つの DCR を使用して、Log Analytics ワークスペース内の複数の宛先にデータを送信できます。宛先ごとに KQL クエリを指定すると、対応する場所に各クエリの結果が送信されます。異なるデータセットを異なるテーブルに送信したり、複数のクエリを使用して異なるデータセットを同じテーブルに送信したりできます。

たとえば、ログインジェスト API を使用して Azure Monitor にイベントデータを送信できます。ほとんどのイベントは分析テーブルに送信する必要があります。そうすることで、通常のクエリの対象となることができます。一方、監査イベントは、コストを削減するために、基本ログ用に構成されたカスタムテーブルに送信する必要があります。

複数の宛先を使用するには、現在のところ、新しい DCR を手動で作成するか、既存の DCR を編集する必要があります。複数の宛先を使用する DCR の例については、「サンプル」セクションを参照してください。

重要

現時点では、DCR 内のテーブルは同じ Log Analytics ワークスペースに存在する必要があります。 1 つのデータソースから複数のワークスペースに送信するには、複数の DCR を使用し、それぞれのワークスペースにデータを送信するようにアプリケーションを構成します。

変換を監視する

変換の正常性とパフォーマンスを監視するログとメトリックの詳細については、「Azure Monitor での DCR データ収集の監視とトラブルシューティング」を参照してください。これには、KQL で発生したエラーの特定と、その実行時間を追跡するためのメトリックの識別が含まれます。

変換のコスト

変換そのものには直接コストは発生しませんが、次のシナリオでは追加料金が発生する可能性があります。

たとえば計算列の追加のように、変換によって受信データのサイズが大きくなる場合は、その追加データに対して標準の取り込み料金が課金されます。
取り込まれたデータが変換によって 50% を超えて削減された場合、50% を超えるフィルター処理されたデータ量に対して課金されます。

変換によって得られるデータ処理料金を計算するには、次の数式を使用します。
[変換によって除外された GB] - ([パイプラインによって取り込まれた GB データ] / 2)。次の表に例を示します。

パイプラインによって取り込まれたデータ	変換によって削除されたデータ	Log Analytics ワークスペースによって取り込まれたデータ	データ処理料金	インジェスト料金
20 GB	12 GB	8 GB	2 GB ¹	8 GB
20 GB	8 GB	12 GB	0 GB	12 GB

¹ この料金では、Log Analytics ワークスペースによって取り込まれたデータの料金は除外されます。

この料金を回避するには、変換が適用される前に、他の方法を使用して、取り込まれたデータをフィルター処理する必要があります。これにより、変換によって処理されるデータの量を減らし、追加コストを最小限に抑えることができます。

Azure Monitor でのログデータの取り込みと保持に関する現在の料金については、「Azure Monitor の価格」を参照してください。

重要

Log Analytics ワークスペースに対して Azure Sentinel が有効になっている場合、変換によってフィルター処理されるデータの量に関係なく、フィルターインジェスト料金は発生しません。

サンプル

以下の Resource Manager テンプレートでは、さまざまなパターンのサンプル DCR を示します。これらのテンプレートは、独自のシナリオ向けの変換に合わせて DCR を作成するための出発点として使用できます。

単一の宛先

次の例は、Azure Monitor エージェントでデータを Syslog テーブルに送信する場合の DCR です。この例では、変換により、メッセージにerrorがあるレコードのデータがフィルター処理されます。

{ 
    "$schema": "https://schema.management.azure.com/schemas/2019-04-01/deploymentTemplate.json#",
    "contentVersion": "1.0.0.0",
    "resources" : [
        {
            "type": "Microsoft.Insights/dataCollectionRules", 
            "name": "singleDestinationDCR", 
            "apiVersion": "2021-09-01-preview", 
            "location": "eastus", 
            "properties": { 
              "dataSources": { 
                "syslog": [ 
                  { 
                    "name": "sysLogsDataSource", 
                    "streams": [ 
                      "Microsoft-Syslog" 
                    ], 
                    "facilityNames": [ 
                      "auth",
                      "authpriv",
                      "cron",
                      "daemon",
                      "mark",
                      "kern",
                      "mail",
                      "news",
                      "syslog",
                      "user",
                      "uucp"
                    ], 
                    "logLevels": [ 
                      "Debug", 
                      "Critical", 
                      "Emergency" 
                    ] 
                  } 
                ] 
              }, 
              "destinations": { 
                "logAnalytics": [ 
                  { 
                    "workspaceResourceId": "/subscriptions/xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx/resourceGroups/my-resource-group/providers/Microsoft.OperationalInsights/workspaces/my-workspace", 
                    "name": "centralWorkspace" 
                  } 
                ] 
              }, 
              "dataFlows": [ 
                { 
                  "streams": [ 
                    "Microsoft-Syslog" 
                  ], 
                  "transformKql": "source | where message has 'error'", 
                  "destinations": [ 
                    "centralWorkspace" 
                  ] 
                } 
              ] 
            }
        }
    ]
}

複数の Azure テーブル

次の例は、ログインジェスト API からのデータを Syslog と SecurityEvent テーブルの両方に送信する場合の DCR です。この DCR には、それぞれ異なる transformKql と OutputStream を持つ異なる dataFlow が必要です。この例では、すべての受信データが Syslog テーブルに送信され、悪意のあるデータは SecurityEvent テーブルにも送信されます。悪意のあるデータを両方のテーブルにレプリケートしたくない場合は、最初のクエリに where ステートメントを追加することで、それらのレコードを削除できます。

{ 
    "$schema": "https://schema.management.azure.com/schemas/2019-04-01/deploymentTemplate.json#",
    "contentVersion": "1.0.0.0",
    "resources" : [
        { 
            "type": "Microsoft.Insights/dataCollectionRules", 
            "name": "multiDestinationDCR", 
            "location": "eastus", 
            "apiVersion": "2021-09-01-preview", 
            "properties": { 
                "dataCollectionEndpointId": "/subscriptions/xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx/resourceGroups/my-resource-group/providers//Microsoft.Insights/dataCollectionEndpoints/my-dce",
                "streamDeclarations": { 
                    "Custom-MyTableRawData": { 
                        "columns": [ 
                            { 
                                "name": "Time", 
                                "type": "datetime" 
                            }, 
                            { 
                                "name": "Computer", 
                                "type": "string" 
                            }, 
                            { 
                                "name": "AdditionalContext", 
                                "type": "string" 
                            } 
                        ] 
                    } 
                }, 
                "destinations": { 
                    "logAnalytics": [ 
                        { 
                            "workspaceResourceId": "/subscriptions/xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx/resourceGroups/my-resource-group/providers/Microsoft.OperationalInsights/workspaces/my-workspace", 
                            "name": "clv2ws1" 
                        }, 
                    ] 
                }, 
                "dataFlows": [ 
                    { 
                        "streams": [ 
                            "Custom-MyTableRawData" 
                        ], 
                        "destinations": [ 
                            "clv2ws1" 
                        ], 
                        "transformKql": "source | project TimeGenerated = Time, Computer, Message = AdditionalContext", 
                        "outputStream": "Microsoft-Syslog" 
                    }, 
                    { 
                        "streams": [ 
                            "Custom-MyTableRawData" 
                        ], 
                        "destinations": [ 
                            "clv2ws1" 
                        ], 
                        "transformKql": "source | where (AdditionalContext has 'malicious traffic!' | project TimeGenerated = Time, Computer, Subject = AdditionalContext", 
                        "outputStream": "Microsoft-SecurityEvent" 
                    } 
                ] 
            } 
        }
    ]
}

Azure テーブルとカスタムテーブルの組み合わせ

以下は、ログインジェスト API からのデータを、それぞれ別の形式で Syslog テーブルとカスタムテーブルの両方に送信する場合の DCR の例です。この DCR には、それぞれ異なる transformKql と OutputStream を持つ異なる dataFlow が必要です。カスタムテーブルを使用する場合は、宛先 (カスタムテーブル) のスキーマに、送信するレコードのスキーマと一致するカスタム列 (カスタム列を追加または削除する方法) が含まれていることを確認することが重要です。たとえば、レコードに SyslogMessage というフィールドがあり、宛先のカスタムテーブルに TimeGenerated と RawData しかない場合は、TimeGenerated フィールドのみが設定されたカスタムテーブルでイベントを受け取り、RawData フィールドは空になります。 SyslogMessage フィールドは、宛先テーブルのスキーマに SyslogMessage という名前の文字列フィールドが含まれていないため、削除されます。

{ 
    "$schema": "https://schema.management.azure.com/schemas/2019-04-01/deploymentTemplate.json#",
    "contentVersion": "1.0.0.0",
    "resources" : [
        { 
            "type": "Microsoft.Insights/dataCollectionRules", 
            "name": "multiDestinationDCR", 
            "location": "eastus", 
            "apiVersion": "2021-09-01-preview", 
            "properties": { 
                "dataCollectionEndpointId": "/subscriptions/xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx/resourceGroups/my-resource-group/providers//Microsoft.Insights/dataCollectionEndpoints/my-dce",
                "streamDeclarations": { 
                    "Custom-MyTableRawData": { 
                        "columns": [ 
                            { 
                                "name": "Time", 
                                "type": "datetime" 
                            }, 
                            { 
                                "name": "Computer", 
                                "type": "string" 
                            }, 
                            { 
                                "name": "AdditionalContext", 
                                "type": "string" 
                            } 
                        ] 
                    } 
                }, 
                "destinations": { 
                    "logAnalytics": [ 
                        { 
                            "workspaceResourceId": "/subscriptions/xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx/resourceGroups/my-resource-group/providers/Microsoft.OperationalInsights/workspaces/my-workspace", 
                            "name": "clv2ws1" 
                        }, 
                    ] 
                }, 
                "dataFlows": [ 
                    { 
                        "streams": [ 
                            "Custom-MyTableRawData" 
                        ], 
                        "destinations": [ 
                            "clv2ws1" 
                        ], 
                        "transformKql": "source | project TimeGenerated = Time, Computer, SyslogMessage = AdditionalContext", 
                        "outputStream": "Microsoft-Syslog" 
                    }, 
                    { 
                        "streams": [ 
                            "Custom-MyTableRawData" 
                        ], 
                        "destinations": [ 
                            "clv2ws1" 
                        ], 
                        "transformKql": "source | extend jsonContext = parse_json(AdditionalContext) | project TimeGenerated = Time, Computer, AdditionalContext = jsonContext, ExtendedColumn=tostring(jsonContext.CounterName)", 
                        "outputStream": "Custom-MyTable_CL" 
                    } 
                ] 
            } 
        }
    ]
}

Azure Monitor でのデータ収集の変換

変換を使用する理由

サポートするテーブル

変換を作成する

複数の宛先

変換を監視する

変換のコスト

サンプル

単一の宛先

複数の Azure テーブル

Azure テーブルとカスタムテーブルの組み合わせ

次のステップ

その他のリソース

Azure Monitor でのデータ収集の変換

変換を使用する理由

サポートするテーブル

変換を作成する

複数の宛先

変換を監視する

変換のコスト

サンプル

単一の宛先

複数の Azure テーブル

Azure テーブルとカスタム テーブルの組み合わせ

次のステップ

その他のリソース

Azure テーブルとカスタムテーブルの組み合わせ