Osvědčené postupy a ukázky pro transformace ve službě Azure Monitor
Transformace ve službě Azure Monitor umožňují filtrovat nebo upravovat příchozí data před jejich odesláním do pracovního prostoru služby Log Analytics. Tento článek obsahuje osvědčené postupy a doporučení pro použití transformací, které zajistí jejich spolehlivost a nákladově efektivní použití. Obsahuje také ukázky pro běžné scénáře, které můžete použít k zahájení vytváření vlastní transformace.
Monitorování transformací
Protože transformace neběží interaktivně, je důležité je monitorovat, aby se zajistilo, že jsou správně spuštěné a nezbírají příliš dlouho zpracování dat. Podrobnosti o protokolech a metrikách, které monitorují stav a výkon transformací, najdete v tématu Monitorování a řešení potíží se shromažďováním dat DCR ve službě Azure Monitor . To zahrnuje identifikaci všech chyb, ke kterým dochází v KQL, a metriky ke sledování jejich doby trvání.
Důležité
Transformace, které trvá příliš dlouho, můžou ovlivnit výkon kanálu shromažďování dat a vést ke ztrátě dat. Pokud vidíte, že transformace trvá příliš dlouho, zvažte přepsání dotazu, abyste optimalizovali jeho výkon.
Následující metriky se automaticky shromažďují pro transformace a měly by se pravidelně kontrolovat, abyste ověřili, že vaše transformace stále běží podle očekávání. Můžete také povolit protokoly chyb DCR ke sledování chyb, ke kterým dochází v transformacích nebo jiných dotazech.
- Doba trvání transformace protokolů za minutu
- Chyby transformace protokolů za minutu
Odeslání dat do jednoho cíle
Následující příklad je DCR pro agenta Azure Monitoru, který odesílá data do Syslog tabulky. V tomto příkladu transformace filtruje data pro záznamy se error zprávou.
{
"$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"
]
}
]
}
}
]
}
Odesílání dat do více cílů
Pomocí transformací můžete odesílat data do více cílů v pracovním prostoru služby Log Analytics pomocí jediného řadiče domény. Zadáte dotaz KQL pro každý cíl a výsledky každého dotazu se odešlou do odpovídajícího umístění. Do různých tabulek můžete odesílat různé sady dat nebo pomocí více dotazů odesílat různé sady dat do stejné tabulky. Pokud chcete použít více cílů, musíte v současné době buď ručně vytvořit nový řadič domény, nebo upravit existující.
Můžete například odesílat data událostí do služby Azure Monitor pomocí rozhraní API pro příjem protokolů. Většina událostí by měla být odeslána analytickou tabulku, kde by se mohla pravidelně dotazovat, zatímco události auditu by se měly odesílat do vlastní tabulky nakonfigurované pro základní protokoly, aby se snížily náklady.
Důležité
V současné době musí být tabulky v DCR ve stejném pracovním prostoru služby Log Analytics. Pokud chcete odesílat do více pracovních prostorů z jednoho zdroje dat, použijte více řadičů domény a nakonfigurujte aplikaci tak, aby do každého z nich odesílala data.
Následující příklad je DCR pro data z rozhraní API pro příjem protokolů, která odesílá data do tabulek i SecurityEvent do Syslog tabulek. Tento řadič domény vyžaduje pro dataFlow každou samostatnou a pro každou jinou transformKql OutputStream . V tomto příkladu Syslog se do tabulky odesílají všechna příchozí data, zatímco do SecurityEvent tabulky se odesílají také škodlivá data. Pokud jste nechtěli replikovat škodlivá data v obou tabulkách, můžete přidat where příkaz pro první dotaz, který tyto záznamy odebere.
{
"$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"
}
]
}
}
]
}
Kombinace Azure a vlastních tabulek
Následující příklad je DCR pro data z rozhraní API pro příjem protokolů, která odesílá data do Syslog tabulky i vlastní tabulky s daty v jiném formátu. Tento řadič domény vyžaduje pro dataFlow každou samostatnou a pro každou jinou transformKql OutputStream . Při použití vlastních tabulek je důležité zajistit, aby schéma cíle (vlastní tabulka) obsahovalo vlastní sloupce (přidání nebo odstranění vlastních sloupců), které odpovídají schématu odesílaných záznamů. Pokud má například záznam pole s názvem SyslogMessage, ale cílová vlastní tabulka obsahuje pouze TimeGenerated a RawData, obdržíte ve vlastní tabulce událost s vyplněným pouze polem TimeGenerated a pole RawData bude prázdné. Pole SyslogMessage se zahodí, protože schéma cílové tabulky neobsahuje řetězcové pole s názvem 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"
}
]
}
}
]
}
Analýza dat
Běžným použitím transformací je analýza příchozích dat do více sloupců tak, aby odpovídala schématu cílové tabulky. Můžete například shromažďovat položky ze souboru protokolu, který není ve strukturovaném formátu, a je třeba analyzovat data do sloupců tabulky. Pomocí operátoru parse v KQL extrahujte data do sloupců, ale dávejte pozor, abyste omezili počet sloupců, které extrahujete v jednom příkazu. Nadměrný počet extrakcí v jednom příkazu může způsobit výrazně vyšší dobu zpracování. Místo toho rozdělte extrakce do více parse příkazů.
Následující dotaz například extrahuje deset polí ze položky protokolu pomocí jednoho příkazu parsování.
source
| parse Message with
* "field1=" Field1: string
" field2=" Field2: string
" field3=" Field3: string
" field4=" Field4: string
" field5=" Field5: string
" field6=" Field6: string
" field7=" Field7: string
" field8=" Field8: string
" field9=" Field9: string
" field10=" Field10: string *
Výše uvedený příkaz lze přepsat na následující příkaz, aby se extrakce rozdělily na dva příkazy.
source
| parse Message with
* "field1=" Field1: string
" field2=" Field2: string
" field3=" Field3: string
" field4=" Field4: string
" field5=" Field5: string *
| parse Message with
* " field6=" Field6: string
" field7=" Field7: string
" field8=" Field8: string
" field9=" Field9: string
" field10=" Field10: string *