Tutorial: Ingest and query monitoring data in Azure Data Explorer

Diagnostic metrics example

Diagnostic metrics are aggregated with a time grain of 1 minute. Following is an example of an Azure Data Explorer metric-event schema on query duration:

{
    "records": [
    {
        "count": 14,
        "total": 0,
        "minimum": 0,
        "maximum": 0,
        "average": 0,
        "resourceId": "/SUBSCRIPTIONS/<subscriptionID>/RESOURCEGROUPS/<resource-group>/PROVIDERS/MICROSOFT.KUSTO/CLUSTERS/<cluster-name>",
        "time": "2018-12-20T17:00:00.0000000Z",
        "metricName": "QueryDuration",
        "timeGrain": "PT1M"
    },
    {
        "count": 12,
        "total": 0,
        "minimum": 0,
        "maximum": 0,
        "average": 0,
        "resourceId": "/SUBSCRIPTIONS/<subscriptionID>/RESOURCEGROUPS/<resource-group>/PROVIDERS/MICROSOFT.KUSTO/CLUSTERS/<cluster-name>",
        "time": "2018-12-21T17:00:00.0000000Z",
        "metricName": "QueryDuration",
        "timeGrain": "PT1M"
    }
    ]
}

Diagnostic logs example

Following is an example of an Azure Data Explorer diagnostic ingestion log:

{
    "time": "2019-08-26T13:22:36.8804326Z",
    "resourceId": "/SUBSCRIPTIONS/<subscriptionID>/RESOURCEGROUPS/<resource-group>/PROVIDERS/MICROSOFT.KUSTO/CLUSTERS/<cluster-name>",
    "operationName": "MICROSOFT.KUSTO/CLUSTERS/INGEST/ACTION",
    "operationVersion": "1.0",
    "category": "FailedIngestion",
    "resultType": "Failed",
    "correlationId": "d59882f1-ad64-4fc4-b2ef-d663b6cc1cc5",
    "properties": {
        "OperationId": "00000000-0000-0000-0000-000000000000",
        "Database": "Kusto",
        "Table": "Table_13_20_prod",
        "FailedOn": "2019-08-26T13:22:36.8804326Z",
        "IngestionSourceId": "d59882f1-ad64-4fc4-b2ef-d663b6cc1cc5",
        "Details":
        {
            "error":
            {
                "code": "BadRequest_DatabaseNotExist",
                "message": "Request is invalid and cannot be executed.",
                "@type": "Kusto.Data.Exceptions.DatabaseNotFoundException",
                "@message": "Database 'Kusto' was not found.",
                "@context":
                {
                    "timestamp": "2019-08-26T13:22:36.7179157Z",
                    "serviceAlias": "<cluster-name>",
                    "machineName": "KEngine000001",
                    "processName": "Kusto.WinSvc.Svc",
                    "processId": 5336,
                    "threadId": 6528,
                    "appDomainName": "Kusto.WinSvc.Svc.exe",
                    "clientRequestd": "DM.IngestionExecutor;a70ddfdc-b471-4fc7-beac-bb0f6e569fe8",
                    "activityId": "f13e7718-1153-4e65-bf82-8583d712976f",
                    "subActivityId": "2cdad9d0-737b-4c69-ac9a-22cf9af0c41b",
                    "activityType": "DN.AdminCommand.DataIngestPullCommand",
                    "parentActivityId": "2f65e533-a364-44dd-8d45-d97460fb5795",
                    "activityStack": "(Activity stack: CRID=DM.IngestionExecutor;a70ddfdc-b471-4fc7-beac-bb0f6e569fe8 ARID=f13e7718-1153-4e65-bf82-8583d712976f > DN.Admin.Client.ExecuteControlCommand/5b764b32-6017-44a2-89e7-860eda515d40 > P.WCF.Service.ExecuteControlCommandInternal..IAdminClientServiceCommunicationContract/c2ef9344-069d-44c4-88b1-a3570697ec77 > DN.FE.ExecuteControlCommand/2f65e533-a364-44dd-8d45-d97460fb5795 > DN.AdminCommand.DataIngestPullCommand/2cdad9d0-737b-4c69-ac9a-22cf9af0c41b)"
                },
                "@permanent": true
            }
        },
        "ErrorCode": "BadRequest_DatabaseNotExist",
        "FailureStatus": "Permanent",
        "RootActivityId": "00000000-0000-0000-0000-000000000000",
        "OriginatesFromUpdatePolicy": false,
        "ShouldRetry": false,
        "IngestionSourcePath": "https://c0skstrldkereneus01.blob.core.chinacloudapi.cn/aam-20190826-temp-e5c334ee145d4b43a3a2d3a96fbac1df/3216_test_3_columns_invalid_8f57f0d161ed4a8c903c6d1073005732_59951f9ca5d143b6bdefe52fa381a8ca.zip"
    }
}

Activity logs example

Azure activity logs are subscription-level logs that provide insight into the operations performed on resources in your subscription. Following is an example of an activity-log event for checking access:

{
    "records": [
    {
        "time": "2018-12-26T16:23:06.1090193Z",
        "resourceId": "/SUBSCRIPTIONS/<subscriptionID>/RESOURCEGROUPS/<resource-group>/PROVIDERS/MICROSOFT.WEB/SITES/CLNB5F73B70-DCA2-47C2-BB24-77B1A2CAAB4D/PROVIDERS/MICROSOFT.AUTHORIZATION",
        "operationName": "MICROSOFT.AUTHORIZATION/CHECKACCESS/ACTION",
        "category": "Action",
        "resultType": "Start",
        "resultSignature": "Started.",
        "durationMs": 0,
        "callerIpAddress": "13.66.225.188",
        "correlationId": "0de9f4bc-4adc-4209-a774-1b4f4ae573ed",
        "identity": {
            "authorization": {
                ...
            },
            "claims": {
                ...
            }
        },
        "level": "Information",
        "location": "global",
        "properties": {
            ...
        }
    },
    {
        "time": "2018-12-26T16:23:06.3040244Z",
        "resourceId": "/SUBSCRIPTIONS/<subscriptionID>/RESOURCEGROUPS/<resource-group>/PROVIDERS/MICROSOFT.WEB/SITES/CLNB5F73B70-DCA2-47C2-BB24-77B1A2CAAB4D/PROVIDERS/MICROSOFT.AUTHORIZATION",
        "operationName": "MICROSOFT.AUTHORIZATION/CHECKACCESS/ACTION",
        "category": "Action",
        "resultType": "Success",
        "resultSignature": "Succeeded.OK",
        "durationMs": 194,
        "callerIpAddress": "13.66.225.188",
        "correlationId": "0de9f4bc-4adc-4209-a774-1b4f4ae573ed",
        "identity": {
            "authorization": {
                ...
            },
            "claims": {
                ...
            }
        },
        "level": "Information",
        "location": "global",
        "properties": {
            "statusCode": "OK",
            "serviceRequestId": "87acdebc-945f-4c0c-b931-03050e085626"
        }
    }]
}

Create tables for the diagnostic metrics

1. In the TestDatabase database, create a table named DiagnosticMetrics to store the diagnostic metrics records. Use the following .create table management command:

   .create table DiagnosticMetrics (Timestamp:datetime, ResourceId:string, MetricName:string, Count:int, Total:double, Minimum:double, Maximum:double, Average:double, TimeGrain:string)
   

2. Select Run to create the table.

   

3. Create the intermediate data table named DiagnosticRawRecords in the TestDatabase database for data manipulation using the following query. Select Run to create the table.

   .create table DiagnosticRawRecords (Records:dynamic)
   

4. Set zero retention policy for the intermediate table:

   .alter-merge table DiagnosticRawRecords policy retention softdelete = 0d
   

Create tables for the diagnostic logs

1. In the TestDatabase database, create a table named DiagnosticLogs to store the diagnostic log records. Use the following .create table management command:

   .create table DiagnosticLogs (Timestamp:datetime, ResourceId:string, OperationName:string, Result:string, OperationId:string, Database:string, Table:string, IngestionSourceId:string, IngestionSourcePath:string, RootActivityId:string, ErrorCode:string, FailureStatus:string, Details:string)
   

2. Select Run to create the table.

3. Create the intermediate data table named DiagnosticRawRecords in the TestDatabase database for data manipulation using the following query. Select Run to create the table.

   .create table DiagnosticRawRecords (Records:dynamic)
   

4. Set zero retention policy for the intermediate table:

   .alter-merge table DiagnosticRawRecords policy retention softdelete = 0d
   

Create tables for the activity logs

1. Create a table named ActivityLogs in the TestDatabase database to receive activity log records. To create the table, run the following Azure Data Explorer query:

   .create table ActivityLogs (Timestamp:datetime, ResourceId:string, OperationName:string, Category:string, ResultType:string, ResultSignature:string, DurationMs:int, IdentityAuthorization:dynamic, IdentityClaims:dynamic, Location:string, Level:string)
   

2. Create the intermediate data table named ActivityLogsRawRecords in the TestDatabase database for data manipulation:

   .create table ActivityLogsRawRecords (Records:dynamic)
   

3. Set zero retention policy for the intermediate table:

   .alter-merge table ActivityLogsRawRecords policy retention softdelete = 0d
   

Map diagnostic metrics and logs to the table

To map the diagnostic metric and log data to the table, use the following query:

.create table DiagnosticRawRecords ingestion json mapping 'DiagnosticRawRecordsMapping' '[{"column":"Records","Properties":{"path":"$.records"}}]'

Map activity logs to the table

To map the activity log data to the table, use the following query:

.create table ActivityLogsRawRecords ingestion json mapping 'ActivityLogsRawRecordsMapping' '[{"column":"Records","Properties":{"path":"$.records"}}]'

Create data update policy for diagnostics metrics

1. Create a function that expands the collection of diagnostic metric records so that each value in the collection receives a separate row. Use the mv-expand operator:

   .create function DiagnosticMetricsExpand() {
      DiagnosticRawRecords
      | mv-expand events = Records
      | where isnotempty(events.metricName)
      | project
          Timestamp = todatetime(events['time']),
          ResourceId = tostring(events.resourceId),
          MetricName = tostring(events.metricName),
          Count = toint(events['count']),
          Total = todouble(events.total),
          Minimum = todouble(events.minimum),
          Maximum = todouble(events.maximum),
          Average = todouble(events.average),
          TimeGrain = tostring(events.timeGrain)
   }
   

2. Add the update policy to the target table. This policy will automatically run the query on any newly ingested data in the DiagnosticRawRecords intermediate data table and ingest its results into the DiagnosticMetrics table:

   .alter table DiagnosticMetrics policy update @'[{"Source": "DiagnosticRawRecords", "Query": "DiagnosticMetricsExpand()", "IsEnabled": "True", "IsTransactional": true}]'
   

Create data update policy for diagnostics logs

1. Create a function that expands the collection of diagnostic logs records so that each value in the collection receives a separate row. You'll enable ingestion logs on an Azure Data Explorer cluster, and use ingestion logs schema. You'll create one table for succeeded and for failed ingestion, while some of the fields will be empty for succeeded ingestion (ErrorCode for example). Use the mv-expand operator:

   .create function DiagnosticLogsExpand() {
       DiagnosticRawRecords
       | mv-expand events = Records
       | where isnotempty(events.operationName)
       | project
           Timestamp = todatetime(events['time']),
           ResourceId = tostring(events.resourceId),
           OperationName = tostring(events.operationName),
           Result = tostring(events.resultType),
           OperationId = tostring(events.properties.OperationId),
           Database = tostring(events.properties.Database),
           Table = tostring(events.properties.Table),
           IngestionSourceId = tostring(events.properties.IngestionSourceId),
           IngestionSourcePath = tostring(events.properties.IngestionSourcePath),
           RootActivityId = tostring(events.properties.RootActivityId),
           ErrorCode = tostring(events.properties.ErrorCode),
           FailureStatus = tostring(events.properties.FailureStatus),
           Details = tostring(events.properties.Details)
   }
   

2. Add the update policy to the target table. This policy will automatically run the query on any newly ingested data in the DiagnosticRawRecords intermediate data table and ingest its results into the DiagnosticLogs table:

   .alter table DiagnosticLogs policy update @'[{"Source": "DiagnosticRawRecords", "Query": "DiagnosticLogsExpand()", "IsEnabled": "True", "IsTransactional": true}]'
   

Create data update policy for activity logs

1. Create a function that expands the collection of activity log records so that each value in the collection receives a separate row. Use the mv-expand operator:

   .create function ActivityLogRecordsExpand() {
       ActivityLogsRawRecords
       | mv-expand events = Records
       | project
           Timestamp = todatetime(events['time']),
           ResourceId = tostring(events.resourceId),
           OperationName = tostring(events.operationName),
           Category = tostring(events.category),
           ResultType = tostring(events.resultType),
           ResultSignature = tostring(events.resultSignature),
           DurationMs = toint(events.durationMs),
           IdentityAuthorization = events.identity.authorization,
           IdentityClaims = events.identity.claims,
           Location = tostring(events.location),
           Level = tostring(events.level)
   }
   

2. Add the update policy to the target table. This policy will automatically run the query on any newly ingested data in the ActivityLogsRawRecords intermediate data table and ingest its results into the ActivityLogs table:

   .alter table ActivityLogs policy update @'[{"Source": "ActivityLogsRawRecords", "Query": "ActivityLogRecordsExpand()", "IsEnabled": "True", "IsTransactional": true}]'
   

Connect diagnostic metrics and logs to your event hub

Select a resource from which to export metrics. Several resource types support exporting diagnostic data, including event hubs namespace, Azure Key Vault, Azure IoT Hub, and Azure Data Explorer clusters. In this tutorial, we'll use an Azure Data Explorer cluster as our resource, we'll review query performance metrics and ingestion results logs.

1. Select your Kusto cluster in the Azure portal.

2. Select Diagnostic settings, and then select the Turn on diagnostics link.

   

3. The Diagnostics settings pane opens. Take the following steps:

   1. Give your diagnostics log data the name ADXExportedData.

   2. Under LOG, select both SucceededIngestion and FailedIngestion check boxes.

   3. Under METRIC, select the Query performance check box.

   4. Select the Stream to an event hub check box.

   5. Select Configure.

      

4. In the Select event hub pane, configure how to export data from diagnostic logs to the event hub you created:

   1. In the Select event hub namespace list, select AzureMonitoringData.
   2. In the Select event hub name list, select DiagnosticData.
   3. In the Select event hub policy name list, select RootManagerSharedAccessKey.
   4. Select OK.

5. Select Save.

Connect activity logs to your event hub

1. In the left menu of the Azure portal, select Activity log.

2. The Activity log window opens. Select Diagnostics settings.

   

3. The Diagnostics settings window opens. Select + Add diagnostic setting.

   

4. A new Diagnostic setting window opens.

   

   Do the following steps:

   1. Enter a name in the Diagnostic setting name field.
   2. On the left-hand side of check boxes, select the platform log(s) you wish to collect from a subscription.
   3. Select the Stream to an event hub check box.
   4. Select your subscription.
   5. In the Event hub namespace list, select AzureMonitoringData.
   6. Optionally, select your Event hub name.
   7. In the Event hub policy name list, select the default event hub policy name.
   8. In the upper-left corner of the window, select Save. An event hub with the name insights-operational-logs will be created (unless you've selected an Event hub name above).

1. Use the following settings in the Data Connection window:

   Data source:

   Setting	Suggested value	Field description
   Data connection name	DiagnosticsLogsConnection	The name of the connection you want to create in Azure Data Explorer.
   Event hub namespace	AzureMonitoringData	The name you chose earlier that identifies your namespace.
   Event hub	DiagnosticData	The event hub you created.
   Consumer group	adxpipeline	The consumer group defined in the event hub you created.

   Target table:

   There are two options for routing: static and dynamic. For this tutorial, you'll use static routing (the default), where you specify the table name, the data format, and the mapping. Leave My data includes routing info unselected.

   Setting	Suggested value	Field description
   Table	DiagnosticRawRecords	The table you created in the TestDatabase database.
   Data format	JSON	The format used in the table.
   Column mapping	DiagnosticRawRecordsMapping	The mapping you created in the TestDatabase database, which maps incoming JSON data to the column names and data types of the DiagnosticRawRecords table.

2. Select Create.

1. Use the following settings in the Data Connection window:

   Data source:

   Setting	Suggested value	Field description
   Data connection name	ActivityLogsConnection	The name of the connection you want to create in Azure Data Explorer.
   Event hub namespace	AzureMonitoringData	The name you chose earlier that identifies your namespace.
   Event hub	insights-operational-logs	The event hub you created.
   Consumer group	$Default	The default consumer group. If needed, you can create a different consumer group.

   Target table:

   There are two options for routing: static and dynamic. For this tutorial, you'll use static routing (the default), where you specify the table name, data format, and mapping. Leave My data includes routing info unselected.

   Setting	Suggested value	Field description
   Table	ActivityLogsRawRecords	The table you created in the TestDatabase database.
   Data format	JSON	The format used in the table.
   Column mapping	ActivityLogsRawRecordsMapping	The mapping you created in the TestDatabase database, which maps incoming JSON data to the column names and data types of the ActivityLogsRawRecords table.

2. Select Create.

Query the diagnostic metrics table

The following query analyzes query duration data from diagnostic metric records in Azure Data Explorer:

DiagnosticMetrics
| where Timestamp > ago(15m) and MetricName == 'QueryDuration'
| summarize avg(Average)

Query results:

Query the diagnostic logs table

This pipeline produces ingestions via an event hub. You'll review the results of these ingestions. The following query analyzes how many ingestions accrued in a minute, including a sample of Database, Table and IngestionSourcePath for each interval:

DiagnosticLogs
| where Timestamp > ago(15m) and OperationName has 'INGEST'
| summarize count(), take_any(Database, Table, IngestionSourcePath) by bin(Timestamp, 1m)

Query results:

Query the activity logs table

The following query analyzes data from activity log records in Azure Data Explorer:

ActivityLogs
| where OperationName == 'MICROSOFT.EVENTHUB/NAMESPACES/AUTHORIZATIONRULES/LISTKEYS/ACTION'
| where ResultType == 'Success'
| summarize avg(DurationMs)

Query results: