Your question is Real-Time Sliding Window Scan. Start with the requirements on the right.
Run and submit as often as you like. When you're ready, talk me through your approach or go straight to the code.
Splunk Observability Cloud receives events in nondecreasing timestamp order. Implement a realtime scan that finds the time window containing the largest number of ERROR events within a given duration.
Implement busiest_error_window(events, window_seconds), where events is a list of two-element lists [timestamp, level]. Timestamps are integer seconds, levels are strings, and events are already sorted by timestamp. A valid window includes error events whose timestamps differ by at most window_seconds, so both endpoints are inclusive.
Return [start_timestamp, end_timestamp, error_count] for the densest window. If multiple windows have the same count, return the one with the earliest start timestamp. The start and end timestamps must be the timestamps of the first and last error events in the selected window. Return [None, None, 0] when the stream contains no error events.
Example 1: events = [[1, "INFO"], [2, "ERROR"], [4, "ERROR"], [7, "ERROR"]], window_seconds = 3 returns [2, 4, 2], because timestamps 2 and 4 are within three seconds, while timestamps 2 and 7 are not.
Example 2: events = [[5, "ERROR"], [6, "WARN"], [6, "ERROR"], [9, "ERROR"]], window_seconds = 1 returns [5, 6, 2], because the first two error events occur within one second.
def busiest_error_window(events, window_seconds):