anomaly-detection

name	anomaly-detection
description	Rule-based anomaly detection for production systems with configurable thresholds, cooldown periods to prevent alert storms, and error pattern tracking for repeated failures.
license	MIT
compatibility	TypeScript/JavaScript, Python
metadata	{"category":"observability","time":"5h","source":"drift-masterguide"}

Anomaly Detection

Rule-based anomaly detection with cooldowns and error pattern tracking.

When to Use This Skill

• Detecting slow job degradation before failures
• Tracking error rate creep over time
• Identifying repeated error patterns
• Preventing alert fatigue with cooldowns

Core Concepts

Production systems fail in subtle ways - jobs getting slower, error rates creeping up, same errors repeating. The solution:

1. Configurable rules with severity levels
2. Cooldown periods to prevent alert storms
3. Error pattern tracking for repeated failures
4. Violation decay to reward recovery

Implementation

TypeScript

enum AnomalyType {
  SLOW_JOB = 'slow_job',
  HIGH_FAILURE_RATE = 'high_failure_rate',
  WORKER_UNHEALTHY = 'worker_unhealthy',
  QUEUE_BACKLOG = 'queue_backlog',
  TIMEOUT_SPIKE = 'timeout_spike',
  REPEATED_ERROR = 'repeated_error',
  MEMORY_SPIKE = 'memory_spike',
  CPU_SPIKE = 'cpu_spike',
}

enum AnomalySeverity {
  CRITICAL = 'critical',
  HIGH = 'high',
  MEDIUM = 'medium',
  LOW = 'low',
}

interface AnomalyAlert {
  id: string;
  anomalyType: AnomalyType;
  severity: AnomalySeverity;
  workerName: string;
  jobId?: string;
  message: string;
  details: Record<string, unknown>;
  detectedAt: Date;
  resolvedAt?: Date;
  ?: ;
}

  {
  : ;
  : ;
  : ;
  : ;
  : ;
  : ;
  : ;
  : ;
  : ;
  : ;
  : ;
}

  {
  : ;
  : ;
  : ;
  :  ;
  : ;
  : ;
}

 : [] = [
  {
    : .,
    : .,
    : ,
    :  ctx. > ctx. * ,
    : ,
    : ,
  },
  {
    : .,
    : .,
    : ,
    :  ctx. > ,
    : ,
    : ,
  },
  {
    : .,
    : .,
    : ,
    :  ctx. === ,
    : ,
    : ,
  },
  {
    : .,
    : .,
    : ,
    :  ctx. > ,
    : ,
    : ,
  },
  {
    : .,
    : .,
    : ,
    :  ctx. > ,
    : ,
    : ,
  },
  {
    : .,
    : .,
    : ,
    :  ctx. > ,
    : ,
    : ,
  },
];

  {
   alerts =  <, >();
   cooldowns =  <, >();
   errorCounts =  <, <, >>();
   timeoutCounts =  <, >();
   alertIdCounter = ;

  (
    : ,
    : {
      : ;
      : ;
      : ;
      : ;
      : ;
      : ;
      : ;
      : ;
    }
  ): [] {
     : [] = [];
    
     failureRate = health. > 
      ? (health. / health.) * 
      : ;

     :  = {
      workerName,
      : health.,
      failureRate,
      : health.,
      : health.,
      : health.,
      : ..(workerName) || ,
      : ,
      : ,
      : health.,
      : health.,
    };

     ( rule  ) {
       (.(workerName, rule.)) ;

       (rule.(ctx)) {
         alert = .(workerName, rule, ctx);
        detected.(alert);
        .(workerName, rule., rule.);
      }
    }

     detected;
  }

  (
    : ,
    : ,
    : ,
    : ,
    : ,
    ?: 
  ): [] {
     : [] = [];

     (!success && error) {
      .(workerName, error);
    }

    
     (durationMs > expectedDurationMs * ) {
       (!.(workerName, .)) {
         rule = [];
         alert = .(workerName, rule, {
          durationMs,
          expectedDurationMs,
        }  );
        alert. = jobId;
        detected.(alert);
        .(workerName, ., );
      }
    }

    
     (error) {
       errorCounts = ..(workerName);
       count = errorCounts?.(error.(, )) || ;
      
       (count >  && !.(workerName, .)) {
         rule = .( r. === .)!;
         alert = .(workerName, rule, {
          : error.(, ),
          : count,
        }  );
        detected.(alert);
        .(workerName, ., );
      }
    }

     detected;
  }

  (: , : ):  {
     alert = ..(alertId);
     (!alert || alert.)  ;

    alert. =  ();
    alert. = resolution;
     ;
  }

  (): [] {
     .(..())
      .( !a.)
      .( {
         order = { : , : , : , :  };
         order[a.] - order[b.];
      });
  }

   (: , : ):  {
     (!..(workerName)) {
      ..(workerName,  ());
    }
     counts = ..(workerName)!;
     key = error.(, );
    counts.(key, (counts.(key) || ) + );
  }

   (: , : , : <>):  {
     id = ;
    
     message = rule.;
     ( [key, value]  .(ctx)) {
      message = message.(, (value));
    }

     :  = {
      id,
      : rule.,
      : rule.,
      workerName,
      message,
      : ctx  <, >,
      :  (),
    };

    ..(id, alert);
     alert;
  }

   (: , : ):  {
     key = ;
     cooldownEnd = ..(key);
     cooldownEnd !==  && cooldownEnd >  ();
  }

   (: , : , : ):  {
     key = ;
    ..(key,  (.() + seconds * ));
  }
}


 :  |  = ;

  ():  {
   (!detector) {
    detector =  ();
  }
   detector;
}

Python

from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Dict, List, Optional, Callable
from enum import Enum


class AnomalyType(str, Enum):
    SLOW_JOB = "slow_job"
    HIGH_FAILURE_RATE = "high_failure_rate"
    WORKER_UNHEALTHY = "worker_unhealthy"
    QUEUE_BACKLOG = "queue_backlog"
    TIMEOUT_SPIKE = "timeout_spike"
    REPEATED_ERROR = "repeated_error"
    MEMORY_SPIKE = "memory_spike"


class AnomalySeverity(str, Enum):
    CRITICAL = "critical"
    HIGH = "high"
    MEDIUM = "medium"
    LOW = "low"


@dataclass
class AnomalyAlert:
    id: str
    anomaly_type: AnomalyType
    severity: AnomalySeverity
    worker_name: str
    message: str
    details: Dict
    detected_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
    job_id: Optional[str] = None
    resolved_at: Optional[datetime] = None
    resolution: Optional[str] = None


@dataclass
 :
    worker_name: 
    status:  = 
    failure_rate:  = 
    queue_depth:  = 
    duration_ms:  = 
    expected_duration_ms:  = 
    timeout_count:  = 
    error_repeat_count:  = 
    error_message:  = 
    memory_mb:  = 
    cpu_percent:  = 



 :
    anomaly_type: AnomalyType
    severity: AnomalySeverity
    description: 
    check_fn: [[RuleContext], ]
    message_template: 
    cooldown_seconds: 


ANOMALY_RULES: [AnomalyRule] = [
    AnomalyRule(
        anomaly_type=AnomalyType.SLOW_JOB,
        severity=AnomalySeverity.MEDIUM,
        description=,
        check_fn= ctx: ctx.duration_ms > ctx.expected_duration_ms * ,
        message_template=,
        cooldown_seconds=,
    ),
    AnomalyRule(
        anomaly_type=AnomalyType.HIGH_FAILURE_RATE,
        severity=AnomalySeverity.HIGH,
        description=,
        check_fn= ctx: ctx.failure_rate > ,
        message_template=,
        cooldown_seconds=,
    ),
    AnomalyRule(
        anomaly_type=AnomalyType.WORKER_UNHEALTHY,
        severity=AnomalySeverity.CRITICAL,
        description=,
        check_fn= ctx: ctx.status == ,
        message_template=,
        cooldown_seconds=,
    ),
    AnomalyRule(
        anomaly_type=AnomalyType.REPEATED_ERROR,
        severity=AnomalySeverity.HIGH,
        description=,
        check_fn= ctx: ctx.error_repeat_count > ,
        message_template=,
        cooldown_seconds=,
    ),
]


 :
     ():
        ._alerts: [, AnomalyAlert] = {}
        ._cooldowns: [, datetime] = {}
        ._error_counts: [, [, ]] = {}
        ._timeout_counts: [, ] = {}
        ._alert_counter = 
    
     () -> [AnomalyAlert]:
        detected: [AnomalyAlert] = []
        
        failure_rate = (jobs_failed / jobs_processed * )  jobs_processed >   
        
        ctx = RuleContext(
            worker_name=worker_name,
            status=status,
            failure_rate=failure_rate,
            queue_depth=queue_depth,
            duration_ms=last_duration_ms,
            expected_duration_ms=expected_duration_ms,
            timeout_count=._timeout_counts.get(worker_name, ),
            memory_mb=memory_mb,
            cpu_percent=cpu_percent,
        )
        
         rule  ANOMALY_RULES:
             ._is_on_cooldown(worker_name, rule.anomaly_type):
                
            
             rule.check_fn(ctx):
                alert = ._create_alert(worker_name, rule, ctx)
                detected.append(alert)
                ._set_cooldown(worker_name, rule.anomaly_type, rule.cooldown_seconds)
        
         detected
    
     () -> [AnomalyAlert]:
        detected: [AnomalyAlert] = []
        
          success  error:
            ._track_error(worker_name, error)
        
        
         duration_ms > expected_duration_ms * :
              ._is_on_cooldown(worker_name, AnomalyType.SLOW_JOB):
                rule = ANOMALY_RULES[]
                ctx = RuleContext(
                    worker_name=worker_name,
                    duration_ms=duration_ms,
                    expected_duration_ms=expected_duration_ms,
                )
                alert = ._create_alert(worker_name, rule, ctx)
                alert.job_id = job_id
                detected.append(alert)
                ._set_cooldown(worker_name, AnomalyType.SLOW_JOB, )
        
        
         error:
            error_counts = ._error_counts.get(worker_name, {})
            count = error_counts.get(error[:], )
            
             count >    ._is_on_cooldown(worker_name, AnomalyType.REPEATED_ERROR):
                rule = (r  r  ANOMALY_RULES  r.anomaly_type == AnomalyType.REPEATED_ERROR)
                ctx = RuleContext(
                    worker_name=worker_name,
                    error_message=error[:],
                    error_repeat_count=count,
                )
                alert = ._create_alert(worker_name, rule, ctx)
                detected.append(alert)
                ._set_cooldown(worker_name, AnomalyType.REPEATED_ERROR, )
        
         detected
    
     () -> :
        alert = ._alerts.get(alert_id)
          alert  alert.resolved_at:
             
        
        alert.resolved_at = datetime.now(timezone.utc)
        alert.resolution = resolution
         
    
     () -> [AnomalyAlert]:
        severity_order = {: , : , : , : }
         (
            [a  a  ._alerts.values()   a.resolved_at],
            key= a: severity_order[a.severity.value]
        )
    
     () -> :
         worker_name   ._error_counts:
            ._error_counts[worker_name] = {}
        key = error[:]
        ._error_counts[worker_name][key] = ._error_counts[worker_name].get(key, ) + 
    
     () -> AnomalyAlert:
        ._alert_counter += 
        alert_id = 
        
        message = rule.message_template
         key, value  ctx.__dict__.items():
            message = message.replace(, (value))
        
        alert = AnomalyAlert(
            =alert_id,
            anomaly_type=rule.anomaly_type,
            severity=rule.severity,
            worker_name=worker_name,
            message=message,
            details=ctx.__dict__,
        )
        
        ._alerts[alert_id] = alert
         alert
    
     () -> :
        key = 
        cooldown_end = ._cooldowns.get(key)
         cooldown_end     cooldown_end > datetime.now(timezone.utc)
    
     () -> :
         datetime  timedelta
        key = 
        ._cooldowns[key] = datetime.now(timezone.utc) + timedelta(seconds=seconds)



_detector: [AnomalyDetector] = 

 () -> AnomalyDetector:
     _detector
     _detector  :
        _detector = AnomalyDetector()
     _detector

Usage Examples

Worker Job Monitoring

const detector = getAnomalyDetector();

async function executeJob(job: Job) {
  const startTime = Date.now();
  
  try {
    await processJob(job);
    const duration = Date.now() - startTime;
    
    const alerts = detector.checkJobExecution(
      'data-processor',
      job.id,
      duration,
      30000, // Expected 30s
      true
    );
    
    for (const alert of alerts) {
      await notifyOps(alert);
    }
  } catch (error) {
    const duration = Date.now() - startTime;
    
    const alerts = detector.checkJobExecution(
      'data-processor',
      job.id,
      duration,
      30000,
      false,
      error.message
    );
    
    for (const alert of alerts) {
      await notifyOps(alert);
    }
    throw error;
  }
}

Periodic Health Checks

setInterval(async () => {
  const health = await getWorkerHealth('data-processor');
  const alerts = detector.checkWorkerHealth('data-processor', health);
  
  for (const alert of alerts) {
    if (alert.severity === 'critical') {
      await pageOnCall(alert);
    } else {
      await notifySlack(alert);
    }
  }
}, 30000);

Best Practices

1. Set cooldowns long enough to prevent alert storms
2. Use severity levels to route alerts appropriately
3. Track error patterns to catch repeated failures
4. Clean up old resolved alerts periodically
5. Tune thresholds based on your baseline metrics

Common Mistakes

• Cooldowns too short (alert fatigue)
• No error pattern tracking (miss repeated failures)
• Same severity for all alerts (everything becomes noise)
• Not resolving alerts (dashboard becomes useless)
• Thresholds too sensitive (false positives)

Related Patterns

• health-checks - Source of health data for anomaly detection
• logging-observability - Structured logging for alert context
• graceful-degradation - Response to detected anomalies