roperator/runner/client/
request.rs

1use crate::config::{ClientConfig, Credentials};
2use crate::k8s_types::K8sType;
3use crate::resource::{K8sResource, ObjectIdRef};
4use crate::runner::client::Error;
5
6use http::{header, Method, Request};
7use hyper::Body;
8use serde_json::Value;
9use url::Url;
10
11#[allow(dead_code)]
12#[derive(Debug, PartialEq, Eq, Clone, Copy)]
13pub enum MergeStrategy {
14    Json,
15    JsonMerge,
16    StrategicMerge,
17}
18
19impl MergeStrategy {
20    fn content_type(self) -> &'static str {
21        match self {
22            MergeStrategy::Json => "application/json-patch+json",
23            MergeStrategy::JsonMerge => "application/merge-patch+json",
24            MergeStrategy::StrategicMerge => "application/strategic-merge-patch+json",
25        }
26    }
27}
28
29#[derive(Debug, PartialEq, Clone)]
30pub struct Patch {
31    merge_strategy: MergeStrategy,
32    value: Value,
33}
34
35impl Patch {
36    pub fn remove_finalizer(resource: &K8sResource, finalizer: &str) -> Patch {
37        let finalizers = resource
38            .as_ref()
39            .pointer("/metadata/finalizers")
40            .and_then(Value::as_array)
41            .map(|finalizers| {
42                finalizers
43                    .iter()
44                    .filter(|f| f.as_str() != Some(finalizer))
45                    .collect::<Vec<_>>()
46            })
47            .unwrap_or_default();
48        let patch = serde_json::json!({
49            "metadata": {
50                "namespace": resource.get_object_id().namespace(),
51                "name": resource.get_object_id().name(),
52                "resourceVersion": resource.resource_version(),
53                "finalizers": finalizers,
54            }
55        });
56        Patch {
57            value: patch,
58            merge_strategy: MergeStrategy::JsonMerge,
59        }
60    }
61
62    pub fn add_finalizer(resource: &K8sResource, finalizer: &str) -> Patch {
63        let mut finalizers = resource
64            .as_ref()
65            .pointer("/metadata/finalizers")
66            .and_then(Value::as_array)
67            .cloned()
68            .unwrap_or_default();
69        finalizers.push(Value::String(finalizer.to_string()));
70        let value = serde_json::json!({
71            "metadata": {
72                "namespace": resource.get_object_id().namespace(),
73                "name": resource.get_object_id().name(),
74                "resourceVersion": resource.resource_version(),
75                "finalizers": finalizers,
76            }
77        });
78        Patch {
79            value,
80            merge_strategy: MergeStrategy::JsonMerge,
81        }
82    }
83}
84
85pub fn patch_request(
86    client_config: &ClientConfig,
87    k8s_type: &K8sType,
88    id: &ObjectIdRef<'_>,
89    patch: &Patch,
90) -> Result<Request<Body>, Error> {
91    let url = make_url(client_config, k8s_type, id.namespace(), Some(id.name()));
92    let header_value = patch.merge_strategy.content_type();
93    let builder =
94        make_req(url, Method::PATCH, client_config).header(header::CONTENT_TYPE, header_value);
95    let body = serde_json::to_vec(&patch.value)?;
96    let req = builder.body(Body::from(body)).unwrap();
97    Ok(req)
98}
99
100#[cfg(feature = "testkit")]
101pub fn get_request(
102    client_config: &ClientConfig,
103    k8s_type: &K8sType,
104    id: &ObjectIdRef<'_>,
105) -> Result<Request<Body>, Error> {
106    let url = make_url(client_config, k8s_type, id.namespace(), Some(id.name()));
107
108    let req = make_req(url, Method::GET, client_config)
109        .body(Body::empty())
110        .unwrap();
111    Ok(req)
112}
113
114pub fn create_request(
115    client_config: &ClientConfig,
116    k8s_type: &K8sType,
117    resource: &Value,
118) -> Result<Request<Body>, Error> {
119    let url = make_url(client_config, k8s_type, get_namespace(resource), None);
120
121    let builder = make_req(url, Method::POST, client_config);
122    let as_vec = serde_json::to_vec(resource)?;
123    let req = builder.body(Body::from(as_vec)).unwrap();
124    Ok(req)
125}
126
127pub fn replace_request(
128    client_config: &ClientConfig,
129    k8s_type: &K8sType,
130    id: &ObjectIdRef<'_>,
131    resource: &Value,
132) -> Result<Request<Body>, Error> {
133    let url = make_url(client_config, k8s_type, id.namespace(), Some(id.name()));
134    let as_vec = serde_json::to_vec(resource)?;
135    let req = make_req(url, Method::PUT, client_config)
136        .body(Body::from(as_vec))
137        .unwrap();
138    Ok(req)
139}
140
141pub fn update_status_request(
142    client_config: &ClientConfig,
143    k8s_type: &K8sType,
144    id: &ObjectIdRef<'_>,
145    new_status: &Value,
146) -> Result<Request<Body>, Error> {
147    let mut url = make_url(client_config, k8s_type, id.namespace(), Some(id.name()));
148    {
149        let mut path = url.path_segments_mut().unwrap();
150        path.push("status");
151    }
152    let as_vec = serde_json::to_vec(new_status)?;
153    let req = make_req(url, Method::PUT, client_config)
154        .body(Body::from(as_vec))
155        .unwrap();
156    Ok(req)
157}
158
159pub fn delete_request(
160    client_config: &ClientConfig,
161    k8s_type: &K8sType,
162    id: &ObjectIdRef<'_>,
163) -> Result<Request<Body>, Error> {
164    let url = make_url(client_config, k8s_type, id.namespace(), Some(id.name()));
165    let req = make_req(url, Method::DELETE, client_config)
166        .body(Body::empty())
167        .unwrap();
168    Ok(req)
169}
170
171pub fn watch_request(
172    client_config: &ClientConfig,
173    k8s_type: &K8sType,
174    resource_version: Option<&str>,
175    label_selector: Option<&str>,
176    timeout_seconds: Option<u32>,
177    namespace: Option<&str>,
178) -> Result<Request<Body>, Error> {
179    let mut url = make_url(client_config, k8s_type, namespace, None);
180    {
181        let mut query = url.query_pairs_mut();
182        query.append_pair("watch", "true");
183        if let Some(vers) = resource_version {
184            query.append_pair("resourceVersion", vers);
185        }
186        if let Some(selector) = label_selector {
187            query.append_pair("labelSelector", selector);
188        }
189        if let Some(timeout) = timeout_seconds {
190            let as_str = format!("{}", timeout);
191            query.append_pair("timeoutSeconds", &as_str);
192        }
193    }
194
195    let req = make_req(url, Method::GET, client_config)
196        .body(Body::empty())
197        .unwrap();
198    Ok(req)
199}
200
201pub fn list_request(
202    client_config: &ClientConfig,
203    k8s_type: &K8sType,
204    label_selector: Option<&str>,
205    namespace: Option<&str>,
206) -> Result<Request<Body>, Error> {
207    let mut url = make_url(client_config, k8s_type, namespace, None);
208    if let Some(selector) = label_selector {
209        let mut query = url.query_pairs_mut();
210        query.append_pair("labelSelector", selector);
211    }
212    let req = make_req(url, Method::GET, client_config)
213        .body(Body::empty())
214        .unwrap();
215    Ok(req)
216}
217
218fn make_req(
219    url: Url,
220    method: http::Method,
221    client_config: &ClientConfig,
222) -> http::request::Builder {
223    let builder = Request::builder()
224        .method(method)
225        .uri(url.into_string())
226        .header(header::ACCEPT, "application/json")
227        .header(header::USER_AGENT, client_config.user_agent.as_str());
228
229    if let Credentials::Header(ref value) = client_config.credentials {
230        builder.header(header::AUTHORIZATION, value)
231    } else {
232        builder
233    }
234}
235
236fn get_namespace(resource: &Value) -> Option<&str> {
237    resource
238        .pointer("/metadata/namespace")
239        .and_then(Value::as_str)
240}
241
242fn make_url(
243    client_config: &ClientConfig,
244    k8s_type: &K8sType,
245    namespace: Option<&str>,
246    name: Option<&str>,
247) -> Url {
248    let mut url = url::Url::parse(client_config.api_server_endpoint.as_str()).unwrap();
249    {
250        let mut segments = url.path_segments_mut().unwrap();
251        let group = k8s_type.group();
252
253        // most k8s resources are under /apis/<group>/<version>, but the so called "core v1" resources
254        // live under /api/v1, which is why we check the group here
255        let prefix = if group.is_empty() { "api" } else { "apis" };
256        segments.push(prefix);
257        if !group.is_empty() {
258            segments.push(group);
259        }
260        segments.push(k8s_type.version());
261        if let Some(ns) = namespace {
262            segments.push("namespaces");
263            segments.push(ns);
264        }
265        segments.push(k8s_type.plural_kind);
266
267        if let Some(n) = name {
268            segments.push(n);
269        }
270    }
271    url
272}
roperator/runner/client/request.rs

roperator/runner/client/
request.rs
