Share via


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

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

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

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

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

受信データのインジェスト時の変換を示す図。

変換を使用する理由

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

カテゴリ 詳細
機密データを削除する データ ソースによっては、プライバシーやコンプライアンス上の理由から、保存するのが望ましくない情報が送信される場合があります。

機密情報をフィルター処理する。 機密情報を含む行全体、または特定の列を除外できます。

機密情報を難読化する。 IP アドレスや電話番号の数字を共通の文字に置き換えます。

別のテーブルに送信する。 ロールベースのアクセス制御構成が異なる別のテーブルに機密レコードを送信します。
詳細情報や計算された情報を使用してデータを強化する 変換を使用してデータに情報を追加することで、ビジネス コンテキストを提供したり、その後のデータ クエリを簡略化したりすることができます。

詳細情報を含む列を追加する。 たとえば、別の列内の IP アドレスが内部のものか外部のものかを識別する列を追加することもできます。

ビジネス固有の情報を追加する。 たとえば、他の列内の位置情報に基づいて会社の部門を示す列を追加することもできます。
データ コストを削減する Log Analytics ワークスペースに送信されるデータについてはインジェスト コストが発生するため、コストを削減するには、不要なデータを除外する必要があります。

行全体を削除する。 たとえば、特定のリソースからリソース ログを収集する診断設定があるが、生成されるすべてのログ エントリが必要であるわけではない場合があります。 特定の条件に一致するレコードを除外する変換を作成します。

各行から列を削除する。 たとえば、データには、冗長なデータや最小値データを含んだ列が含まれている場合があります。 そのような場合は、不要な列を除外する変換を作成できます。

列内の重要なデータを解析する。 貴重なデータが特定の列に埋もれているテーブルがある場合があります。 変換を使用すると、重要なデータを解析して新しい列を生成し、元の列を削除できます。

特定の行を基本ログに送信する。 基本的なクエリ機能を必要とするデータ内の行を基本ログ テーブルに送信することで、インジェスト コストを削減できます。
変換先に合わせてデータをフォーマットする 変換先テーブルの構造と一致しない形式でデータがデータ ソースから送信されることがあります。 変換を利用し、必須のスキーマにデータを再フォーマットします。

サポートするテーブル

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

変換を作成する

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

データ コレクション リファレンス
ログ インジェスト 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 1 8 GB
20 GB 8 GB 12 GB 0 GB 12 GB

1 この料金では、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 からのデータを SyslogSecurityEvent テーブルの両方に送信する場合の DCR です。 この DCR には、それぞれ異なる transformKqlOutputStream を持つ異なる 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 には、それぞれ異なる transformKqlOutputStream を持つ異なる 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" 
                    } 
                ] 
            } 
        }
    ]
}

次のステップ