Monitoring Streams through SQL*Plus

As with everything Oracle these days you can monitor/administer Streams via SQL scripts or via Enterprise Manager Grid Control. This brief post will show some simple queries and a further post will show some screenshots of some of the Grid Control pages related to Streams.

Usual disclaimer - scripts/queries used at your own risk with no liability accepted.

Monitoring the capture process

-- General information about each capture process:

COLUMN capture_name HEADING 'Capture|Process|Name' FORMAT A15
COLUMN queue_name HEADING 'Capture|Process|Queue' FORMAT A15
COLUMN rule_set_name HEADING 'Positive|Rule Set' FORMAT A15
COLUMN negative_rule_set_name HEADING 'Negative| Rule Set' FORMAT A15
COLUMN status HEADING 'Capture|Process|Status' FORMAT A15
SELECT capture_name, queue_name, rule_set_name, negative_rule_set_name, status
FROM dba_capture;
Capture         Capture                                         Capture
Process         Process         Positive        Negative        Process
Name            Queue           Rule Set         Rule Set       Status
--------------- --------------- --------------- --------------- --------
STREAMS_CAPTURE STREAMS_CAPTURE RULESET$_78                     ENABLED
                _QUEUE

-- Change capture information for each process:

COLUMN CAPTURE_NAME HEADING 'Capture|Name' FORMAT A7
COLUMN PROCESS_NAME HEADING 'Capture|Process|Number' FORMAT A7
COLUMN SID HEADING 'Session|ID' FORMAT 9999
COLUMN SERIAL# HEADING 'Session|Serial|Number' FORMAT 9999
COLUMN STATE HEADING 'State' FORMAT A27
COLUMN TOTAL_MESSAGES_CAPTURED HEADING 'Redo|Entries|Evaluated|In Detail' FORMAT 9999999
COLUMN TOTAL_MESSAGES_ENQUEUED HEADING 'Total|LCRs|Enqueued' FORMAT 999999
SELECT c.capture_name,
SUBSTR(s.program,INSTR(s.program,'(')+1,4) PROCESS_NAME,
c.sid,
c.serial#,
c.state,
c.total_messages_captured,
c.total_messages_enqueued
FROM v$streams_capture c, v$session s
WHERE c.sid = s.sid
AND c.serial# = s.serial#;
                                                                 Redo
        Capture         Session                               Entries    Total
Capture Process Session  Serial                             Evaluated     LCRs
Name    Number       ID  Number State                       In Detail Enqueued
------- ------- ------- ------- --------------------------- --------- --------
STREAMS C001        131      49 CAPTURING CHANGES              933522   343060
_CAPTUR
E

-- Change capture latency:

COLUMN CAPTURE_NAME HEADING 'Capture|Process|Name' FORMAT A12
COLUMN LATENCY_SECONDS HEADING 'Latency|in|Seconds' FORMAT 999999
COLUMN CREATE_TIME HEADING 'Event Creation|Time' FORMAT A20
COLUMN ENQUEUE_TIME HEADING 'Enqueue Time' FORMAT A20
COLUMN ENQUEUE_MESSAGE_NUMBER HEAD 'Message|Number' FORMAT 999999999
SELECT CAPTURE_NAME,
TO_CHAR((ENQUEUE_TIME-ENQUEUE_MESSAGE_CREATE_TIME)*86400) LATENCY_SECONDS,
TO_CHAR(ENQUEUE_MESSAGE_CREATE_TIME, 'HH24:MI:SS MM/DD/YY')
CREATE_TIME,
TO_CHAR(ENQUEUE_TIME, 'HH24:MI:SS MM/DD/YY') ENQUEUE_TIME,
TO_CHAR(ENQUEUE_MESSAGE_NUMBER) ENQUEUE_MESSAGE_NUMBER
FROM V$STREAMS_CAPTURE;
Capture      Latency
Process      in                                       Event Creation
Name         Seconds                                  Time
------------ ---------------------------------------- ------------------
                     Message
Enqueue Time         Number
-------------------- ----------------------------------------
STREAMS_CAPT 0                                        08:55:25 08/17/09
URE
08:55:25 08/17/09    7874143550786

-- When the last message was captured from the redo stream:

COLUMN CAPTURE_NAME HEADING 'Capture|Process|Name' FORMAT A12
COLUMN AVAILABLE_MESSAGE_NUMBER FORMAT A25
SELECT capture_name, logminer_id,
TO_CHAR(available_message_number) AVAILABLE_MESSAGE_NUMBER,
TO_CHAR(available_message_create_time, 'HH24:MI:SS MM/DD/YY')
  AVAILABLE_MESSAGE_CREATE_TIME
FROM v$streams_capture;
Capture
Process
Name         LOGMINER_ID AVAILABLE_MESSAGE_NUMBER  AVAILABLE_MESSAGE
------------ ----------- ------------------------- -----------------
STREAMS_CAPT          65 7874143550897             08:56:26 08/17/09
URE

-- When changes are entered into the capture queue:

COLUMN CAPTURE_NAME HEADING 'Capture|Process|Name' FORMAT A12
COLUMN LATENCY_SECONDS HEADING 'Latency|in|Seconds' FORMAT 999999
COLUMN ENQUEUE_TIME HEADING 'Enqueue Time' FORMAT A20
COLUMN CREATE_TIME HEADING 'Event Creation|Time' FORMAT A20
SELECT capture_name,
(enqueue_time-enqueue_message_create_time)*86400 LATENCY_SECONDS,
TO_CHAR(enqueue_message_create_time, 'HH24:MI:SS MM/DD/YY') CREATE_TIME,
TO_CHAR(enqueue_time, 'HH24:MI:SS MM/DD/YY') ENQUEUE_TIME,
TO_CHAR(enqueue_message_number) MESSAGE_NUMBER
FROM v$streams_capture;
Capture      Latency
Process           in Event Creation
Name         Seconds Time                 Enqueue Time
------------ ------- -------------------- -----------------
MESSAGE_NUMBER
----------------------------------------
STREAMS_CAPT       0 08:57:37 08/17/09    08:57:37 08/17/09
URE
7874143551014

Monitoring the Apply Process

-- Apply latency:

column latency heading "Latency|in|Seconds" format 9999
column creation heading "Event Creation" format a17
column last_dequeue heading "Last Dequeue Time" format a20
column dequeued_message_number heading "Dequeued|Message Number"
SELECT (dequeue_time-dequeued_message_create_time)*8600 LATENCY,
TO_CHAR(dequeued_message_create_time,'hh24:mi:ss mm/dd/yy') CREATION,
TO_CHAR(dequeue_time,'hh24:mi:ss mm/dd/yy') LAST_DEQUEUE,
TO_CHAR(dequeued_message_number) DEQUEUED_MESSAGE_NUMBER
FROM v$streams_apply_reader;
Latency
     in
Seconds Event Creation    Last Dequeue Time
------- ----------------- ------------------
Dequeued
Message Number
----------------------------------------
   1792 09:06:00 08/17/09 14:06:00 08/17/09
7874143552338

-- When the Apply Process starts working on captured changes:

column latency heading "Latency|in|Seconds" format 9999
column creation heading "Event Creation" format a17
column last_dequeue heading "Last Dequeue Time" format a20
column dequeued_message_number heading "Dequeued|Message Number"
SELECT apply_name,
TO_CHAR((dequeue_time-dequeued_message_create_time)*86400) LATENCY,
TO_CHAR(dequeued_message_create_time,'HH24:MI:SS MM/DD/YY') CREATION,
TO_CHAR(dequeue_time,'HH24:MI:SS MM/DD/YY') LAST_DEQUEUE,
TO_CHAR(dequeued_message_number) DEQUEUED_MESSAGE_NUMBER,
state
FROM v$streams_apply_reader;
                               Latency
                               in
APPLY_NAME                     Seconds
------------------------------ ---------------------------
                                       Dequeued
Event Creation    Last Dequeue Time    Message Number
----------------- -------------------- -------------------
STATE
-----------------
STREAMS_APPLY                  18002
09:07:52 08/17/09 14:07:54 08/17/09    7874143552552
DEQUEUE MESSAGES

-- Streams Apply errors:

SELECT apply_name, source_database, local_transaction_id, error_number,
       error_message, message_count
FROM dba_apply_error;

Happily there are currently no errors in my Streams Apply process so I can't show you an example :-) The "LOCAL_TRANSACTION_ID" is important here - you can use this to retry or delete transactions that have errored (see the following code scripts).

-- Retrying a specific apply error transaction:
After you correct the conditions that caused an apply error, you can retry the transaction by running the EXECUTE_ERROR procedure in the DBMS_APPLY_ADM package without specifying a user procedure. In this case,  the transaction is executed without any custom processing.

BEGIN
DBMS_APPLY_ADM.EXECUTE_ERROR(
local_transaction_id => 'local transaction ID from DBA_APPLY_ERROR',
execute_as_user => FALSE,
user_procedure => NULL);
END;
/

-- Retrying all error transactions for an apply process:
After you correct the conditions that caused all of the apply errors for an apply process, you can retry all of the error transactions by running the EXECUTE_ALL_ERRORS procedure in the DBMS_APPLY_ADM package. For example, to retry all of the error transactions for an apply process named streams_apply, you can run the following procedure:

BEGIN
DBMS_APPLY_ADM.EXECUTE_ALL_ERRORS(
apply_name => 'streams_apply',
execute_as_user => FALSE);
END;
/

-- Deleting a specific apply error transaction:
If an error transaction should not be applied, then you can delete the transaction from the error queue using the DELETE_ERROR procedure in the DBMS_APPLY_ADM package. For example, to delete a transaction with the transaction identifier 5.4.312, run the following procedure:

EXEC DBMS_APPLY_ADM.DELETE_ERROR(local_transaction_id => '5.4.312');

-- Deleting all error transactions for an apply process:
If none of the error transactions should be applied, then you can delete all of the error transactions by running the DELETE_ALL_ERRORS procedure in the DBMS_APPLY_ADM package. For example, to delete all of the error transactions for an apply process named streams_apply, you can run the following procedure:

EXEC DBMS_APPLY_ADM.DELETE_ALL_ERRORS(apply_name => 'streams_apply');

Removing a Streams configuration
To remove the Streams configuration from your database execute the following:

EXEC DBMS_STREAMS_ADM.REMOVE_STREAMS_CONFIGURATION();

Cheers,

Ian