k8s_controller/
events.rs

1//! Publishing Kubernetes [events](https://kubernetes.io/docs/reference/kubernetes-api/cluster-resources/event-v1/)
2//! about the resources a controller reconciles.
3//!
4//! Events are what `kubectl describe` shows beneath an object, which makes
5//! them the place to explain to someone without access to the controller's
6//! logs why an object is not converging.
7//!
8//! An [`EventRecorder`] can be given to a [`Controller`](crate::Controller)
9//! via [`with_event_recorder`](crate::Controller::with_event_recorder), in
10//! which case every failed reconciliation is published as an event on the
11//! resource being reconciled (see [`Context::failure_event`](crate::Context::failure_event)).
12//! Reconcilers can publish events of their own through
13//! [`TraceMetadata::publish_event`](crate::TraceMetadata::publish_event), or
14//! through [`EventRecorder::publish`] directly, which also works outside of
15//! reconciliation (for instance, from a background task).
16//!
17//! Each controller should have its own recorder, whose [`Reporter`] names
18//! that controller, even when several controllers in a process reconcile
19//! the same kind of resource. The reporter is the only thing in an event
20//! that identifies which controller published it.
21//!
22//! # Aggregation
23//!
24//! Publishing an event identical to the one last published for the same
25//! object, reason, and action, within [`SERIES_WINDOW`], does not create a
26//! new event object. Instead it increments the `series.count` of the existing
27//! one, so that a reconciliation retrying on a backoff does not bury its
28//! resource in identical events. Any difference, including in the note,
29//! starts a new event, so that an event always describes the most recent
30//! occurrence accurately.
31//!
32//! Failure events published by a controller additionally start afresh after
33//! the resource next reconciles successfully, so a failure that recurs after
34//! a recovery is reported as a new event rather than as a continuation of
35//! the old one.
36//!
37//! # Timeouts
38//!
39//! Each publish is bounded by the recorder's [timeout](EventRecorder::with_timeout),
40//! so that an unresponsive API server delays the reconciliation publishing
41//! an event by at most that long.
42//!
43//! # RBAC
44//!
45//! Publishing requires `create` and `patch` permissions on `events` in the
46//! `events.k8s.io` API group, in every namespace the controller reconciles
47//! resources in. Events about cluster-scoped resources are created in the
48//! `default` namespace.
49
50use std::collections::HashMap;
51use std::sync::{Arc, Mutex};
52use std::time::{Duration, Instant};
53
54use k8s_openapi::api::core::v1::ObjectReference;
55use k8s_openapi::api::events::v1::Event as KubeEvent;
56use k8s_openapi::apimachinery::pkg::apis::meta::v1::MicroTime;
57use k8s_openapi::jiff::Timestamp;
58use kube::api::{Api, ObjectMeta, Patch, PatchParams, PostParams};
59use kube::{Client, Resource, ResourceExt};
60
61pub use kube_runtime::events::{EventType, Reporter};
62
63/// How long after an event was last published that publishing it again
64/// still aggregates into it, rather than creating a new event.
65///
66/// This matches the aggregation window of client-go's event correlator.
67pub const SERIES_WINDOW: Duration = Duration::from_secs(10 * 60);
68
69/// The Kubernetes API server rejects events whose note is longer than this
70/// many bytes.
71pub const MAX_NOTE_BYTES: usize = 1024;
72
73/// The default [timeout](EventRecorder::with_timeout) for publishing an
74/// event.
75pub const DEFAULT_TIMEOUT: Duration = Duration::from_secs(5);
76
77/// Namespace used for events about cluster-scoped resources, which have no
78/// namespace of their own.
79const CLUSTER_EVENT_NAMESPACE: &str = "default";
80
81/// An event to publish about a resource.
82///
83/// `reason` and `action` must each be at most 128 characters, or the API
84/// server will reject the event.
85#[derive(Clone, Debug, PartialEq)]
86pub struct Event {
87    /// Whether this is an ordinary event or a problem. Shown as `Type` by
88    /// `kubectl describe`.
89    pub type_: EventType,
90    /// Why the event happened, as a machine-readable `PascalCase` word (for
91    /// instance `ReconcileFailed`). Shown as `Reason` by `kubectl describe`,
92    /// and what tools and dashboards group events by, so it should be
93    /// treated as a stable interface.
94    pub reason: String,
95    /// What was being done when the event happened, as a machine-readable
96    /// `PascalCase` word (for instance `Reconcile`). Not shown by
97    /// `kubectl describe`.
98    pub action: String,
99    /// A human-readable description of what happened. Shown as `Message` by
100    /// `kubectl describe`. Truncated to [`MAX_NOTE_BYTES`] if longer.
101    pub note: Option<String>,
102    /// Another object involved in the event, if any.
103    pub related: Option<ObjectReference>,
104}
105
106/// An error publishing an event.
107#[derive(Debug, thiserror::Error)]
108#[non_exhaustive]
109pub enum PublishError {
110    /// The API server rejected the event, or could not be reached.
111    #[error(transparent)]
112    Kube(#[from] kube::Error),
113    /// Publishing did not complete within the recorder's
114    /// [timeout](EventRecorder::with_timeout).
115    #[error("timed out after {0:?} publishing event")]
116    Timeout(Duration),
117}
118
119/// Publishes Kubernetes events on resources on behalf of one controller,
120/// aggregating repeats as described in the [module documentation](self).
121///
122/// Share a recorder (via [`Arc`]) between a [`Controller`](crate::Controller)
123/// and whatever else publishes events on that controller's behalf, but not
124/// between controllers.
125pub struct EventRecorder {
126    client: Client,
127    reporter: Reporter,
128    timeout: Duration,
129    series: Mutex<HashMap<SeriesKey, Slot>>,
130}
131
132/// The series published under one [`SeriesKey`], locked for the whole of a
133/// publish so that concurrent publishes of the same event cannot both
134/// create it, or both patch it to the same count.
135type Slot = Arc<tokio::sync::Mutex<Option<Series>>>;
136
137/// Identifies the events that can aggregate into one another.
138#[derive(Clone, Debug, PartialEq, Eq, Hash)]
139struct SeriesKey {
140    /// Whether these are the failure events the controller publishes when
141    /// reconciling the resource fails, as opposed to events published
142    /// through [`EventRecorder::publish`] or
143    /// [`TraceMetadata::publish_event`](crate::TraceMetadata::publish_event).
144    failure: bool,
145    uid: String,
146    reason: String,
147    action: String,
148}
149
150/// The last event published under a [`SeriesKey`].
151struct Series {
152    event: Event,
153    namespace: String,
154    name: String,
155    count: i32,
156    last_published: Instant,
157}
158
159impl std::fmt::Debug for EventRecorder {
160    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
161        f.debug_struct("EventRecorder")
162            .field("reporter", &self.reporter)
163            .field("timeout", &self.timeout)
164            .finish_non_exhaustive()
165    }
166}
167
168impl EventRecorder {
169    /// Creates a recorder that publishes events as `reporter`, with the
170    /// [default timeout](DEFAULT_TIMEOUT).
171    ///
172    /// [`Reporter::controller`] becomes each event's `reportingController`,
173    /// shown by `kubectl describe` as where the event came from. It should
174    /// name the individual controller, and must be a qualified name (for
175    /// instance `"my-operator.example.com/widgets"`), or the API server
176    /// rejects the event. [`Reporter::instance`] becomes each event's
177    /// `reportingInstance`, and should identify the replica, for which the
178    /// pod name is a good choice; if it is `None`, the controller name is
179    /// used instead.
180    pub fn new(client: Client, reporter: Reporter) -> Self {
181        Self {
182            client,
183            reporter,
184            timeout: DEFAULT_TIMEOUT,
185            series: Mutex::new(HashMap::new()),
186        }
187    }
188
189    /// Sets how long publishing an event may take, including any wait for
190    /// a concurrent publish of the same event, before giving up with
191    /// [`PublishError::Timeout`].
192    pub fn with_timeout(mut self, timeout: Duration) -> Self {
193        self.timeout = timeout;
194        self
195    }
196
197    /// The reporter this recorder publishes events as.
198    pub fn reporter(&self) -> &Reporter {
199        &self.reporter
200    }
201
202    /// Publishes `event` about `resource`.
203    ///
204    /// A note longer than [`MAX_NOTE_BYTES`] is truncated to fit rather than
205    /// having the whole event rejected.
206    ///
207    /// Events are informational, so callers should usually log a failure to
208    /// publish one rather than let it change what they do next.
209    pub async fn publish<K>(&self, resource: &K, event: &Event) -> Result<(), PublishError>
210    where
211        K: Resource,
212        K::DynamicType: Default,
213    {
214        let reference = resource.object_ref(&Default::default());
215        self.publish_to(&reference, event, false).await
216    }
217
218    /// Forgets every event published about `resource`, so that the next
219    /// event published about it creates a new event object even if it is
220    /// identical to the last one.
221    ///
222    /// This is useful when an event reports a state that the resource can
223    /// leave and later re-enter, and each entry should be reported as a
224    /// distinct occurrence.
225    pub fn forget<K: Resource>(&self, resource: &K) {
226        if let Some(uid) = resource.meta().uid.as_deref() {
227            self.series.lock().unwrap().retain(|key, _| key.uid != uid);
228        }
229    }
230
231    /// Forgets the failure events published about the object with the given
232    /// `uid`.
233    pub(crate) fn forget_failures(&self, uid: &str) {
234        self.series
235            .lock()
236            .unwrap()
237            .retain(|key, _| key.uid != uid || !key.failure);
238    }
239
240    /// Publishes `event` about the object `reference` refers to, as a
241    /// failure event if `failure` is set.
242    pub(crate) async fn publish_to(
243        &self,
244        reference: &ObjectReference,
245        event: &Event,
246        failure: bool,
247    ) -> Result<(), PublishError> {
248        tokio::time::timeout(
249            self.timeout,
250            self.publish_unbounded(reference, event, failure),
251        )
252        .await
253        .map_err(|_| PublishError::Timeout(self.timeout))?
254        .map_err(PublishError::Kube)
255    }
256
257    async fn publish_unbounded(
258        &self,
259        reference: &ObjectReference,
260        event: &Event,
261        failure: bool,
262    ) -> Result<(), kube::Error> {
263        let event = Event {
264            note: event.note.as_deref().map(truncate_note),
265            ..(*event).clone()
266        };
267        let namespace = reference
268            .namespace
269            .clone()
270            .unwrap_or_else(|| CLUSTER_EVENT_NAMESPACE.to_owned());
271        let Some(uid) = reference.uid.clone() else {
272            self.create(&namespace, reference, &event).await?;
273            return Ok(());
274        };
275        let slot = self.slot(SeriesKey {
276            failure,
277            uid,
278            reason: event.reason.clone(),
279            action: event.action.clone(),
280        });
281        let mut series = slot.lock().await;
282
283        if let Some(series) = series
284            .as_mut()
285            .filter(|s| s.event == event && s.last_published.elapsed() < SERIES_WINDOW)
286        {
287            let count = series.count.saturating_add(1);
288            match self
289                .patch_series(&series.namespace, &series.name, count)
290                .await
291            {
292                Ok(()) => {
293                    series.count = count;
294                    series.last_published = Instant::now();
295                    return Ok(());
296                }
297                // The API server deletes events some time after their last
298                // update (an hour, by default), which a long-running series
299                // can outlive. Start a new one in that case.
300                Err(kube::Error::Api(e)) if e.code == 404 => {}
301                Err(e) => return Err(e),
302            }
303        }
304
305        let created = self.create(&namespace, reference, &event).await?;
306        *series = Some(Series {
307            event,
308            namespace,
309            name: created.name_any(),
310            count: 1,
311            last_published: Instant::now(),
312        });
313        Ok(())
314    }
315
316    /// Returns the slot for `key`, evicting the slots of series that can no
317    /// longer be aggregated into.
318    fn slot(&self, key: SeriesKey) -> Slot {
319        let mut slots = self.series.lock().unwrap();
320        let now = Instant::now();
321        slots.retain(|k, slot| {
322            *k == key
323                || slot.try_lock().map_or(true, |series| {
324                    series
325                        .as_ref()
326                        .is_some_and(|s| now.duration_since(s.last_published) < SERIES_WINDOW)
327                })
328        });
329        Arc::clone(slots.entry(key).or_default())
330    }
331
332    async fn create(
333        &self,
334        namespace: &str,
335        reference: &ObjectReference,
336        event: &Event,
337    ) -> Result<KubeEvent, kube::Error> {
338        let api: Api<KubeEvent> = Api::namespaced(self.client.clone(), namespace);
339        let now = Timestamp::now();
340        let event = KubeEvent {
341            metadata: ObjectMeta {
342                name: Some(event_name(reference.name.as_deref(), now)),
343                namespace: Some(namespace.to_owned()),
344                ..Default::default()
345            },
346            action: Some(event.action.clone()),
347            reason: Some(event.reason.clone()),
348            note: event.note.clone(),
349            type_: Some(
350                match event.type_ {
351                    EventType::Normal => "Normal",
352                    EventType::Warning => "Warning",
353                }
354                .to_owned(),
355            ),
356            event_time: Some(MicroTime(now)),
357            regarding: Some(reference.clone()),
358            related: event.related.clone(),
359            reporting_controller: Some(self.reporter.controller.clone()),
360            reporting_instance: Some(
361                self.reporter
362                    .instance
363                    .clone()
364                    .unwrap_or_else(|| self.reporter.controller.clone()),
365            ),
366            ..Default::default()
367        };
368        api.create(&PostParams::default(), &event).await
369    }
370
371    async fn patch_series(
372        &self,
373        namespace: &str,
374        name: &str,
375        count: i32,
376    ) -> Result<(), kube::Error> {
377        let api: Api<KubeEvent> = Api::namespaced(self.client.clone(), namespace);
378        let patch = serde_json::json!({
379            "series": {
380                "count": count,
381                "lastObservedTime": MicroTime(Timestamp::now()),
382            },
383        });
384        api.patch(name, &PatchParams::default(), &Patch::Merge(&patch))
385            .await?;
386        Ok(())
387    }
388}
389
390/// The name of a new event about the object named `regarding`, published at
391/// `time`: `<regarding>.<time in hex nanoseconds>`, as client-go names them,
392/// or `event.<time in hex nanoseconds>` if that would not be a valid name.
393///
394/// Not every object's name forms a valid event name: some kinds only require
395/// names to be path segments (RBAC roles may be named `system:foo` or `Foo`),
396/// and a long name leaves no room for the suffix. The event's `regarding`
397/// reference identifies the object either way.
398///
399/// NOTE: `generateName` cannot produce this shape. The API server validates
400/// the `generateName` prefix as a DNS subdomain in its own right, which may
401/// end in `-` but not in `.`, so a `<regarding>.` prefix fails every create
402/// with a 422.
403fn event_name(regarding: Option<&str>, time: Timestamp) -> String {
404    let suffix = format!(".{:x}", time.as_nanosecond());
405    regarding
406        .map(|regarding| format!("{regarding}{suffix}"))
407        .filter(|name| is_dns_subdomain(name))
408        .unwrap_or_else(|| format!("event{suffix}"))
409}
410
411/// Whether `name` is a lowercase RFC 1123 subdomain: at most 253 bytes of
412/// `.`-separated labels, each at most 63 bytes of lowercase alphanumerics and
413/// `-`, starting and ending with an alphanumeric.
414///
415/// NOTE: The API server's check for event names (`IsDNS1123Subdomain`) does
416/// not limit the length of labels. Enforcing it anyway keeps this check at
417/// least as strict as the API server's.
418pub(crate) fn is_dns_subdomain(name: &str) -> bool {
419    name.len() <= 253
420        && name.split('.').all(|label| {
421            let bytes = label.as_bytes();
422            (1..=63).contains(&bytes.len())
423                && bytes[0].is_ascii_alphanumeric()
424                && bytes[bytes.len() - 1].is_ascii_alphanumeric()
425                && bytes
426                    .iter()
427                    .all(|&b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'-')
428        })
429}
430
431/// Shortens `note` to at most [`MAX_NOTE_BYTES`], on a character boundary.
432fn truncate_note(note: &str) -> String {
433    if note.len() <= MAX_NOTE_BYTES {
434        return note.to_owned();
435    }
436    const ELLIPSIS: &str = "...";
437    let mut end = MAX_NOTE_BYTES - ELLIPSIS.len();
438    while !note.is_char_boundary(end) {
439        end -= 1;
440    }
441    format!("{}{ELLIPSIS}", &note[..end])
442}
443
444#[cfg(test)]
445mod tests {
446    use super::*;
447    use crate::test_util::{MockApiServer, block_on};
448
449    use k8s_openapi::api::core::v1::ConfigMap;
450    use serde_json::json;
451
452    fn config_map(uid: &str) -> ConfigMap {
453        ConfigMap {
454            metadata: ObjectMeta {
455                name: Some("cm".to_owned()),
456                namespace: Some("ns".to_owned()),
457                uid: Some(uid.to_owned()),
458                ..Default::default()
459            },
460            ..Default::default()
461        }
462    }
463
464    fn event(note: &str) -> Event {
465        Event {
466            type_: EventType::Warning,
467            reason: "Broken".to_owned(),
468            action: "Reconcile".to_owned(),
469            note: Some(note.to_owned()),
470            related: None,
471        }
472    }
473
474    fn recorder(server: &MockApiServer) -> EventRecorder {
475        EventRecorder::new(
476            server.client(),
477            Reporter {
478                controller: "test.example.com".to_owned(),
479                instance: Some("pod-0".to_owned()),
480            },
481        )
482    }
483
484    #[test]
485    fn test_truncate_note() {
486        assert_eq!(truncate_note("short"), "short");
487
488        let exact = "a".repeat(MAX_NOTE_BYTES);
489        assert_eq!(truncate_note(&exact), exact);
490
491        let long = "a".repeat(MAX_NOTE_BYTES + 1);
492        let truncated = truncate_note(&long);
493        assert_eq!(truncated.len(), MAX_NOTE_BYTES);
494        assert!(truncated.ends_with("..."));
495
496        let multibyte = "é".repeat(MAX_NOTE_BYTES);
497        let truncated = truncate_note(&multibyte);
498        assert!(truncated.len() <= MAX_NOTE_BYTES);
499        assert!(truncated.ends_with("..."));
500    }
501
502    #[test]
503    fn creates_event_with_reporter_and_reference() {
504        block_on(async {
505            let server = MockApiServer::new();
506            let recorder = recorder(&server);
507            recorder
508                .publish(&config_map("uid-1"), &event("it broke"))
509                .await
510                .unwrap();
511
512            let requests = server.requests();
513            assert_eq!(requests.len(), 1);
514            let req = &requests[0];
515            assert_eq!(req.method, "POST");
516            assert_eq!(req.path, "/apis/events.k8s.io/v1/namespaces/ns/events");
517            assert_eq!(req.body["type"], "Warning");
518            assert_eq!(req.body["reason"], "Broken");
519            assert_eq!(req.body["action"], "Reconcile");
520            assert_eq!(req.body["note"], "it broke");
521            assert_eq!(req.body["reportingController"], "test.example.com");
522            assert_eq!(req.body["reportingInstance"], "pod-0");
523            assert_eq!(req.body["regarding"]["kind"], "ConfigMap");
524            assert_eq!(req.body["regarding"]["uid"], "uid-1");
525            let name = req.body["metadata"]["name"].as_str().unwrap();
526            assert!(name.starts_with("cm."), "{name}");
527        });
528    }
529
530    #[test]
531    fn test_event_name() {
532        let time = Timestamp::from_nanosecond(0x1234_5678_9abc_def0).unwrap();
533        let suffix = ".123456789abcdef0";
534        let fallback = format!("event{suffix}");
535        assert_eq!(event_name(Some("cm"), time), format!("cm{suffix}"));
536        assert_eq!(
537            event_name(Some("12345678-1234-1234-1234-123456789012"), time),
538            format!("12345678-1234-1234-1234-123456789012{suffix}")
539        );
540        assert_eq!(event_name(None, time), fallback);
541
542        // names that only satisfy path-segment rules, as RBAC names may
543        assert_eq!(event_name(Some("system:controller:foo"), time), fallback);
544        assert_eq!(event_name(Some("Foo"), time), fallback);
545
546        let longest_label = "a".repeat(63);
547        assert_eq!(
548            event_name(Some(&longest_label), time),
549            format!("{longest_label}{suffix}")
550        );
551        assert_eq!(event_name(Some(&"a".repeat(64)), time), fallback);
552
553        let longest = vec!["a".repeat(63); 4].join(".")[..253 - suffix.len()].to_owned();
554        assert_eq!(
555            event_name(Some(&longest), time),
556            format!("{longest}{suffix}")
557        );
558        assert_eq!(event_name(Some(&format!("{longest}a")), time), fallback);
559    }
560
561    #[test]
562    fn test_is_dns_subdomain() {
563        assert!(is_dns_subdomain("a"));
564        assert!(is_dns_subdomain("example.com"));
565        assert!(is_dns_subdomain("a-1.b-2"));
566        assert!(is_dns_subdomain(&"a".repeat(63)));
567        assert!(!is_dns_subdomain(""));
568        assert!(!is_dns_subdomain("a."));
569        assert!(!is_dns_subdomain(".a"));
570        assert!(!is_dns_subdomain("a..b"));
571        assert!(!is_dns_subdomain("-a"));
572        assert!(!is_dns_subdomain("a-"));
573        assert!(!is_dns_subdomain("A"));
574        assert!(!is_dns_subdomain("a:b"));
575        assert!(!is_dns_subdomain(&"a".repeat(64)));
576        assert!(!is_dns_subdomain(&vec!["a".repeat(63); 5].join(".")[..254]));
577    }
578
579    #[test]
580    fn aggregates_identical_events() {
581        block_on(async {
582            let server = MockApiServer::new();
583            let recorder = recorder(&server);
584            let cm = config_map("uid-1");
585            for _ in 0..3 {
586                recorder.publish(&cm, &event("it broke")).await.unwrap();
587            }
588
589            let requests = server.requests();
590            assert_eq!(requests.len(), 3);
591            assert_eq!(requests[0].method, "POST");
592            let name = requests[0].created_name();
593            for (req, count) in requests[1..].iter().zip([2, 3]) {
594                assert_eq!(req.method, "PATCH");
595                assert_eq!(
596                    req.path,
597                    format!("/apis/events.k8s.io/v1/namespaces/ns/events/{name}")
598                );
599                assert_eq!(req.body["series"]["count"], json!(count));
600                assert!(req.body["series"]["lastObservedTime"].is_string());
601            }
602        });
603    }
604
605    #[test]
606    fn concurrent_identical_events_aggregate() {
607        block_on(async {
608            let server = MockApiServer::new();
609            let recorder = recorder(&server);
610            let cm = config_map("uid-1");
611            let ev = event("it broke");
612            let results =
613                futures::future::join_all((0..3).map(|_| recorder.publish(&cm, &ev))).await;
614            assert!(results.iter().all(Result::is_ok));
615
616            let requests = server.requests();
617            let methods: Vec<_> = requests.iter().map(|r| r.method.as_str()).collect();
618            assert_eq!(methods, ["POST", "PATCH", "PATCH"]);
619            assert_eq!(requests[1].body["series"]["count"], json!(2));
620            assert_eq!(requests[2].body["series"]["count"], json!(3));
621        });
622    }
623
624    #[test]
625    fn changed_note_starts_new_event() {
626        block_on(async {
627            let server = MockApiServer::new();
628            let recorder = recorder(&server);
629            let cm = config_map("uid-1");
630            recorder.publish(&cm, &event("first cause")).await.unwrap();
631            recorder.publish(&cm, &event("second cause")).await.unwrap();
632            recorder.publish(&cm, &event("second cause")).await.unwrap();
633
634            let methods: Vec<_> = server.requests().into_iter().map(|r| r.method).collect();
635            assert_eq!(methods, ["POST", "POST", "PATCH"]);
636        });
637    }
638
639    #[test]
640    fn forget_starts_new_event() {
641        block_on(async {
642            let server = MockApiServer::new();
643            let recorder = recorder(&server);
644            let cm = config_map("uid-1");
645            recorder.publish(&cm, &event("it broke")).await.unwrap();
646            recorder.forget(&cm);
647            recorder.publish(&cm, &event("it broke")).await.unwrap();
648
649            let methods: Vec<_> = server.requests().into_iter().map(|r| r.method).collect();
650            assert_eq!(methods, ["POST", "POST"]);
651        });
652    }
653
654    #[test]
655    fn forget_failures_keeps_other_events() {
656        block_on(async {
657            let server = MockApiServer::new();
658            let recorder = recorder(&server);
659            let cm = config_map("uid-1");
660            let reference = cm.object_ref(&());
661            recorder
662                .publish_to(&reference, &event("it broke"), true)
663                .await
664                .unwrap();
665            recorder.publish(&cm, &event("it broke")).await.unwrap();
666            recorder.forget_failures("uid-1");
667            recorder
668                .publish_to(&reference, &event("it broke"), true)
669                .await
670                .unwrap();
671            recorder.publish(&cm, &event("it broke")).await.unwrap();
672
673            let methods: Vec<_> = server.requests().into_iter().map(|r| r.method).collect();
674            assert_eq!(methods, ["POST", "POST", "POST", "PATCH"]);
675        });
676    }
677
678    #[test]
679    fn times_out_and_recovers() {
680        block_on(async {
681            let server = MockApiServer::new();
682            let recorder = recorder(&server).with_timeout(Duration::from_millis(50));
683            let cm = config_map("uid-1");
684            server.hang_next_request();
685            let err = recorder.publish(&cm, &event("it broke")).await.unwrap_err();
686            assert!(matches!(err, PublishError::Timeout(_)), "{err:?}");
687
688            recorder.publish(&cm, &event("it broke")).await.unwrap();
689            let methods: Vec<_> = server.requests().into_iter().map(|r| r.method).collect();
690            assert_eq!(methods, ["POST"]);
691        });
692    }
693
694    #[test]
695    fn expired_event_is_recreated() {
696        block_on(async {
697            let server = MockApiServer::new();
698            let recorder = recorder(&server);
699            let cm = config_map("uid-1");
700            recorder.publish(&cm, &event("it broke")).await.unwrap();
701            server.fail_next_patch(404);
702            recorder.publish(&cm, &event("it broke")).await.unwrap();
703            recorder.publish(&cm, &event("it broke")).await.unwrap();
704
705            let requests = server.requests();
706            let methods: Vec<_> = requests.iter().map(|r| r.method.clone()).collect();
707            assert_eq!(methods, ["POST", "PATCH", "POST", "PATCH"]);
708            assert_eq!(requests[3].body["series"]["count"], json!(2));
709        });
710    }
711
712    #[test]
713    fn cluster_scoped_events_go_to_default_namespace() {
714        block_on(async {
715            let server = MockApiServer::new();
716            let recorder = recorder(&server);
717            let ns = k8s_openapi::api::core::v1::Namespace {
718                metadata: ObjectMeta {
719                    name: Some("some-namespace".to_owned()),
720                    uid: Some("uid-1".to_owned()),
721                    ..Default::default()
722                },
723                ..Default::default()
724            };
725            recorder.publish(&ns, &event("it broke")).await.unwrap();
726
727            let requests = server.requests();
728            assert_eq!(
729                requests[0].path,
730                "/apis/events.k8s.io/v1/namespaces/default/events"
731            );
732        });
733    }
734}
k8s_controller/events.rs

k8s_controller/
events.rs
