Measuring latency between availability group nodes in real-time We have a monitor in which we would like to display the replication latency between primary and secondary node in milliseconds in real-time.

We're using this query:

;WITH 
    AG_Stats AS 
            (
            SELECT AR.replica_server_name,
                   HARS.role_desc, 
                   Db_name(DRS.database_id) [DBName], 
                   DRS.last_commit_time
            FROM   sys.dm_hadr_database_replica_states DRS 
            INNER JOIN sys.availability_replicas AR ON DRS.replica_id = AR.replica_id 
            INNER JOIN sys.dm_hadr_availability_replica_states HARS ON AR.group_id = HARS.group_id 
                AND AR.replica_id = HARS.replica_id 
            ),
    Pri_CommitTime AS 
            (
            SELECT  replica_server_name
                    , DBName
                    , last_commit_time
            FROM    AG_Stats
            WHERE   role_desc = 'PRIMARY'
            ),
    Sec_CommitTime AS 
            (
            SELECT  replica_server_name
                    , DBName
                    , last_commit_time
            FROM    AG_Stats
            WHERE   role_desc = 'SECONDARY'
            )
SELECT p.replica_server_name [primary_replica]
    , p.[DBName] AS [DatabaseName]
    , s.replica_server_name [secondary_replica]
    , s.last_commit_time
    , p.last_commit_time
    , DATEDIFF(ms,s.last_commit_time,p.last_commit_time) AS [Sync_Lag_MS]
FROM Pri_CommitTime p
LEFT JOIN Sec_CommitTime s ON [s].[DBName] = [p].[DBName]


Result:

primary_replica DatabaseName secondary_replica last_commit_time        last_commit_time        Sync_Lag_MS
--------------- ------------ ----------------- ----------------------- ----------------------- -----------
XXXXX       NextGenAG    ANGSQLD01         2018-06-08 13:18:08.853 2018-06-08 13:18:08.853           0


However the last_commit_time columns are not updated in real-time. They are updated every 5-10 minutes, which is not the case based on the OLTP workload which performs write operations every few milliseconds.
You can technically do it using XEvents but you have to build your own tool to stream the data in near to realtime to your dashboard .. this involves data extraction and streaming.

Workflow will be as below :

Collect AG info:

 USE tempdb
                  IF OBJECT_ID('AGInfo') IS NOT NULL
                      BEGIN
                        DROP TABLE AGInfo
                   END 
                  IF OBJECT_ID('LatencyCollectionStatus') IS NOT NULL
                      BEGIN
                        DROP TABLE LatencyCollectionStatus
                      END
                   CREATE TABLE LatencyCollectionStatus(
                        [collection_status] [NVARCHAR](60)  NULL,
                        [start_timestamp] [DATETIMEOFFSET] NULL,
                        [startutc_timestamp] [DATETIMEOFFSET] NULL
                    )
                  INSERT INTO LatencyCollectionStatus(collection_status, start_timestamp, startutc_timestamp) values ('Started', GETDATE(), GETUTCDATE())
                  SELECT
                  AGC.name as agname
                  , RCS.replica_server_name as replica_name
                  , ARS.role_desc as agrole
                  INTO AGInfo
                  FROM
                      sys.availability_groups_cluster AS AGC
                      INNER JOIN sys.dm_hadr_availability_replica_cluster_states AS RCS
                      ON
                      RCS.group_id = AGC.group_id
                      INNER JOIN sys.dm_hadr_availability_replica_states AS ARS
                      ON
                      ARS.replica_id = RCS.replica_id
                      where AGC.name =  N'YOUR AG NAME '-- change here !!


Create XE Session as below :

IF EXISTS (select * from sys.server_event_sessions 
                WHERE name = N'AlwaysOn_Data_Movement_Tracing')
                    BEGIN
                    DROP EVENT SESSION [AlwaysOn_Data_Movement_Tracing] ON SERVER 
                    END
                CREATE EVENT SESSION [AlwaysOn_Data_Movement_Tracing] ON SERVER ADD EVENT sqlserver.hadr_apply_log_block, 
ADD EVENT sqlserver.hadr_capture_log_block, 
ADD EVENT sqlserver.hadr_database_flow_control_action, 
ADD EVENT sqlserver.hadr_db_commit_mgr_harden, 
ADD EVENT sqlserver.hadr_log_block_send_complete, 
ADD EVENT sqlserver.hadr_send_harden_lsn_message, 
ADD EVENT sqlserver.hadr_transport_flow_control_action, 
ADD EVENT sqlserver.log_flush_complete, 
ADD EVENT sqlserver.log_flush_start, 
ADD EVENT sqlserver.recovery_unit_harden_log_timestamps, 
ADD EVENT sqlserver.log_block_pushed_to_logpool, 
ADD EVENT sqlserver.hadr_transport_receive_log_block_message, 
ADD EVENT sqlserver.hadr_receive_harden_lsn_message, 
ADD EVENT sqlserver.hadr_log_block_group_commit, 
ADD EVENT sqlserver.hadr_log_block_compression, 
ADD EVENT sqlserver.hadr_log_block_decompression, 
ADD EVENT sqlserver.hadr_lsn_send_complete, 
ADD EVENT sqlserver.hadr_capture_filestream_wait, 
ADD EVENT sqlserver.hadr_capture_vlfheader ADD TARGET package0.event_file(SET filename=N'AlwaysOn_Data_Movement_Tracing.xel',max_file_size=(25),max_rollover_files=(4))
                WITH (MAX_MEMORY=4096 KB,EVENT_RETENTION_MODE=ALLOW_SINGLE_EVENT_LOSS,MAX_DISPATCH_LATENCY=30 SECONDS,MAX_EVENT_SIZE=0 KB,MEMORY_PARTITION_MODE=NONE,TRACK_CAUSALITY=OFF,STARTUP_STATE=ON)

                ALTER EVENT SESSION [AlwaysOn_Data_Movement_Tracing] ON SERVER STATE = START


Extract XEvent Info:

                BEGIN TRANSACTION
                USE tempdb
                IF OBJECT_ID('#EventXml') IS NOT NULL
                BEGIN
                    DROP TABLE #EventXml
                END 

                SELECT 
                    xe.event_name, 
                    CAST(xe.event_data AS XML) AS event_data
                INTO #EventXml
                FROM
                (
                SELECT
                        object_name AS event_name,
                        CAST(event_data AS XML) AS event_data
                    FROM sys.fn_xe_file_target_read_file(
                                'AlwaysOn_Data_Movement_Tracing*.xel', 
                                NULL, NULL, NULL)
                    WHERE object_name IN ('hadr_log_block_group_commit',
                                'log_block_pushed_to_logpool',
                                'log_flush_start',
                                'log_flush_complete',
                                'hadr_log_block_compression',
                                'hadr_capture_log_block',
                                'hadr_capture_filestream_wait',
                                'hadr_log_block_send_complete',
                                'hadr_receive_harden_lsn_message',
                                'hadr_db_commit_mgr_harden',
                                'recovery_unit_harden_log_timestamps',
                                'hadr_capture_vlfheader',
                                'hadr_log_block_decompression',
                                'hadr_apply_log_block',
                                'hadr_send_harden_lsn_message',
                                'hadr_log_block_decompression',
                                'hadr_lsn_send_complete',
                                'hadr_transport_receive_log_block_message')

                ) xe

                IF OBJECT_ID('DMReplicaEvents') IS NOT NULL
                BEGIN
                    DROP TABLE DMReplicaEvents
                END 

                SET ANSI_NULLS ON

                SET QUOTED_IDENTIFIER ON

                CREATE TABLE DMReplicaEvents(
                    [server_name] [NVARCHAR](128) NULL,
                    [event_name] [NVARCHAR](60) NOT NULL,
                    [log_block_id] [BIGINT] NULL,
                    [database_id] [INT] NULL,
                    [processing_time] [BIGINT] NULL,
                    [start_timestamp] [BIGINT] NULL,
                    [publish_timestamp] [DATETIMEOFFSET] NULL,
                    [log_block_size] [BIGINT] NULL,
                    [target_availability_replica_id] [UNIQUEIDENTIFIER] NULL,
                    [local_availability_replica_id] [UNIQUEIDENTIFIER] NULL,
                    [database_replica_id] [UNIQUEIDENTIFIER] NULL,
                    [mode] [BIGINT] NULL,
                    [availability_group_id] [UNIQUEIDENTIFIER] NULL,
                    [pending_writes]  [BIGINT] NULL
                )

                IF OBJECT_ID('LatencyResults') IS NOT NULL
                BEGIN
                    DROP TABLE LatencyResults
                END 
                CREATE TABLE LatencyResults(
                   [event_name] [NVARCHAR](60) NOT NULL,
                   [processing_time] [BIGINT] NULL,
                   [publish_timestamp] [DATETIMEOFFSET] NULL,
                   [server_commit_mode] [NVARCHAR](60) NULL
                )


                INSERT INTO DMReplicaEvents
                SELECT 
                    @@SERVERNAME AS server_name,
                    xe.event_name,
                    AoData.value('(data[@name="log_block_id"]/value)[1]', 'BIGINT') AS log_block_id,
                    NULL AS database_id,
                    AoData.value('(data[@name="total_processing_time"]/value)[1]', 'BIGINT') AS processing_time,
                    AoData.value('(data[@name="start_timestamp"]/value)[1]', 'BIGINT') AS start_timestamp,
                    CAST(SUBSTRING(CAST(xe.event_data AS NVARCHAR(MAX)), 75, 24) AS DATETIMEOFFSET) AS publish_timestamp,
                    AoData.value('(data[@name="log_block_size"]/value)[1]', 'BIGINT') AS log_block_size,
                    NULL AS target_availability_replica_id,
                    NULL AS local_availability_replica_id,
                    NULL AS database_replica_id,
                    NULL AS mode,
                    NULL AS availability_group_id,
                    NULL AS pending_writes
                FROM #EventXml AS xe
                CROSS APPLY xe.event_data.nodes('/event')  AS T(AoData)
                WHERE xe.event_name = 'hadr_log_block_send_complete'

                GO


                INSERT INTO DMReplicaEvents
                SELECT 
                    @@SERVERNAME AS server_name,
                    xe.event_name,
                    AoData.value('(data[@name="log_block_id"]/value)[1]', 'BIGINT') AS log_block_id,
                    AoData.value('(data[@name="database_id"]/value)[1]', 'INT') AS database_id,
                    AoData.value('(data[@name="duration"]/value)[1]', 'BIGINT') AS processing_time,
                    AoData.value('(data[@name="start_timestamp"]/value)[1]', 'BIGINT') AS start_timestamp,
                    CAST(SUBSTRING(CAST(xe.event_data AS NVARCHAR(MAX)), 65, 24) AS DATETIMEOFFSET) AS publish_timestamp,
                    NULL AS log_block_size,
                    NULL AS target_availability_replica_id,
                    NULL AS local_availability_replica_id,
                    NULL AS database_replica_id,
                    NULL AS mode,
                    NULL AS availability_group_id,
                    AoData.value('(data[@name="pending_writes"]/value)[1]','BIGINT') AS pending_writes
                FROM #EventXml AS xe
                CROSS APPLY xe.event_data.nodes('/event')  AS T(AoData)
                WHERE xe.event_name = 'log_flush_complete'

                GO

                INSERT INTO DMReplicaEvents
                SELECT 
                    @@SERVERNAME AS server_name,
                    xe.event_name,
                    NULL AS log_block_id,
                    AoData.value('(data[@name="database_id"]/value)[1]', 'BIGINT') AS database_id,
                    AoData.value('(data[@name="time_to_commit"]/value)[1]', 'BIGINT') AS processing_time,
                    NULL AS start_timestamp,
                    CAST(SUBSTRING(CAST(xe.event_data AS NVARCHAR(MAX)), 72, 24) AS DATETIMEOFFSET) AS publish_timestamp,
                    NULL AS log_block_size,
                    AoData.value('(data[@name="replica_id"]/value)[1]', 'UNIQUEIDENTIFIER') AS target_availability_replica_id,
                    NULL AS local_availability_replica_id,
                    AoData.value('(data[@name="ag_database_id"]/value)[1]', 'UNIQUEIDENTIFIER') AS database_replica_id,
                    NULL AS mode,
                    AoData.value('(data[@name="group_id"]/value)[1]','UNIQUEIDENTIFIER') AS availability_group_id,
                    NULL AS pending_writes
                FROM #EventXml AS xe
                CROSS APPLY xe.event_data.nodes('/event')  AS T(AoData)
                WHERE xe.event_name = 'hadr_db_commit_mgr_harden'

                GO


                INSERT INTO DMReplicaEvents
                SELECT 
                    @@SERVERNAME AS server_name,
                    xe.event_name,
                    AoData.value('(data[@name="log_block_id"]/value)[1]', 'BIGINT') AS log_block_id,
                    AoData.value('(data[@name="database_id"]/value)[1]', 'BIGINT') AS database_id,
                    AoData.value('(data[@name="processing_time"]/value)[1]', 'BIGINT') AS processing_time,
                    AoData.value('(data[@name="start_timestamp"]/value)[1]', 'BIGINT') AS start_timestamp,
                    CAST(SUBSTRING(CAST(xe.event_data AS NVARCHAR(MAX)), 82, 24) AS DATETIMEOFFSET) AS publish_timestamp,
                    NULL AS log_block_size,
                    NULL AS target_availability_replica_id,
                    NULL AS local_availability_replica_id,
                    NULL AS database_replica_id,
                    NULL AS mode,
                    NULL AS availability_group_id,
                    NULL AS pending_writes
                FROM #EventXml AS xe
                CROSS APPLY xe.event_data.nodes('/event')  AS T(AoData)
                WHERE xe.event_name = 'recovery_unit_harden_log_timestamps'

                GO

                INSERT INTO DMReplicaEvents
                SELECT 
                    @@SERVERNAME AS server_name,
                    xe.event_name,
                    AoData.value('(data[@name="log_block_id"]/value)[1]', 'BIGINT') AS log_block_id,
                    AoData.value('(data[@name="database_id"]/value)[1]', 'BIGINT') AS database_id,
                    AoData.value('(data[@name="processing_time"]/value)[1]', 'BIGINT') AS processing_time,
                    AoData.value('(data[@name="start_timestamp"]/value)[1]', 'BIGINT') AS start_timestamp,
                    CAST(SUBSTRING(CAST(xe.event_data AS NVARCHAR(MAX)), 73, 24) AS DATETIMEOFFSET) AS publish_timestamp,
                    AoData.value('(data[@name="uncompressed_size"]/value)[1]', 'INT') AS log_block_size,
                    AoData.value('(data[@name="availability_replica_id"]/value)[1]', 'UNIQUEIDENTIFIER') AS target_availability_replica_id,
                    NULL AS local_availability_replica_id,
                    NULL AS database_replica_id,
                    NULL AS mode,
                    NULL AS availability_group_id,
                    NULL AS pending_writes
                FROM #EventXml AS xe
                CROSS APPLY xe.event_data.nodes('/event')  AS T(AoData)
                WHERE xe.event_name = 'hadr_log_block_compression'

                GO


                INSERT INTO DMReplicaEvents
                SELECT 
                    @@SERVERNAME AS server_name,
                    xe.event_name,
                    AoData.value('(data[@name="log_block_id"]/value)[1]', 'BIGINT') AS log_block_id,
                    AoData.value('(data[@name="database_id"]/value)[1]', 'BIGINT') AS database_id,
                    AoData.value('(data[@name="processing_time"]/value)[1]', 'BIGINT') AS processing_time,
                    AoData.value('(data[@name="start_timestamp"]/value)[1]', 'BIGINT') AS start_timestamp,
                    CAST(SUBSTRING(CAST(xe.event_data AS NVARCHAR(MAX)), 75, 24) AS DATETIMEOFFSET) AS publish_timestamp,
                    AoData.value('(data[@name="uncompressed_size"]/value)[1]', 'BIGINT') AS log_block_size,
                    AoData.value('(data[@name="availability_replica_id"]/value)[1]', 'UNIQUEIDENTIFIER') AS target_availability_replica_id,
                    NULL AS local_availability_replica_id,
                    NULL AS database_replica_id,
                    NULL AS mode,
                    NULL AS availability_group_id,
                    NULL AS pending_writes
                FROM #EventXml AS xe
                CROSS APPLY xe.event_data.nodes('/event')  AS T(AoData)
                WHERE xe.event_name = 'hadr_log_block_decompression'

                INSERT INTO DMReplicaEvents
                SELECT 
                    @@SERVERNAME AS server_name,
                    xe.event_name,
                    AoData.value('(data[@name="log_block_id"]/value)[1]', 'BIGINT') AS log_block_id,
                    NULL AS database_id,
                    AoData.value('(data[@name="total_sending_time"]/value)[1]', 'BIGINT') AS processing_time,
                    AoData.value('(data[@name="start_timestamp"]/value)[1]', 'BIGINT') AS start_timestamp,
                    CAST(SUBSTRING(CAST(xe.event_data AS NVARCHAR(MAX)), 69, 24) AS DATETIMEOFFSET) AS publish_timestamp,
                    NULL AS log_block_size,
                    NULL AS target_availability_replica_id,
                    NULL AS local_availability_replica_id,
                    NULL AS database_replica_id,
                    NULL AS mode,
                    NULL AS availability_group_id,
                    NULL AS pending_writes
                FROM #EventXml AS xe
                CROSS APPLY xe.event_data.nodes('/event')  AS T(AoData)
                WHERE xe.event_name = 'hadr_lsn_send_complete'

                INSERT INTO DMReplicaEvents
                SELECT 
                    @@SERVERNAME AS server_name,
                    xe.event_name,
                    AoData.value('(data[@name="log_block_id"]/value)[1]', 'BIGINT') AS log_block_id,
                    NULL AS database_id,
                    AoData.value('(data[@name="processing_time"]/value)[1]', 'BIGINT') AS processing_time,
                    AoData.value('(data[@name="start_timestamp"]/value)[1]', 'BIGINT') AS start_timestamp,
                    CAST(SUBSTRING(CAST(xe.event_data AS NVARCHAR(MAX)), 87, 24) AS DATETIMEOFFSET) AS publish_timestamp,
                    NULL AS log_block_size,
                    AoData.value('(data[@name="target_availability_replica_id"]/value)[1]', 'UNIQUEIDENTIFIER') AS target_availability_replica_id,
                    AoData.value('(data[@name="local_availability_replica_id"]/value)[1]', 'UNIQUEIDENTIFIER') AS local_availability_replica_id,
                    AoData.value('(data[@name="target_availability_replica_id"]/value)[1]', 'UNIQUEIDENTIFIER') AS database_replica_id,
                    AoData.value('(data[@name="mode"]/value)[1]', 'BIGINT') AS mode,
                    AoData.value('(data[@name="availability_group_id"]/value)[1]','UNIQUEIDENTIFIER') AS availability_group_id,
                    NULL AS pending_writes
                FROM #EventXml AS xe
                CROSS APPLY xe.event_data.nodes('/event')  AS T(AoData)
                WHERE xe.event_name = 'hadr_transport_receive_log_block_message'


                DELETE
                FROM DMReplicaEvents
                WHERE CAST(publish_timestamp AS DATETIME) &lt; DATEADD(minute, -2, CAST((SELECT MAX(publish_timestamp) from DMReplicaEvents) as DATETIME))
                COMMIT
                GO