Azure Monitor의 데이터 수집 변환

아티클
04/10/2024

Azure Monitor의 변환을 사용하면 들어오는 데이터를 Log Analytics 작업 영역으로 보내기 전에 필터링하거나 수정할 수 있습니다. 이 문서에서는 변환과 구현 방법에 대한 기본 설명을 제공합니다. 변형을 만들기 위한 다른 콘텐츠에 대한 링크를 제공합니다.

변환은 데이터 원본이 데이터를 전달한 후 대상으로 보내기 전에 데이터 수집 파이프라인의 Azure Monitor에서 수행됩니다. 데이터 원본은 데이터를 보내기 전에 자체 필터링을 수행할 수 있지만 대상으로 보내기 전에 추가 조작을 위해 변환에 의존할 수 있습니다.

변환은 DCR(데이터 수집 규칙)에 정의되어 있으며 수신 데이터의 각 항목에 개별적으로 적용되는 KQL(Kusto 쿼리 언어) 문을 사용합니다. 들어오는 데이터의 형식을 이해하고 대상에서 예상하는 구조로 출력을 만들어야 합니다.

다음 다이어그램은 들어오는 데이터에 대한 변환 프로세스와 사용할 수 있는 샘플 쿼리를 보여 줍니다. 변환 쿼리 작성에 대한 자세한 내용은 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 클러스터	컨테이너 인사이트의 데이터 변환
Azure Event Hubs	자습서: Azure Event Hubs의 이벤트를 Azure Monitor 로그로 수집(공개 미리 보기)

여러 대상

변환을 사용하면 단일 DCR을 사용하여 Log Analytics 작업 영역의 여러 대상으로 데이터를 보낼 수 있습니다. 각 대상에 대해 KQL 쿼리를 제공하면 각 쿼리의 결과가 해당 위치로 전송됩니다. 서로 다른 데이터 집합을 서로 다른 테이블로 보내거나 여러 쿼리를 사용하여 서로 다른 데이터 집합을 동일한 테이블로 보낼 수 있습니다.

예를 들어, 로그 수집 API를 사용하여 이벤트 데이터를 Azure Monitor로 보낼 수 있습니다. 대부분의 이벤트는 정기적으로 쿼리할 수 있는 분석 테이블로 전송되어야 하며, 감사 이벤트는 비용 절감을 위해 기본 로그용으로 구성된 사용자 지정 표로 전송되어야 합니다.

여러 대상을 사용하려면 현재 수동으로 새 DCR을 만들거나 기존 DCR을 편집해야 합니다. 여러 대상을 사용하는 DCR의 예는 샘플 섹션을 참조하세요.

Important

현재 DCR의 테이블은 동일한 Log Analytics 작업 영역에 있어야 합니다. 단일 데이터 원본에서 여러 작업 영역으로 보내려면 여러 DCR을 사용하고 각각에 데이터를 보내도록 애플리케이션을 구성합니다.

변환 모니터링

변환의 상태 및 성능을 모니터링하는 로그 및 메트릭에 대한 자세한 내용은 Azure Monitor에서 DCR 데이터 수집 모니터링 및 문제 해결을 참조하세요. 여기에는 KQL 및 메트릭에서 발생하는 오류를 식별하여 실행 기간을 추적하는 것이 포함됩니다.

변환 비용

변환 자체에는 직접 요금이 발생하지 않지만 다음 시나리오에서는 추가 요금이 발생할 수 있습니다.

변환으로 인해 계산 열을 추가하는 등 수신 데이터의 크기가 증가하면 추가 데이터에 대한 표준 수집 요금이 청구됩니다.
변환으로 인해 수집된 데이터가 50% 이상 감소하면 50%를 초과하는 필터링된 데이터 양에 대해 요금이 청구됩니다.

변환으로 인한 데이터 처리 요금을 계산하려면 다음 수식을 사용합니다.
[변환에 의해 필터링된 GB] - ([파이프라인에서 수집된 GB 데이터] / 2) 다음 표에서는 두 가지 예제를 보여 줍니다.

파이프라인에서 수집된 데이터	변환으로 삭제된 데이터	Log Analytics 작업 영역에서 수집된 데이터	데이터 처리 요금	수집 요금
20GB	12GB	8GB	2GB ¹	8GB
20GB	8GB	12GB	0GB	12GB

¹ 이 요금은 Log Analytics 작업 영역에서 수집한 데이터에 대한 요금을 제외합니다.

이 요금을 피하려면 변환을 적용하기 전에 대체 방법을 사용하여 수집한 데이터를 필터링하는 것이 좋습니다. 이렇게 하면 변환으로 처리된 데이터의 양을 줄일 수 있으므로 추가 비용을 최소화할 수 있습니다.

Azure Monitor에서 로그 데이터 수집 및 보관에 대한 현재 요금은 Azure Monitor 가격 책정을 참조하세요.

Important

Log Analytics 작업 영역에 대해 Azure Sentinel을 사용하도록 설정한 경우 변환이 필터링하는 데이터 양에 관계없이 필터링 수집 요금이 청구되지 않습니다.

샘플

다음 Resource Manager 템플릿은 패턴이 다른 샘플 DCR을 보여 줍니다. 이러한 템플릿을 고유의 시나리오에 대한 변환이 포함된 DCR 만들기의 시작점으로 사용할 수 있습니다.

단일 대상

다음 예에서는 데이터를 Syslog 테이블로 전송하는 Azure Monitor 에이전트용 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 테이블

다음 예에서는 Syslog 및 SecurityEvent 테이블 모두에 데이터를 전송하는 로그 수집 API의 데이터에 대한 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와 사용자 지정 표의 조합

다음 예에서는 다른 형식의 데이터가 있는 Syslog 테이블과 사용자 지정 표 모두에 데이터를 전송하는 로그 수집 API의 데이터에 대한 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" 
                    } 
                ] 
            } 
        }
    ]
}