1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
//! Log- and event-tail spawns and their key handlers. The scroll /
//! follow / filter surface both tails share is `super::tail::TailView`.
use super::*;
impl App {
/// Key handler for the `:logs-tail` streaming overlay. j/k scroll, G
/// snaps back to follow-mode (auto-tail), g jumps to top (and pauses
/// follow), / opens a regex filter, n clears it, esc/q closes the
/// overlay and tears down the polling task.
pub(crate) fn handle_log_tail_key(&mut self, key: KeyEvent) {
// Group-switcher: Tab opens a Picker over the env's discovered CW
// log groups. Handled up-front before the borrow of
// `current_overlay` below so the picker open can re-borrow `self`.
// (In filter-entry mode Tab is input, not the switcher.)
if matches!(key.code, KeyCode::Tab) {
let in_filter = matches!(
self.current_overlay.as_ref(),
Some(Overlay::LogTail { view, .. }) if view.filter_active
);
if !in_filter {
self.open_log_group_picker();
return;
}
}
let Some(Overlay::LogTail { view, events, .. }) = self.current_overlay.as_mut() else {
return;
};
let outcome = tail::handle_tail_key(view, key);
// Clamp the scroll to the buffer: `g` sets the u16::MAX
// sentinel, and without a ceiling every subsequent `j` costs
// one dead press (~63k of them) before the view moves again.
view.scroll = view.scroll.min(events.len() as u16);
if outcome == tail::TailKeyOutcome::Close {
// Reap so a late `LogTailOpened` from the aborted task can't
// re-open the overlay the user just dismissed.
tail::reap_tail_task(&mut self.log_tail_task, &mut self.log_tail_session);
self.current_overlay = None;
}
}
/// Open a Picker over the env's discovered CW log groups so the operator
/// can switch the tailed group from inside the streaming overlay.
/// Pre-selects the currently-tailed group; no-op (with a status hint) if
/// no groups have been discovered for this env.
fn open_log_group_picker(&mut self) {
let Some(Overlay::LogTail { log_group, .. }) = self.current_overlay.as_ref() else {
return;
};
let current_group = log_group.clone();
let groups: Vec<String> = self
.detail
.as_ref()
.and_then(|d| d.cw_log_groups.clone())
.unwrap_or_default();
if groups.is_empty() {
self.status_message = Some(
"no CW log groups discovered for this env — try `:logs-tail <full-group-name>`"
.into(),
);
return;
}
self.picker = Some(Picker::new(
PickerKind::LogGroup,
groups,
Some(current_group.as_str()),
));
self.mode = Mode::Picker;
}
/// Open a streaming CW Logs view for `env_name`. If `explicit_group` is
/// `None`, discovers the env's log groups and picks the most useful one
/// via `pick_default_log_group`. Aborts any active log-tail task before
/// starting the new one, then spawns a polling loop that sends
/// `AppMsg::LogTailEvents` every ~2s. The overlay opens immediately in
/// a "discovering" state and gets replaced with the LogTail variant
/// once the group is known.
pub(crate) fn spawn_logs_tail(&mut self, env_name: String, explicit_group: Option<String>) {
// Tear down any prior session so we don't have two pollers racing.
tail::reap_tail_task(&mut self.log_tail_task, &mut self.log_tail_session);
let session_id = self.log_tail_session;
let client = self.client_for_env(&env_name);
let tx = self.msg_tx.clone();
let gen = self.generation;
let env_for_msg = env_name.clone();
// In-flight ack: the LogTail overlay opens itself when data lands.
let handle = tokio::spawn(async move {
// Resolve the client once for the whole poll loop — a tail
// that re-resolved per poll would re-assume every two
// seconds under `:account`.
let aws = match client.resolve().await {
Ok(aws) => aws,
Err(e) => {
let _ = tx.send(AppMsg::LogTailEvents {
gen,
session_id,
next_since_ms: 0,
result: Err(flatten_err("cached_client", e)),
});
return;
}
};
// Resolve the log group up front. If the user supplied one,
// trust it (no DescribeLogGroups round-trip); otherwise discover.
let group = match explicit_group {
Some(g) => g,
None => match aws.discover_env_log_groups(&env_for_msg).await {
Ok(groups) => match pick_default_log_group(&groups) {
Some(g) => g,
None => {
let _ = tx.send(AppMsg::LogTailEvents {
gen,
session_id,
next_since_ms: 0,
result: Err(format!(
"no CW log groups under /aws/elasticbeanstalk/{env_for_msg}/ — enable streaming with `:logs-stream on`"
)),
});
return;
}
},
Err(e) => {
let _ = tx.send(AppMsg::LogTailEvents {
gen,
session_id,
next_since_ms: 0,
result: Err(format!("discover log groups: {e}")),
});
return;
}
},
};
// First batch: fetch the last 5 minutes so the overlay isn't
// empty on open.
let mut since_ms = chrono::Utc::now().timestamp_millis() - 5 * 60 * 1000;
// Send an "opening" message that tells the App handler what log
// group resolved + replaces the overlay with a real LogTail.
let _ = tx.send(AppMsg::LogTailOpened {
gen,
session_id,
env_name: env_for_msg.clone(),
log_group: group.clone(),
since_ms,
});
let mut boundary_ids: std::collections::HashSet<String> = Default::default();
loop {
match aws
.fetch_recent_log_events(&group, since_ms, 1000, &boundary_ids)
.await
{
Ok((events, next_since, carry)) => {
let next_since_ms = next_since;
let _ = tx.send(AppMsg::LogTailEvents {
gen,
session_id,
next_since_ms,
result: Ok(events),
});
since_ms = next_since;
boundary_ids = carry;
}
Err(e) => {
let _ = tx.send(AppMsg::LogTailEvents {
gen,
session_id,
next_since_ms: since_ms,
result: Err(format!("{e}")),
});
// Keep going on errors — transient throttling
// shouldn't kill the session.
}
}
tokio::time::sleep(std::time::Duration::from_secs(2)).await;
}
});
self.log_tail_task = Some(handle);
}
/// `:event-tail` — open the cross-fleet event tail overlay and
/// start its polling task. First batch is the fleet's most recent
/// events regardless of age (so the overlay isn't empty on open);
/// subsequent polls pass a `start_time` watermark so a busy fleet
/// re-ships only what's new. DescribeEvents is more
/// throttle-sensitive than FilterLogEvents, hence the 5s cadence
/// (vs logs-tail's 2s). Errors keep the loop alive.
pub(crate) fn spawn_event_tail(&mut self) {
// Tear down any prior session so we don't have two pollers racing.
tail::reap_tail_task(&mut self.event_tail_task, &mut self.event_tail_session);
let session_id = self.event_tail_session;
let aws = self.aws.clone();
let tx = self.msg_tx.clone();
let gen = self.generation;
let handle = tokio::spawn(async move {
// Install the (empty) overlay first so the operator sees
// the tail open immediately; the first batch fills it.
let _ = tx.send(AppMsg::EventTailOpened { gen, session_id });
let mut since_ms = match aws.list_events(EVENT_TAIL_FIRST_BATCH).await {
Ok(mut events) => {
// DescribeEvents returns newest-first; the ring
// buffer appends oldest-first.
events.reverse();
// Watermark = newest event + 1ms, NOT local now():
// clamping to now() would skip anything that lands
// between the server-side snapshot and our clock
// (eventual consistency / clock skew). Events in
// (newest, now] weren't in this batch, so there's
// no duplicate risk. Empty fleet history falls
// back to now.
let watermark = if events.is_empty() {
chrono::Utc::now().timestamp_millis()
} else {
next_event_watermark_ms(&events, 0)
};
let _ = tx.send(AppMsg::EventTailEvents {
gen,
session_id,
result: Ok(events),
});
watermark
}
Err(e) => {
let _ = tx.send(AppMsg::EventTailEvents {
gen,
session_id,
result: Err(format!("{e}")),
});
chrono::Utc::now().timestamp_millis()
}
};
loop {
tokio::time::sleep(std::time::Duration::from_secs(5)).await;
match aws.list_events_since(since_ms, EVENT_TAIL_POLL_BATCH).await {
Ok((mut events, truncated)) => {
since_ms = next_event_watermark_ms(&events, since_ms);
events.reverse();
if truncated {
// The poll filled its page budget with a
// token still in hand. DescribeEvents
// returns newest-first, so what we didn't
// fetch is OLDER than everything here — and
// the watermark has just moved past it, so
// no later poll can reach it.
//
// Two things had to be true for the marker
// to actually reach the operator, and
// neither was: a truncated poll can carry
// more events than the overlay's ring
// holds, so the marker (inserted first,
// oldest) was evicted by its own batch; and
// it carries no env or application, so any
// active filter dropped it. Trim to fit,
// keeping the newest — the older ones would
// be evicted anyway — and stamp it with a
// severity the filter exempts.
let cap = crate::app::EVENT_TAIL_MAX_EVENTS.saturating_sub(1);
let dropped_locally = events.len().saturating_sub(cap);
if dropped_locally > 0 {
events.drain(..dropped_locally);
}
let detail = if dropped_locally > 0 {
format!(
" ({dropped_locally} more fetched but beyond the \
overlay's {}-event buffer)",
crate::app::EVENT_TAIL_MAX_EVENTS
)
} else {
String::new()
};
events.insert(
0,
crate::aws::Event {
at: None,
env: String::new(),
application: String::new(),
message: format!(
"… older events in this window were not fetched \
(per-poll batch limit reached){detail}"
),
severity: crate::app::EVENT_TAIL_GAP_SEVERITY.into(),
version_label: None,
},
);
}
let _ = tx.send(AppMsg::EventTailEvents {
gen,
session_id,
result: Ok(events),
});
}
Err(e) => {
let _ = tx.send(AppMsg::EventTailEvents {
gen,
session_id,
result: Err(format!("{e}")),
});
// Keep going on errors — transient throttling
// shouldn't kill the session.
}
}
}
});
self.event_tail_task = Some(handle);
}
/// Key handling while the `:event-tail` overlay is open — the
/// same surface as [`handle_log_tail_key`] minus the log-group
/// picker (there's no group to switch; the tail is fleet-wide).
pub(crate) fn handle_event_tail_key(&mut self, key: KeyEvent) {
let Some(Overlay::EventTail { view, events, .. }) = self.current_overlay.as_mut() else {
return;
};
let outcome = tail::handle_tail_key(view, key);
// Same scroll ceiling as the log tail — `g`'s u16::MAX
// sentinel must not leave `j` dead for thousands of presses.
view.scroll = view.scroll.min(events.len() as u16);
if outcome == tail::TailKeyOutcome::Close {
// Reap so a late `EventTailOpened` from the aborted task
// can't re-open the dismissed overlay.
tail::reap_tail_task(&mut self.event_tail_task, &mut self.event_tail_session);
self.current_overlay = None;
}
}
}