Skip to content

Commit 76e7681

Browse files
committed
Houston, we have CronJobs.
- Add CronJob and Job reflectors - Thread Job into the owner graph - Add CronJob piggyback host - Add first section, kube_cron_job_info_v1 - Chart updated for new defaults filter - Basic tests
1 parent 6d2f94c commit 76e7681

11 files changed

Lines changed: 288 additions & 3 deletions

File tree

charts/cmk-rustik/values.yaml

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -130,3 +130,4 @@ emitAll:
130130
deployments: true
131131
daemonsets: true
132132
statefulsets: true
133+
cronjobs: true

metrics-cache/src/cli_args.rs

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -152,6 +152,10 @@ pub struct CliArgs {
152152
/// Emit all StatefulSet resources rather than only annotated ones
153153
#[arg(long = "all-statefulsets")]
154154
pub all_statefulsets: bool,
155+
156+
/// Emit all CronJob resources rather than only annotated ones
157+
#[arg(long = "all-cronjobs")]
158+
pub all_cronjobs: bool,
155159
}
156160

157161
/// Convert a numeric argument given by the user as seconds into a Duration.

metrics-cache/src/host_settings.rs

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -47,6 +47,7 @@ pub struct AlwaysEmitted {
4747
pub deployments: bool,
4848
pub daemonsets: bool,
4949
pub statefulsets: bool,
50+
pub cronjobs: bool,
5051
}
5152

5253
impl AlwaysEmitted {
@@ -58,6 +59,7 @@ impl AlwaysEmitted {
5859
deployments: args.all_deployments,
5960
daemonsets: args.all_daemonsets,
6061
statefulsets: args.all_statefulsets,
62+
cronjobs: args.all_cronjobs,
6163
}
6264
}
6365
}

metrics-cache/src/ingest/reflectors.rs

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
use futures_util::StreamExt;
22
use k8s_openapi::api::apps::v1::{DaemonSet, Deployment, ReplicaSet, StatefulSet};
3+
use k8s_openapi::api::batch::v1::{CronJob, Job};
34
use k8s_openapi::api::core::v1::{Namespace, Node, PersistentVolume, PersistentVolumeClaim, Pod};
45
use kube::runtime::reflector::store::WriterDropped;
56
use kube::runtime::watcher::Config as WatchConfig;
@@ -148,6 +149,8 @@ define_reflectors! {
148149
persistent_volumes: PersistentVolume => "PersistentVolume",
149150
persistent_volume_claims: PersistentVolumeClaim => "PersistentVolumeClaim",
150151
statefulsets: StatefulSet => "StatefulSet",
152+
cronjobs: CronJob => "CronJob",
153+
jobs: Job => "Job",
151154
}
152155

153156
/// The inner-state of a reflector. This gets updated by the reflector's
Lines changed: 73 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,73 @@
1+
use k8s_openapi::api::batch::v1;
2+
use k8s_openapi::api::core::v1::Pod;
3+
use k8s_openapi::apimachinery::pkg::apis::meta::v1::ObjectMeta;
4+
use std::sync::Arc;
5+
6+
use crate::host_settings::HostSettings;
7+
use crate::piggyback::{AggregationHost, Meta, PiggybackHost};
8+
use crate::section::{
9+
cronjob::KubeCronJobInfoV1,
10+
writeable::{SectionError, WriteableSection},
11+
};
12+
use crate::snapshot::Snapshot;
13+
14+
pub struct CronJob<'a> {
15+
api: &'a v1::CronJob,
16+
meta: Meta<'a>,
17+
snapshot: &'a Snapshot,
18+
settings: &'a HostSettings,
19+
uid: &'a str,
20+
}
21+
22+
impl CronJob<'_> {
23+
pub fn new<'a>(
24+
api: &'a v1::CronJob,
25+
snapshot: &'a Snapshot,
26+
settings: &'a HostSettings,
27+
) -> Option<CronJob<'a>> {
28+
let meta = Meta::from_resource(api)?;
29+
let uid = api.metadata.uid.as_deref()?;
30+
Some(CronJob {
31+
api,
32+
meta,
33+
snapshot,
34+
settings,
35+
uid,
36+
})
37+
}
38+
}
39+
40+
impl AggregationHost for CronJob<'_> {
41+
fn snapshot(&self) -> &Snapshot {
42+
self.snapshot
43+
}
44+
45+
fn pods(&self) -> impl Iterator<Item = &Arc<Pod>> {
46+
self.snapshot
47+
.owner_graph
48+
.pods_by_controller(self.uid)
49+
.iter()
50+
}
51+
}
52+
53+
impl PiggybackHost for CronJob<'_> {
54+
fn metadata(&self) -> Option<&ObjectMeta> {
55+
Some(&self.api.metadata)
56+
}
57+
58+
fn kind(&self) -> &str {
59+
&self.meta.kind
60+
}
61+
62+
fn emit(&self) -> Vec<Result<WriteableSection, SectionError>> {
63+
let me = self.meta.piggyback_hostname(&self.settings.cluster_name);
64+
let mut out = Vec::new();
65+
if let Some(kube_cron_job_info_v1) =
66+
KubeCronJobInfoV1::from_cron_job(self.api, self.settings)
67+
{
68+
out.push(WriteableSection::of(&me, &kube_cron_job_info_v1));
69+
}
70+
out.extend(self.aggregation_sections(&me));
71+
out
72+
}
73+
}

metrics-cache/src/piggyback/mod.rs

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
pub mod aggregation_host;
22
pub mod cluster;
3+
pub mod cronjob;
34
pub mod daemonset;
45
pub mod deployment;
56
pub mod namespace;
@@ -14,6 +15,7 @@ use tracing::warn;
1415
use crate::host_settings::HostSettings;
1516
pub(crate) use crate::piggyback::aggregation_host::AggregationHost;
1617
use crate::piggyback::cluster::Cluster;
18+
use crate::piggyback::cronjob::CronJob;
1719
use crate::piggyback::daemonset::DaemonSet;
1820
use crate::piggyback::deployment::Deployment;
1921
use crate::piggyback::namespace::Namespace;
@@ -187,6 +189,9 @@ pub fn emit_all(snap: &Snapshot, settings: &HostSettings) -> Vec<WriteableSectio
187189
always.statefulsets,
188190
|n| StatefulSet::new(n, snap, settings),
189191
));
192+
out.extend(collect(snap.stores.cronjobs.iter(), always.cronjobs, |n| {
193+
CronJob::new(n, snap, settings)
194+
}));
190195

191196
// Cluster is a special snowflake, there aren't any reflectors to iterate
192197
out.extend(collect(std::iter::once(()), true, |()| {
Lines changed: 119 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,119 @@
1+
use k8s_openapi::api::batch::v1::CronJob;
2+
use serde::Serialize;
3+
use std::collections::BTreeMap;
4+
5+
use crate::host_settings::HostSettings;
6+
use crate::section::Section;
7+
use crate::section::common::LabelRef;
8+
9+
/// CronJob info. (`kube_cron_job_info_v1`)
10+
#[derive(Serialize)]
11+
pub(crate) struct KubeCronJobInfoV1<'a> {
12+
pub name: &'a str,
13+
pub namespace: &'a str,
14+
pub creation_timestamp: Option<f64>,
15+
pub labels: BTreeMap<&'a str, LabelRef<'a>>,
16+
/// Annotations filtered with user input.
17+
///
18+
/// After receiving the annotations from the Kubernetes API, we cannot
19+
/// process all of them as HostLabels. FilteredAnnotations are those
20+
/// annotations, which can be processed. This means that the annotations can
21+
/// no longer be arbitrary json objects and that options from the
22+
/// `Kubernetes` rule have been taken into account.
23+
pub annotations: BTreeMap<&'a str, &'a str>,
24+
pub schedule: &'a str,
25+
pub concurrency_policy: &'a str,
26+
pub failed_jobs_history_limit: i32,
27+
pub successful_jobs_history_limit: i32,
28+
pub suspend: bool,
29+
pub cluster: &'a str,
30+
pub kubernetes_cluster_hostname: &'a str,
31+
}
32+
33+
impl<'a> KubeCronJobInfoV1<'a> {
34+
pub fn from_cron_job(
35+
cron_job: &'a CronJob,
36+
settings: &'a HostSettings,
37+
) -> Option<KubeCronJobInfoV1<'a>> {
38+
let section = KubeCronJobInfoV1 {
39+
name: cron_job.metadata.name.as_deref()?,
40+
namespace: cron_job.metadata.namespace.as_deref()?,
41+
creation_timestamp: cron_job
42+
.metadata
43+
.creation_timestamp
44+
.as_ref()
45+
.map(|t| t.0.as_millisecond() as f64 / 1000.0),
46+
labels: cron_job
47+
.metadata
48+
.labels
49+
.as_ref()
50+
.map(LabelRef::from_map)
51+
.unwrap_or_default(),
52+
annotations: cron_job
53+
.metadata
54+
.annotations
55+
.as_ref()
56+
.map(|m| settings.annotation_key_pattern.filter(m))
57+
.unwrap_or_default(),
58+
schedule: &cron_job.spec.schedule,
59+
concurrency_policy: cron_job.spec.concurrency_policy.as_deref()?,
60+
failed_jobs_history_limit: cron_job.spec.failed_jobs_history_limit?,
61+
successful_jobs_history_limit: cron_job.spec.successful_jobs_history_limit?,
62+
suspend: cron_job.spec.suspend?,
63+
cluster: &settings.cluster_name,
64+
kubernetes_cluster_hostname: &settings.cluster_host_name,
65+
};
66+
Some(section)
67+
}
68+
}
69+
70+
impl Section for KubeCronJobInfoV1<'_> {
71+
const NAME: &'static str = "kube_cron_job_info_v1";
72+
}
73+
74+
#[cfg(test)]
75+
mod tests {
76+
use super::*;
77+
use regex::Regex;
78+
79+
use crate::host_settings::AnnotationKeyPattern;
80+
use crate::test_support::{cron_job, host_settings};
81+
82+
#[test]
83+
fn kube_cron_job_info_v1() {
84+
let cron_job = cron_job("important-job");
85+
let mut settings = host_settings();
86+
insta::assert_json_snapshot!(KubeCronJobInfoV1::from_cron_job(&cron_job, &settings));
87+
88+
// This pattern should only match one annotation
89+
settings.annotation_key_pattern =
90+
AnnotationKeyPattern::Pattern(Regex::new("^example").unwrap());
91+
assert_eq!(
92+
KubeCronJobInfoV1::from_cron_job(&cron_job, &settings)
93+
.unwrap()
94+
.annotations
95+
.len(),
96+
1
97+
);
98+
99+
// Ignore all annotations, emit 0 of them
100+
settings.annotation_key_pattern = AnnotationKeyPattern::IgnoreAll;
101+
assert_eq!(
102+
KubeCronJobInfoV1::from_cron_job(&cron_job, &settings)
103+
.unwrap()
104+
.annotations
105+
.len(),
106+
0
107+
);
108+
109+
// Import all annotations (fixture default, captured by insta above, but let's be explicit)
110+
settings.annotation_key_pattern = AnnotationKeyPattern::ImportAll;
111+
assert_eq!(
112+
KubeCronJobInfoV1::from_cron_job(&cron_job, &settings)
113+
.unwrap()
114+
.annotations
115+
.len(),
116+
2
117+
);
118+
}
119+
}

metrics-cache/src/section/mod.rs

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
pub mod common;
2+
pub mod cronjob;
23
pub mod namespace;
34
pub mod performance;
45
pub mod pod;
Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,21 @@
1+
---
2+
source: metrics-cache/src/section/cronjob.rs
3+
expression: "KubeCronJobInfoV1::from_cron_job(&cron_job, &settings)"
4+
---
5+
{
6+
"name": "important-job",
7+
"namespace": "the-actual-coolest-namespace-of-all-time",
8+
"creation_timestamp": 1718824965.0,
9+
"labels": {},
10+
"annotations": {
11+
"checkmk.com/promote-to-host": "true",
12+
"example.com/cool-animal": "monkeys"
13+
},
14+
"schedule": "30 0,8,16 * * *",
15+
"concurrency_policy": "Allow",
16+
"failed_jobs_history_limit": 10,
17+
"successful_jobs_history_limit": 5,
18+
"suspend": false,
19+
"cluster": "the-cluster",
20+
"kubernetes_cluster_hostname": "cluster.host.tld"
21+
}

metrics-cache/src/snapshot/owner_graph.rs

Lines changed: 19 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
use k8s_openapi::api::apps::v1::ReplicaSet;
2+
use k8s_openapi::api::batch::v1::Job;
23
use k8s_openapi::api::core::v1::Pod;
34
use k8s_openapi::apimachinery::pkg::apis::meta::v1::OwnerReference;
45
use kube::ResourceExt;
@@ -40,7 +41,7 @@ impl OwnerGraph {
4041
/// [`crate::snapshot::Snapshot::new()`].
4142
pub fn from_frozen_stores(stores: &FrozenStores) -> Self {
4243
let owner_ref_by_uid =
43-
Self::map_object_uids_to_owner_ref(&stores.pods, &stores.replicasets);
44+
Self::map_object_uids_to_owner_ref(&stores.pods, &stores.replicasets, &stores.jobs);
4445
let pods_by_controller = Self::get_pods_by_controller(&stores.pods, &owner_ref_by_uid);
4546
OwnerGraph {
4647
owner_ref_by_uid,
@@ -58,6 +59,7 @@ impl OwnerGraph {
5859
fn map_object_uids_to_owner_ref(
5960
pods: &[Arc<Pod>],
6061
replicasets: &[Arc<ReplicaSet>],
62+
jobs: &[Arc<Job>],
6163
) -> HashMap<Uid, OwnerReference> {
6264
let mut map = HashMap::new();
6365
for pod in pods {
@@ -80,6 +82,16 @@ impl OwnerGraph {
8082
map.insert(Uid(uid.into()), owner_controller.to_owned());
8183
}
8284
}
85+
for job in jobs {
86+
if let Some(owner_controller) = job
87+
.owner_references()
88+
.iter()
89+
.find(|r| r.controller == Some(true))
90+
&& let Some(uid) = job.metadata.uid.clone()
91+
{
92+
map.insert(Uid(uid.into()), owner_controller.to_owned());
93+
}
94+
}
8395
map
8496
}
8597

@@ -201,6 +213,7 @@ mod tests {
201213
fn map_object_uids_to_owner_ref() {
202214
let pod1 = pod_owned_by("pod1", "pod1-uid", owner_ref("ReplicaSet", "rs", "rs-uid"));
203215
let rs = replicaset_owned_by("rs", "rs-uid", owner_ref("Deployment", "dep", "dep-uid"));
216+
let job = job_owned_by("job1", "job1-uid", owner_ref("CronJob", "cj", "cj-uid"));
204217

205218
// Non-controller owner references are skipped
206219
let mut non_controller = owner_ref("ReplicaSet", "rs", "rs-uid");
@@ -210,11 +223,13 @@ mod tests {
210223
let map = OwnerGraph::map_object_uids_to_owner_ref(
211224
&[pod1.into(), pod_owned_by_non_controller.into()],
212225
&[rs.into()],
226+
&[job.into()],
213227
);
214228

215-
assert_eq!(map.len(), 2); // the pod with no controller is dropped
229+
assert_eq!(map.len(), 3); // the pod with no controller is dropped
216230
assert_eq!(map["pod1-uid"].uid, "rs-uid");
217231
assert_eq!(map["rs-uid"].uid, "dep-uid");
232+
assert_eq!(map["job1-uid"].uid, "cj-uid");
218233
}
219234

220235
#[test]
@@ -228,7 +243,8 @@ mod tests {
228243
replicaset_owned_by("rs", "rs-uid", owner_ref("Deployment", "dep", "dep-uid"))
229244
.into(),
230245
];
231-
let owner_ref_by_uid = OwnerGraph::map_object_uids_to_owner_ref(&pods, &replicasets);
246+
let jobs = [job_owned_by("job1", "job1-uid", owner_ref("CronJob", "cj", "cj-uid")).into()];
247+
let owner_ref_by_uid = OwnerGraph::map_object_uids_to_owner_ref(&pods, &replicasets, &jobs);
232248
let pods_owned_by_rs = OwnerGraph::get_pods_by_controller(&pods, &owner_ref_by_uid);
233249

234250
assert_eq!(pods_owned_by_rs.len(), 2); // two controllers: rs-uid and dep-uid

0 commit comments

Comments
 (0)