Setting the file. One moment.
Baseline Stats · Authoring Log Alerts · PostHog/skills · Skills Docs
Back to the top of the page scripts/baseline_stats.py
scripts/ baseline_stats.py
Python · 164 lines · 5 KB
16
"interval": "7h"
17 }
18
19 Output (stdout): JSON with `stats` (p50/p95/p99/max), `suggested_threshold_count`
20 scaled to the alert window, and a `health` field flagging baselines that are too
21 sparse, too flat, or too spiky to alert on usefully.
22
23 Exit codes:
24 0 — stats produced
25 1 — invalid input (no ranges, malformed JSON, etc.)
26 """
27
28 from __future__ import annotations
29
30 import argparse
31 import json
32 import sys
33 from datetime import datetime
34 from typing import Any
35
36
37 def parse_args () -> argparse.Namespace:
38 p = argparse.ArgumentParser(
39 description = __doc__ ,
40 formatter_class = argparse.RawDescriptionHelpFormatter,
41 )
42 p.add_argument(
43 "--window-minutes" ,
44 type = int ,
45 required = True ,
46 choices = [ 5 , 10 , 15 , 30 , 60 ],
47 help = "Alert window in minutes (must match logs-alerts-create.window_minutes)." ,
48 )
49 p.add_argument(
50 "--floor" ,
51 type = int ,
52 default = 5 ,
53 help = "Minimum threshold (default: 5). Stops the suggestion collapsing on tiny services." ,
54 )
55 p.add_argument(
56 "--min-buckets" ,
57 type = int ,
58 default = 12 ,
59 help = "Minimum non-empty buckets for a useful baseline (default: 12)." ,
60 )
61 return p.parse_args()
62
63
64 def parse_iso (s: str ) -> datetime:
65 # logs-count-ranges currently returns naive ISO; some clients add Z.
66 cleaned = s.replace( "Z" , "+00:00" )
67 return datetime.fromisoformat(cleaned)
68
69
70 def percentile (sorted_counts: list[ int ], q: float ) -> float :
71 # Matches numpy default (linear interpolation between nearest ranks).
72 if not sorted_counts:
73 return 0.0
74 if len (sorted_counts) == 1 :
75 return float (sorted_counts[ 0 ])
76 rank = q * ( len (sorted_counts) - 1 )
77 lo = int (rank)
78 hi = min (lo + 1 , len (sorted_counts) - 1 )
79 frac = rank - lo
80 return sorted_counts[lo] * ( 1 - frac) + sorted_counts[hi] * frac
81
82
83 def main () -> int :
84 args = parse_args()
85
86 try :
87 data = json.load(sys.stdin)
88 except json.JSONDecodeError as e:
89 print ( f "Could not parse stdin as JSON: { e } " , file = sys.stderr)
90 return 1
91
92 ranges = data.get( "ranges" ) if isinstance (data, dict ) else None
93 if not ranges:
94 print (
95 "No buckets in input — `ranges` is empty or missing. "
96 "Either the filter matched nothing, or you piped the wrong response." ,
97 file = sys.stderr,
98 )
99 return 1
100
101 counts = [r[ "count" ] for r in ranges if isinstance (r, dict ) and "count" in r]
102 if not counts:
103 print ( "Bucket entries are missing `count` fields." , file = sys.stderr)
104 return 1
105
106 try :
107 first = ranges[ 0 ]
108 bucket_minutes = (parse_iso(first[ "date_to" ]) - parse_iso(first[ "date_from" ])).total_seconds() / 60
109 except ( KeyError , ValueError ) as e:
110 print ( f "Could not derive bucket width from first range: { e } " , file = sys.stderr)
111 return 1
112
113 if bucket_minutes <= 0 :
114 print ( "Bucket width is non-positive — input looks corrupt." , file = sys.stderr)
115 return 1
116
117 sorted_counts = sorted (counts)
118 n = len (counts)
119
120 mid = n // 2
121 p50 = float (sorted_counts[mid]) if n % 2 else (sorted_counts[mid - 1 ] + sorted_counts[mid]) / 2
122 p95 = percentile(sorted_counts, 0.95 )
123 p99 = percentile(sorted_counts, 0.99 )
124 bucket_max = sorted_counts[ - 1 ]
125
126 bucket_threshold = max (p99, p50 * 3 , args.floor)
127 scale = args.window_minutes / bucket_minutes
128 suggested = max (args.floor, round (bucket_threshold * scale))
129
130 health: list[ str ] = []
131 if n < args.min_buckets:
132 health.append( f "sparse: { n } _of_ { args.min_buckets } _buckets" )
133 if bucket_max == 0 :
134 health.append( "empty" )
135 elif p95 > 0 and bucket_max / p95 >= 10 :
136 health.append( "spiky" )
137 elif p50 > 0 and (p95 / p50) <= 1.5 :
138 health.append( "flat" )
139
140 output: dict[ str , Any] = {
141 "n_buckets" : n,
142 "bucket_minutes" : round (bucket_minutes, 2 ),
143 "alert_window_minutes" : args.window_minutes,
144 "stats" : {
145 "p50" : round (p50, 2 ),
146 "p95" : round (p95, 2 ),
147 "p99" : round (p99, 2 ),
148 "max" : bucket_max,
149 },
150 "suggested_threshold_count" : suggested,
151 "rationale" : (
152 f "max(p99= { round (p99, 2 ) } , median*3= { round (p50 * 3 , 2 ) } , floor= { args.floor } ) "
153 f "scaled from { bucket_minutes :.0f} m bucket to { args.window_minutes } m window"
154 ),
155 "health" : health,
156 }
157
158 json.dump(output, sys.stdout, indent = 2 )
159 sys.stdout.write( " \n " )
160 return 0
161
162
163 if __name__ == "__main__" :
164 sys.exit(main())