⚠️ Error Handling & Monitoring Best Practices¶

CREATE TABLE dbo.pipeline_errors (
    error_id BIGINT IDENTITY(1,1) PRIMARY KEY,
    error_timestamp DATETIME2 DEFAULT GETUTCDATE(),
    pipeline_name NVARCHAR(256),
    pipeline_run_id NVARCHAR(256),
    activity_name NVARCHAR(256),
    activity_type NVARCHAR(128),
    error_code NVARCHAR(128),
    error_message NVARCHAR(MAX),
    error_classification NVARCHAR(64),  -- TRANSIENT, PERMANENT, DATA_QUALITY, PERMISSION, TIMEOUT, RESOURCE
    severity NVARCHAR(16),              -- CRITICAL, HIGH, MEDIUM, LOW
    source_system NVARCHAR(256),
    affected_table NVARCHAR(256),
    affected_record_count INT,
    retry_count INT DEFAULT 0,
    max_retries INT DEFAULT 3,
    is_resolved BIT DEFAULT 0,
    resolution_notes NVARCHAR(MAX),
    resolved_by NVARCHAR(256),
    resolved_timestamp DATETIME2,
    alert_sent BIT DEFAULT 0,
    alert_channel NVARCHAR(128),
    stack_trace NVARCHAR(MAX),
    input_parameters NVARCHAR(MAX),
    environment NVARCHAR(64),           -- DEV, STAGING, PROD
    correlation_id NVARCHAR(256),
    parent_pipeline_run_id NVARCHAR(256)
);

-- Fast lookup by pipeline and time
CREATE NONCLUSTERED INDEX IX_pipeline_errors_timestamp
    ON dbo.pipeline_errors (error_timestamp DESC)
    INCLUDE (pipeline_name, severity, is_resolved);

-- Unresolved errors dashboard
CREATE NONCLUSTERED INDEX IX_pipeline_errors_unresolved
    ON dbo.pipeline_errors (is_resolved, severity)
    WHERE is_resolved = 0;

-- Correlation tracking across nested pipelines
CREATE NONCLUSTERED INDEX IX_pipeline_errors_correlation
    ON dbo.pipeline_errors (correlation_id)
    INCLUDE (pipeline_name, activity_name, error_classification);

-- Error classification analysis
CREATE NONCLUSTERED INDEX IX_pipeline_errors_classification
    ON dbo.pipeline_errors (error_classification, error_timestamp DESC);

CREATE VIEW dbo.vw_error_summary AS
SELECT
    CAST(error_timestamp AS DATE) AS error_date,
    pipeline_name,
    error_classification,
    severity,
    COUNT(*) AS error_count,
    SUM(CASE WHEN is_resolved = 1 THEN 1 ELSE 0 END) AS resolved_count,
    SUM(CASE WHEN is_resolved = 0 THEN 1 ELSE 0 END) AS unresolved_count,
    AVG(DATEDIFF(MINUTE, error_timestamp, ISNULL(resolved_timestamp, GETUTCDATE()))) AS avg_resolution_minutes,
    MAX(error_timestamp) AS last_occurrence
FROM dbo.pipeline_errors
GROUP BY
    CAST(error_timestamp AS DATE),
    pipeline_name,
    error_classification,
    severity;

CREATE PROCEDURE dbo.usp_log_pipeline_error
    @pipeline_name NVARCHAR(256),
    @pipeline_run_id NVARCHAR(256),
    @activity_name NVARCHAR(256),
    @activity_type NVARCHAR(128),
    @error_code NVARCHAR(128),
    @error_message NVARCHAR(MAX),
    @source_system NVARCHAR(256) = NULL,
    @affected_table NVARCHAR(256) = NULL,
    @affected_record_count INT = NULL,
    @retry_count INT = 0,
    @stack_trace NVARCHAR(MAX) = NULL,
    @input_parameters NVARCHAR(MAX) = NULL,
    @environment NVARCHAR(64) = 'PROD',
    @correlation_id NVARCHAR(256) = NULL,
    @parent_pipeline_run_id NVARCHAR(256) = NULL
AS
BEGIN
    SET NOCOUNT ON;

    DECLARE @error_classification NVARCHAR(64);
    DECLARE @severity NVARCHAR(16);
    DECLARE @max_retries INT = 3;

    -- Classify the error based on error code and message patterns
    SET @error_classification = CASE
        -- Transient errors (retriable)
        WHEN @error_code IN ('ConnectionTimeout', 'ServiceUnavailable', '429', '503', '504')
            THEN 'TRANSIENT'
        WHEN @error_message LIKE '%timeout%' OR @error_message LIKE '%throttl%'
            THEN 'TRANSIENT'
        WHEN @error_message LIKE '%deadlock%' OR @error_message LIKE '%lock timeout%'
            THEN 'TRANSIENT'

        -- Permission errors
        WHEN @error_code IN ('403', 'Forbidden', 'Unauthorized', '401')
            THEN 'PERMISSION'
        WHEN @error_message LIKE '%access denied%' OR @error_message LIKE '%not authorized%'
            THEN 'PERMISSION'

        -- Data quality errors
        WHEN @error_message LIKE '%schema mismatch%' OR @error_message LIKE '%type mismatch%'
            THEN 'DATA_QUALITY'
        WHEN @error_message LIKE '%null value%' OR @error_message LIKE '%constraint violation%'
            THEN 'DATA_QUALITY'
        WHEN @error_message LIKE '%duplicate%' OR @error_message LIKE '%unique%'
            THEN 'DATA_QUALITY'

        -- Timeout errors
        WHEN @error_code LIKE '%Timeout%' OR @error_message LIKE '%exceeded the timeout%'
            THEN 'TIMEOUT'

        -- Resource errors
        WHEN @error_message LIKE '%out of memory%' OR @error_message LIKE '%disk space%'
            THEN 'RESOURCE'
        WHEN @error_message LIKE '%capacity%' OR @error_message LIKE '%throttl%'
            THEN 'RESOURCE'

        -- Default: permanent
        ELSE 'PERMANENT'
    END;

    -- Determine severity based on classification and context
    SET @severity = CASE
        -- Critical: compliance pipelines or permanent failures in production
        WHEN @environment = 'PROD' AND @error_classification = 'PERMANENT'
            AND (@pipeline_name LIKE '%compliance%' OR @pipeline_name LIKE '%ctr%'
                 OR @pipeline_name LIKE '%sar%' OR @pipeline_name LIKE '%hipaa%')
            THEN 'CRITICAL'

        -- Critical: resource exhaustion in production
        WHEN @environment = 'PROD' AND @error_classification = 'RESOURCE'
            THEN 'CRITICAL'

        -- High: any permanent error in production
        WHEN @environment = 'PROD' AND @error_classification IN ('PERMANENT', 'PERMISSION')
            THEN 'HIGH'

        -- High: data quality errors affecting gold layer
        WHEN @error_classification = 'DATA_QUALITY' AND @affected_table LIKE 'gold_%'
            THEN 'HIGH'

        -- Medium: transient errors that have exhausted retries
        WHEN @error_classification = 'TRANSIENT' AND @retry_count >= @max_retries
            THEN 'MEDIUM'

        -- Medium: data quality errors in silver layer
        WHEN @error_classification = 'DATA_QUALITY' AND @affected_table LIKE 'silver_%'
            THEN 'MEDIUM'

        -- Low: transient errors still retrying, dev/staging errors
        WHEN @environment IN ('DEV', 'STAGING') THEN 'LOW'
        WHEN @error_classification = 'TRANSIENT' AND @retry_count < @max_retries THEN 'LOW'

        ELSE 'MEDIUM'
    END;

    -- Determine if alerts should be sent
    DECLARE @alert_sent BIT = 0;
    DECLARE @alert_channel NVARCHAR(128) = NULL;

    IF @severity IN ('CRITICAL', 'HIGH') AND @environment = 'PROD'
    BEGIN
        SET @alert_sent = 1;
        SET @alert_channel = CASE
            WHEN @severity = 'CRITICAL' THEN 'Teams+Email+PagerDuty'
            WHEN @severity = 'HIGH' THEN 'Teams+Email'
            ELSE 'Teams'
        END;
    END;

    -- Insert the error record
    INSERT INTO dbo.pipeline_errors (
        pipeline_name, pipeline_run_id, activity_name, activity_type,
        error_code, error_message, error_classification, severity,
        source_system, affected_table, affected_record_count,
        retry_count, max_retries, stack_trace, input_parameters,
        environment, correlation_id, parent_pipeline_run_id,
        alert_sent, alert_channel
    )
    VALUES (
        @pipeline_name, @pipeline_run_id, @activity_name, @activity_type,
        @error_code, @error_message, @error_classification, @severity,
        @source_system, @affected_table, @affected_record_count,
        @retry_count, @max_retries, @stack_trace, @input_parameters,
        @environment, @correlation_id, @parent_pipeline_run_id,
        @alert_sent, @alert_channel
    );

    -- Return the error_id and classification for downstream use
    SELECT
        SCOPE_IDENTITY() AS error_id,
        @error_classification AS error_classification,
        @severity AS severity,
        @alert_sent AS alert_sent;
END;

CREATE PROCEDURE dbo.usp_get_error_summary
    @start_date DATETIME2 = NULL,
    @end_date DATETIME2 = NULL,
    @environment NVARCHAR(64) = NULL,
    @severity_filter NVARCHAR(16) = NULL,
    @unresolved_only BIT = 0
AS
BEGIN
    SET NOCOUNT ON;

    -- Default to last 7 days if no dates provided
    IF @start_date IS NULL SET @start_date = DATEADD(DAY, -7, GETUTCDATE());
    IF @end_date IS NULL SET @end_date = GETUTCDATE();

    -- Summary by pipeline and classification
    SELECT
        pipeline_name,
        error_classification,
        severity,
        COUNT(*) AS total_errors,
        SUM(CASE WHEN is_resolved = 1 THEN 1 ELSE 0 END) AS resolved,
        SUM(CASE WHEN is_resolved = 0 THEN 1 ELSE 0 END) AS unresolved,
        MIN(error_timestamp) AS first_occurrence,
        MAX(error_timestamp) AS last_occurrence,
        AVG(retry_count) AS avg_retries,
        AVG(CASE WHEN is_resolved = 1
            THEN DATEDIFF(MINUTE, error_timestamp, resolved_timestamp)
            ELSE NULL END) AS avg_resolution_minutes
    FROM dbo.pipeline_errors
    WHERE error_timestamp BETWEEN @start_date AND @end_date
        AND (@environment IS NULL OR environment = @environment)
        AND (@severity_filter IS NULL OR severity = @severity_filter)
        AND (@unresolved_only = 0 OR is_resolved = 0)
    GROUP BY pipeline_name, error_classification, severity
    ORDER BY
        CASE severity
            WHEN 'CRITICAL' THEN 1
            WHEN 'HIGH' THEN 2
            WHEN 'MEDIUM' THEN 3
            WHEN 'LOW' THEN 4
        END,
        COUNT(*) DESC;

    -- Hourly error trend
    SELECT
        DATEADD(HOUR, DATEDIFF(HOUR, 0, error_timestamp), 0) AS error_hour,
        error_classification,
        COUNT(*) AS error_count
    FROM dbo.pipeline_errors
    WHERE error_timestamp BETWEEN @start_date AND @end_date
        AND (@environment IS NULL OR environment = @environment)
    GROUP BY
        DATEADD(HOUR, DATEDIFF(HOUR, 0, error_timestamp), 0),
        error_classification
    ORDER BY error_hour DESC;

    -- Top recurring errors
    SELECT TOP 10
        error_code,
        LEFT(error_message, 200) AS error_summary,
        pipeline_name,
        COUNT(*) AS occurrence_count,
        MAX(error_timestamp) AS last_seen
    FROM dbo.pipeline_errors
    WHERE error_timestamp BETWEEN @start_date AND @end_date
        AND (@environment IS NULL OR environment = @environment)
    GROUP BY error_code, LEFT(error_message, 200), pipeline_name
    ORDER BY COUNT(*) DESC;
END;

CREATE PROCEDURE dbo.usp_resolve_error
    @error_id BIGINT,
    @resolved_by NVARCHAR(256),
    @resolution_notes NVARCHAR(MAX)
AS
BEGIN
    SET NOCOUNT ON;

    UPDATE dbo.pipeline_errors
    SET
        is_resolved = 1,
        resolved_by = @resolved_by,
        resolved_timestamp = GETUTCDATE(),
        resolution_notes = @resolution_notes
    WHERE error_id = @error_id;

    SELECT @@ROWCOUNT AS rows_updated;
END;

CREATE PROCEDURE dbo.usp_bulk_resolve_transient
    @pipeline_run_id NVARCHAR(256)
AS
BEGIN
    SET NOCOUNT ON;

    -- If the pipeline run ultimately succeeded, resolve all transient errors from that run
    UPDATE dbo.pipeline_errors
    SET
        is_resolved = 1,
        resolved_by = 'SYSTEM_AUTO_RESOLVE',
        resolved_timestamp = GETUTCDATE(),
        resolution_notes = 'Auto-resolved: pipeline run completed successfully after retry'
    WHERE pipeline_run_id = @pipeline_run_id
        AND error_classification = 'TRANSIENT'
        AND is_resolved = 0;

    SELECT @@ROWCOUNT AS auto_resolved_count;
END;