kube_core/
subresource.rs

1//! Request builder types and parameters for subresources
2use crate::{
3    params::{DeleteParams, PostParams},
4    request::{Error, JSON_MIME, Request},
5};
6use jiff::{Timestamp, Unit};
7use std::fmt::Debug;
8
9pub use k8s_openapi::api::autoscaling::v1::{Scale, ScaleSpec, ScaleStatus};
10
11// ----------------------------------------------------------------------------
12// Log subresource
13// ----------------------------------------------------------------------------
14
15/// Params for logging
16#[derive(Default, Clone, Debug)]
17pub struct LogParams {
18    /// The container for which to stream logs. Defaults to only container if there is one container in the pod.
19    pub container: Option<String>,
20    /// Follow the log stream of the pod. Defaults to `false`.
21    pub follow: bool,
22    /// If set, the number of bytes to read from the server before terminating the log output.
23    /// This may not display a complete final line of logging, and may return slightly more or slightly less than the specified limit.
24    pub limit_bytes: Option<i64>,
25    /// If `true`, then the output is pretty printed.
26    pub pretty: bool,
27    /// Return previous terminated container logs. Defaults to `false`.
28    pub previous: bool,
29    /// A relative time in seconds before the current time from which to show logs.
30    /// If this value precedes the time a pod was started, only logs since the pod start will be returned.
31    /// If this value is in the future, no logs will be returned. Only one of sinceSeconds or sinceTime may be specified.
32    pub since_seconds: Option<i64>,
33    /// An RFC3339 timestamp from which to show logs. If this value
34    /// precedes the time a pod was started, only logs since the pod start will be returned.
35    /// If this value is in the future, no logs will be returned.
36    /// Only one of sinceSeconds or sinceTime may be specified.
37    pub since_time: Option<Timestamp>,
38    /// If set, the number of lines from the end of the logs to show.
39    /// If not specified, logs are shown from the creation of the container or sinceSeconds or sinceTime
40    pub tail_lines: Option<i64>,
41    /// If `true`, add an RFC3339 or RFC3339Nano timestamp at the beginning of every line of log output. Defaults to `false`.
42    pub timestamps: bool,
43}
44
45impl Request {
46    /// Get a pod logs
47    pub fn logs(&self, name: &str, lp: &LogParams) -> Result<http::Request<Vec<u8>>, Error> {
48        let target = format!("{}/{}/log?", self.url_path, name);
49        let mut qp = form_urlencoded::Serializer::new(target);
50
51        if let Some(container) = &lp.container {
52            qp.append_pair("container", container);
53        }
54
55        if lp.follow {
56            qp.append_pair("follow", "true");
57        }
58
59        if let Some(lb) = &lp.limit_bytes {
60            qp.append_pair("limitBytes", &lb.to_string());
61        }
62
63        if lp.pretty {
64            qp.append_pair("pretty", "true");
65        }
66
67        if lp.previous {
68            qp.append_pair("previous", "true");
69        }
70
71        if let Some(ss) = &lp.since_seconds {
72            qp.append_pair("sinceSeconds", &ss.to_string());
73        } else if let Some(st) = &lp.since_time {
74            // Unwrapping here is ok as `round` only errors if a.) the smallest unit is larger than hour (we set second as smallest unit)
75            // or b.) a rounding increment is configured (which we don't set) that does not fit into 86400 seconds (a day)
76            // without any remainder.
77            // `jiff::Timestamp` provides RFC3339 via `Display`, docs: https://docs.rs/jiff/latest/jiff/struct.Timestamp.html#impl-Display-for-Timestamp
78            qp.append_pair("sinceTime", &st.round(Unit::Second).unwrap().to_string());
79        }
80
81        if let Some(tl) = &lp.tail_lines {
82            qp.append_pair("tailLines", &tl.to_string());
83        }
84
85        if lp.timestamps {
86            qp.append_pair("timestamps", "true");
87        }
88
89        let urlstr = qp.finish();
90        let req = http::Request::get(urlstr);
91        req.body(vec![]).map_err(Error::BuildRequest)
92    }
93}
94
95// ----------------------------------------------------------------------------
96// Eviction subresource
97// ----------------------------------------------------------------------------
98
99/// Params for evictable objects
100#[derive(Default, Clone)]
101pub struct EvictParams {
102    /// How the eviction should occur
103    pub delete_options: Option<DeleteParams>,
104    /// How the http post should occur
105    pub post_options: PostParams,
106}
107
108impl Request {
109    /// Create an eviction
110    pub fn evict(&self, name: &str, ep: &EvictParams) -> Result<http::Request<Vec<u8>>, Error> {
111        let target = format!("{}/{}/eviction?", self.url_path, name);
112        // This is technically identical to Request::create, but different url
113        let pp = &ep.post_options;
114        pp.validate()?;
115        let mut qp = form_urlencoded::Serializer::new(target);
116        pp.populate_qp(&mut qp);
117        let urlstr = qp.finish();
118        // eviction body parameters are awkward, need metadata with name
119        let data = serde_json::to_vec(&serde_json::json!({
120            "delete_options": ep.delete_options,
121            "metadata": { "name": name }
122        }))
123        .map_err(Error::SerializeBody)?;
124        let req = http::Request::post(urlstr).header(http::header::CONTENT_TYPE, JSON_MIME);
125        req.body(data).map_err(Error::BuildRequest)
126    }
127}
128
129// ----------------------------------------------------------------------------
130// Attach subresource
131// ----------------------------------------------------------------------------
132/// Parameters for attaching to a container in a Pod.
133///
134/// - One of `stdin`, `stdout`, or `stderr` must be `true`.
135/// - `stderr` and `tty` cannot both be `true` because multiplexing is not supported with TTY.
136#[cfg(feature = "ws")]
137#[cfg_attr(docsrs, doc(cfg(feature = "ws")))]
138#[derive(Debug)]
139pub struct AttachParams {
140    /// The name of the container to attach.
141    /// Defaults to the only container if there is only one container in the pod.
142    pub container: Option<String>,
143    /// Attach to the container's standard input. Defaults to `false`.
144    ///
145    /// Call [`AttachedProcess::stdin`](https://docs.rs/kube/*/kube/api/struct.AttachedProcess.html#method.stdin) to obtain a writer.
146    pub stdin: bool,
147    /// Attach to the container's standard output. Defaults to `true`.
148    ///
149    /// Call [`AttachedProcess::stdout`](https://docs.rs/kube/*/kube/api/struct.AttachedProcess.html#method.stdout) to obtain a reader.
150    pub stdout: bool,
151    /// Attach to the container's standard error. Defaults to `true`.
152    ///
153    /// Call [`AttachedProcess::stderr`](https://docs.rs/kube/*/kube/api/struct.AttachedProcess.html#method.stderr) to obtain a reader.
154    pub stderr: bool,
155    /// Allocate TTY. Defaults to `false`.
156    ///
157    /// NOTE: Terminal resizing is not implemented yet.
158    pub tty: bool,
159
160    /// The maximum amount of bytes that can be written to the internal `stdin`
161    /// pipe before the write returns `Poll::Pending`.
162    /// Defaults to 1024.
163    ///
164    /// This is not sent to the server.
165    pub max_stdin_buf_size: Option<usize>,
166    /// The maximum amount of bytes that can be written to the internal `stdout`
167    /// pipe before the write returns `Poll::Pending`.
168    /// Defaults to 1024.
169    ///
170    /// This is not sent to the server.
171    pub max_stdout_buf_size: Option<usize>,
172    /// The maximum amount of bytes that can be written to the internal `stderr`
173    /// pipe before the write returns `Poll::Pending`.
174    /// Defaults to 1024.
175    ///
176    /// This is not sent to the server.
177    pub max_stderr_buf_size: Option<usize>,
178}
179
180#[cfg(feature = "ws")]
181#[cfg_attr(docsrs, doc(cfg(feature = "ws")))]
182impl Default for AttachParams {
183    // Default matching the server's defaults.
184    fn default() -> Self {
185        Self {
186            container: None,
187            stdin: false,
188            stdout: true,
189            stderr: true,
190            tty: false,
191            max_stdin_buf_size: None,
192            max_stdout_buf_size: None,
193            max_stderr_buf_size: None,
194        }
195    }
196}
197
198#[cfg(feature = "ws")]
199#[cfg_attr(docsrs, doc(cfg(feature = "ws")))]
200impl AttachParams {
201    /// Default parameters for an tty exec with stdin and stdout
202    #[must_use]
203    pub fn interactive_tty() -> Self {
204        Self {
205            stdin: true,
206            stdout: true,
207            stderr: false,
208            tty: true,
209            ..Default::default()
210        }
211    }
212
213    /// Specify the container to execute in.
214    #[must_use]
215    pub fn container<T: Into<String>>(mut self, container: T) -> Self {
216        self.container = Some(container.into());
217        self
218    }
219
220    /// Set `stdin` field.
221    #[must_use]
222    pub fn stdin(mut self, enable: bool) -> Self {
223        self.stdin = enable;
224        self
225    }
226
227    /// Set `stdout` field.
228    #[must_use]
229    pub fn stdout(mut self, enable: bool) -> Self {
230        self.stdout = enable;
231        self
232    }
233
234    /// Set `stderr` field.
235    #[must_use]
236    pub fn stderr(mut self, enable: bool) -> Self {
237        self.stderr = enable;
238        self
239    }
240
241    /// Set `tty` field.
242    #[must_use]
243    pub fn tty(mut self, enable: bool) -> Self {
244        self.tty = enable;
245        self
246    }
247
248    /// Set `max_stdin_buf_size` field.
249    #[must_use]
250    pub fn max_stdin_buf_size(mut self, size: usize) -> Self {
251        self.max_stdin_buf_size = Some(size);
252        self
253    }
254
255    /// Set `max_stdout_buf_size` field.
256    #[must_use]
257    pub fn max_stdout_buf_size(mut self, size: usize) -> Self {
258        self.max_stdout_buf_size = Some(size);
259        self
260    }
261
262    /// Set `max_stderr_buf_size` field.
263    #[must_use]
264    pub fn max_stderr_buf_size(mut self, size: usize) -> Self {
265        self.max_stderr_buf_size = Some(size);
266        self
267    }
268
269    pub(crate) fn validate(&self) -> Result<(), Error> {
270        if !self.stdin && !self.stdout && !self.stderr {
271            return Err(Error::Validation(
272                "AttachParams: one of stdin, stdout, or stderr must be true".into(),
273            ));
274        }
275
276        if self.stderr && self.tty {
277            // Multiplexing is not supported with TTY
278            return Err(Error::Validation(
279                "AttachParams: tty and stderr cannot both be true".into(),
280            ));
281        }
282
283        Ok(())
284    }
285
286    fn append_to_url_serializer(&self, qp: &mut form_urlencoded::Serializer<String>) {
287        if self.stdin {
288            qp.append_pair("stdin", "true");
289        }
290        if self.stdout {
291            qp.append_pair("stdout", "true");
292        }
293        if self.stderr {
294            qp.append_pair("stderr", "true");
295        }
296        if self.tty {
297            qp.append_pair("tty", "true");
298        }
299        if let Some(container) = &self.container {
300            qp.append_pair("container", container);
301        }
302    }
303
304    #[cfg(feature = "kubelet-debug")]
305    // https://github.com/kubernetes/kubernetes/blob/466d9378dbb0a185df9680657f5cd96d5e5aab57/pkg/apis/core/types.go#L6005-L6013
306    pub(crate) fn append_to_url_serializer_local(&self, qp: &mut form_urlencoded::Serializer<String>) {
307        if self.stdin {
308            qp.append_pair("input", "1");
309        }
310        if self.stdout {
311            qp.append_pair("output", "1");
312        }
313        if self.stderr {
314            qp.append_pair("error", "1");
315        }
316        if self.tty {
317            qp.append_pair("tty", "1");
318        }
319    }
320}
321
322#[cfg(feature = "ws")]
323#[cfg_attr(docsrs, doc(cfg(feature = "ws")))]
324impl Request {
325    /// Attach to a pod
326    pub fn attach(&self, name: &str, ap: &AttachParams) -> Result<http::Request<Vec<u8>>, Error> {
327        ap.validate()?;
328
329        let target = format!("{}/{}/attach?", self.url_path, name);
330        let mut qp = form_urlencoded::Serializer::new(target);
331        ap.append_to_url_serializer(&mut qp);
332
333        let req = http::Request::get(qp.finish());
334        req.body(vec![]).map_err(Error::BuildRequest)
335    }
336}
337
338// ----------------------------------------------------------------------------
339// Exec subresource
340// ----------------------------------------------------------------------------
341#[cfg(feature = "ws")]
342#[cfg_attr(docsrs, doc(cfg(feature = "ws")))]
343impl Request {
344    /// Execute command in a pod
345    pub fn exec<I, T>(
346        &self,
347        name: &str,
348        command: I,
349        ap: &AttachParams,
350    ) -> Result<http::Request<Vec<u8>>, Error>
351    where
352        I: IntoIterator<Item = T>,
353        T: Into<String>,
354    {
355        ap.validate()?;
356
357        let target = format!("{}/{}/exec?", self.url_path, name);
358        let mut qp = form_urlencoded::Serializer::new(target);
359        ap.append_to_url_serializer(&mut qp);
360
361        for c in command.into_iter() {
362            qp.append_pair("command", &c.into());
363        }
364
365        let req = http::Request::get(qp.finish());
366        req.body(vec![]).map_err(Error::BuildRequest)
367    }
368}
369
370// ----------------------------------------------------------------------------
371// Portforward subresource
372// ----------------------------------------------------------------------------
373#[cfg(feature = "ws")]
374#[cfg_attr(docsrs, doc(cfg(feature = "ws")))]
375impl Request {
376    /// Request to forward ports of a pod
377    pub fn portforward(&self, name: &str, ports: &[u16]) -> Result<http::Request<Vec<u8>>, Error> {
378        if ports.is_empty() {
379            return Err(Error::Validation("ports cannot be empty".into()));
380        }
381        if ports.len() > 128 {
382            return Err(Error::Validation(
383                "the number of ports cannot be more than 128".into(),
384            ));
385        }
386
387        if ports.len() > 1 {
388            let mut seen = std::collections::HashSet::with_capacity(ports.len());
389            for port in ports.iter() {
390                if seen.contains(port) {
391                    return Err(Error::Validation(format!(
392                        "ports must be unique, found multiple {port}"
393                    )));
394                }
395                seen.insert(port);
396            }
397        }
398
399        let base_url = format!("{}/{}/portforward?", self.url_path, name);
400        let mut qp = form_urlencoded::Serializer::new(base_url);
401        qp.append_pair(
402            "ports",
403            &ports.iter().map(|p| p.to_string()).collect::<Vec<_>>().join(","),
404        );
405
406        let req = http::Request::get(qp.finish());
407        req.body(vec![]).map_err(Error::BuildRequest)
408    }
409}
410
411// ----------------------------------------------------------------------------
412// tests
413// ----------------------------------------------------------------------------
414
415/// Cheap sanity check to ensure type maps work as expected
416#[cfg(test)]
417mod test {
418    use crate::{request::Request, resource::Resource};
419    use jiff::Timestamp;
420    use k8s::core::v1 as corev1;
421    use k8s_openapi::api as k8s;
422
423    use crate::subresource::LogParams;
424
425    #[test]
426    fn logs_all_params() {
427        let url = corev1::Pod::url_path(&(), Some("ns"));
428        let lp = LogParams {
429            container: Some("nginx".into()),
430            follow: true,
431            limit_bytes: Some(10 * 1024 * 1024),
432            pretty: true,
433            previous: true,
434            since_seconds: Some(3600),
435            since_time: None,
436            tail_lines: Some(4096),
437            timestamps: true,
438        };
439        let req = Request::new(url).logs("mypod", &lp).unwrap();
440        assert_eq!(
441            req.uri(),
442            "/api/v1/namespaces/ns/pods/mypod/log?&container=nginx&follow=true&limitBytes=10485760&pretty=true&previous=true&sinceSeconds=3600&tailLines=4096&timestamps=true"
443        );
444    }
445
446    #[test]
447    fn logs_since_time() {
448        let url = corev1::Pod::url_path(&(), Some("ns"));
449        let date: Timestamp = "2023-10-19T13:14:26Z".parse().unwrap();
450        let lp = LogParams {
451            since_seconds: None,
452            since_time: Some(date),
453            ..Default::default()
454        };
455        let req = Request::new(url).logs("mypod", &lp).unwrap();
456        assert_eq!(
457            req.uri(),
458            "/api/v1/namespaces/ns/pods/mypod/log?&sinceTime=2023-10-19T13%3A14%3A26Z" // cross-referenced with kubectl
459        );
460    }
461}
kube_core/subresource.rs

kube_core/
subresource.rs
