1. <?php
  3. /**
  4. * Flags anomalies in message queues for scheduled runs.
  5. * Supports older PHP versions (PHP 5.6+).
  6. *
  7. * @param array $queue_data Array of message queue data. Each element should be an array
  8. * with keys like 'message_id', 'timestamp', 'processing_time', 'status'.
  9. * @param array $thresholds Array of anomaly thresholds. Defaults provided.
  10. * @return array Array of flagged anomalies. Each element is an array with details about the anomaly.
  11. */
  12. function flagQueueAnomalies(array $queue_data, array $thresholds = []): array
  13. {
  14. // Default thresholds
  15. if (empty($thresholds)) {
  16. $thresholds = [
  17. 'processing_time_high' => 5, // seconds
  18. 'processing_time_low' => 0.1, // seconds
  19. 'message_count_high' => 100, // messages
  20. 'message_count_low' => 1, // messages
  21. ];
  22. }
  25. $anomalies = [];
  27. if (empty($queue_data)) {
  28. return $anomalies; // Return empty array if no data
  29. }
  31. $previous_processing_times = [];
  32. $previous_message_counts = [];
  34. foreach ($queue_data as $item) {
  35. // Validate data structure (basic check)
  36. if (!is_array($item) || !isset($item['timestamp']) || !isset($item['processing_time']) || !isset($item['status'])) {
  37. continue; // Skip invalid data
  38. }
  40. $timestamp = $item['timestamp'];
  41. $processing_time = $item['processing_time'];
  42. $status = $item['status'];
  45. // Processing Time Anomaly Detection
  46. if ($processing_time > $thresholds['processing_time_high']) {
  47. $anomalies[] = [
  48. 'message_id' => $item['message_id'],
  49. 'timestamp' => $timestamp,
  50. 'anomaly_type' => 'processing_time_high',
  51. 'value' => $processing_time,
  52. 'threshold' => $thresholds['processing_time_high'],
  53. ];
  54. } elseif ($processing_time < $thresholds['processing_time_low']) {
  55. $anomalies[] = [
  56. 'message_id' => $item['message_id'],
  57. 'timestamp' => $timestamp,
  58. 'anomaly_type' => 'processing_time_low',
  59. 'value' => $processing_time,
  60. 'threshold' => $thresholds['processing_time_low'],
  61. ];
  62. }
  64. // Message Count Anomaly Detection
  65. if (count($queue_data) > 1 && (count($queue_data) - $previous_message_counts) > $thresholds['message_count_high']) {
  66. $anomalies[] = [
  67. 'message_id' => 'general', // Use general for aggregate anomalies
  68. 'timestamp' => $timestamp,
  69. 'anomaly_type' => 'message_count_high',
  70. 'value' => count($queue_data),
  71. 'threshold' => $thresholds['message_count_high'],
  72. ];
  73. }
  74. if (count($queue_data) > 1 && (count($queue_data) - $previous_message_counts) < $thresholds['message_count_low']) {
  75. $anomalies[] = [
  76. 'message_id' => 'general', // Use general for aggregate anomalies
  77. 'timestamp' => $timestamp,
  78. 'anomaly_type' => 'message_count_low',
  79. 'value' => count($queue_data),
  80. 'threshold' => $thresholds['message_count_low'],
  81. ];
  82. }
  84. $previous_processing_times[] = $processing_time;
  85. $previous_message_counts[] = count($queue_data);
  86. }
  88. return $anomalies;
  89. }

Add your comment