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
324
325
326
use either::Either;
use futures::Stream;
use serde::{de::DeserializeOwned, Serialize};
use std::marker::PhantomData;
use crate::{
api::{DeleteParams, ListParams, Meta, ObjectList, PatchParams, PostParams, Resource, WatchEvent},
client::{Client, Status},
Result,
};
/// An easy Api interaction helper
///
/// The upsides of working with this rather than a `Resource` directly are:
/// - easiers serialization interface (no figuring out return types)
/// - client hidden within, less arguments
///
/// But the downsides are:
/// - openapi types can take up a large amount of memory
/// - openapi types can be annoying to wrangle with their heavy Option use
/// - no control over requests (opinionated)
#[derive(Clone)]
pub struct Api<K> {
/// The request creator object
pub(crate) resource: Resource,
/// The client to use (from this library)
pub(crate) client: Client,
/// Underlying Object unstored
pub(crate) phantom: PhantomData<K>,
}
/// Expose same interface as Api for controlling scope/group/versions/ns
impl<K> Api<K>
where
K: k8s_openapi::Resource,
{
/// Cluster level resources, or resources viewed across all namespaces
pub fn all(client: Client) -> Self {
let resource = Resource::all::<K>();
Self {
resource,
client,
phantom: PhantomData,
}
}
/// Namespaced resource within a given namespace
pub fn namespaced(client: Client, ns: &str) -> Self {
let resource = Resource::namespaced::<K>(ns);
Self {
resource,
client,
phantom: PhantomData,
}
}
/// Consume self and return the [`Client`]
pub fn into_client(self) -> Client {
self.into()
}
}
/// PUSH/PUT/POST/GET abstractions
impl<K> Api<K>
where
K: Clone + DeserializeOwned + Meta,
{
/// Get a named resource
///
/// ```no_run
/// use kube::{Api, Client};
/// use k8s_openapi::api::core::v1::Pod;
/// #[tokio::main]
/// async fn main() -> Result<(), kube::Error> {
/// let client = Client::try_default().await?;
/// let pods: Api<Pod> = Api::namespaced(client, "apps");
/// let p: Pod = pods.get("blog").await?;
/// Ok(())
/// }
/// ```
pub async fn get(&self, name: &str) -> Result<K> {
let req = self.resource.get(name)?;
self.client.request::<K>(req).await
}
/// Get a list of resources
///
/// You get use this to get everything, or a subset matching fields/labels, say:
///
/// ```no_run
/// use kube::{api::{Api, ListParams, Meta}, Client};
/// use k8s_openapi::api::core::v1::Pod;
/// #[tokio::main]
/// async fn main() -> Result<(), kube::Error> {
/// let client = Client::try_default().await?;
/// let pods: Api<Pod> = Api::namespaced(client, "apps");
/// let lp = ListParams::default().labels("app=blog"); // for this app only
/// for p in pods.list(&lp).await? {
/// println!("Found Pod: {}", Meta::name(&p));
/// }
/// Ok(())
/// }
/// ```
pub async fn list(&self, lp: &ListParams) -> Result<ObjectList<K>> {
let req = self.resource.list(&lp)?;
self.client.request::<ObjectList<K>>(req).await
}
/// Create a resource
///
/// This function requires a type that Serializes to `K`, which can be:
/// 1. Raw string yaml
/// - easy to port from existing files
/// - error prone (run-time errors on typos due to failed serialize attempts)
/// - very error prone (can write invalid yaml)
/// 2. An instance of the struct itself
/// - easy to instantiate for CRDs (you define the struct)
/// - dense to instantiate for k8s-openapi types (due to many optionals)
/// - compile-time safety
/// - but still possible to write invalid native types (validation at apiserver)
/// 3. `serde_json::json!` macro instantiated `serde_json::Value`
/// - Tradeoff between the two
/// - Easy partially filling of native k8s-openapi types (most fields optional)
/// - Partial safety against runtime errors (at least you must write valid json)
pub async fn create(&self, pp: &PostParams, data: &K) -> Result<K>
where
K: Serialize,
{
let bytes = serde_json::to_vec(&data)?;
let req = self.resource.create(&pp, bytes)?;
self.client.request::<K>(req).await
}
/// Delete a named resource
///
/// When you get a `K` via `Left`, your delete has started.
/// When you get a `Status` via `Right`, this should be a a 2XX style
/// confirmation that the object being gone.
///
/// 4XX and 5XX status types are returned as an `Err(kube::Error::Api)`
///
/// ```no_run
/// use kube::{api::{Api, DeleteParams}, Client};
/// use k8s_openapi::apiextensions_apiserver::pkg::apis::apiextensions::v1beta1 as apiexts;
/// use apiexts::CustomResourceDefinition;
/// #[tokio::main]
/// async fn main() -> Result<(), kube::Error> {
/// let client = Client::try_default().await?;
/// let crds: Api<CustomResourceDefinition> = Api::all(client);
/// crds.delete("foos.clux.dev", &DeleteParams::default()).await?
/// .map_left(|o| println!("Deleting CRD: {:?}", o.status))
/// .map_right(|s| println!("Deleted CRD: {:?}", s));
/// Ok(())
/// }
/// ```
pub async fn delete(&self, name: &str, dp: &DeleteParams) -> Result<Either<K, Status>> {
let req = self.resource.delete(name, &dp)?;
self.client.request_status::<K>(req).await
}
/// Delete a collection of resources
///
/// When you get an `ObjectList<K>` via `Left`, your delete has started.
/// When you get a `Status` via `Right`, this should be a a 2XX style
/// confirmation that the object being gone.
///
/// 4XX and 5XX status types are returned as an `Err(kube::Error::Api)`
///
/// ```no_run
/// use kube::{api::{Api, ListParams, Meta}, Client};
/// use k8s_openapi::api::core::v1::Pod;
/// #[tokio::main]
/// async fn main() -> Result<(), kube::Error> {
/// let client = Client::try_default().await?;
/// let pods: Api<Pod> = Api::namespaced(client, "apps");
/// match pods.delete_collection(&ListParams::default()).await? {
/// either::Left(list) => {
/// let names: Vec<_> = list.iter().map(Meta::name).collect();
/// println!("Deleting collection of pods: {:?}", names);
/// },
/// either::Right(status) => {
/// println!("Deleted collection of pods: status={:?}", status);
/// }
/// }
/// Ok(())
/// }
/// ```
pub async fn delete_collection(&self, lp: &ListParams) -> Result<Either<ObjectList<K>, Status>> {
let req = self.resource.delete_collection(&lp)?;
self.client.request_status::<ObjectList<K>>(req).await
}
/// Patch a resource a subset of its properties
///
/// In all patch methods except PatchStrategy::Apply, you must set:
/// `metadata.resourceVersion` to get k8s to accept the update.
///
/// Thus to use these older patch methods you must first do a `get` then a `patch`.
///
/// When using `PatchStrategy::Apply`, this restriction is not necessary,
/// however, you **must** serialize your data using `serde_yaml`.
///
/// ```no_run
/// use kube::{api::{Api, PatchParams, Meta}, Client};
/// use k8s_openapi::api::core::v1::Pod;
/// #[tokio::main]
/// async fn main() -> Result<(), anyhow::Error> {
/// let client = Client::try_default().await?;
/// let pods: Api<Pod> = Api::namespaced(client, "apps");
/// let ss_apply = PatchParams::default_apply().force();
/// let patch = serde_yaml::to_vec(&serde_json::json!({
/// "apiVersion": "v1",
/// "kind": "Pod",
/// "metadata": {
/// "name": "blog"
/// },
/// "spec": {
/// "activeDeadlineSeconds": 5
/// }
/// }))?;
/// let o_patched = pods.patch("blog", &ss_apply, patch).await?;
/// Ok(())
/// }
/// ```
pub async fn patch(&self, name: &str, pp: &PatchParams, patch: Vec<u8>) -> Result<K> {
let req = self.resource.patch(name, &pp, patch)?;
self.client.request::<K>(req).await
}
/// Replace a resource entirely with a new one
///
/// This is used just like `Api::create`, but with one additional instruction:
/// You must set `metadata.resourceVersion` in the provided data because k8s
/// will not accept an update unless you actually knew what the last version was.
///
/// Thus, to use this function, you need to do a `get` then a `replace` with its result.
///
/// ```no_run
/// use kube::{api::{Api, PostParams, Meta}, Client};
/// use k8s_openapi::api::batch::v1::Job;
/// #[tokio::main]
/// async fn main() -> Result<(), kube::Error> {
/// let client = Client::try_default().await?;
/// let jobs: Api<Job> = Api::namespaced(client, "apps");
/// let j = jobs.get("baz").await?;
/// let j_new: Job = serde_json::from_value(serde_json::json!({
/// "apiVersion": "batch/v1",
/// "kind": "Job",
/// "metadata": {
/// "name": "baz",
/// "resourceVersion": Meta::resource_ver(&j),
/// },
/// "spec": {
/// "template": {
/// "metadata": {
/// "name": "empty-job-pod"
/// },
/// "spec": {
/// "containers": [{
/// "name": "empty",
/// "image": "alpine:latest"
/// }],
/// "restartPolicy": "Never",
/// }
/// }
/// }
/// }))?;
/// jobs.replace("baz", &PostParams::default(), &j_new).await?;
/// Ok(())
/// }
/// ```
///
/// Consider mutating the result of `api.get` rather than recreating it.
pub async fn replace(&self, name: &str, pp: &PostParams, data: &K) -> Result<K>
where
K: Serialize,
{
let bytes = serde_json::to_vec(&data)?;
let req = self.resource.replace(name, &pp, bytes)?;
self.client.request::<K>(req).await
}
/// Watch a list of resources
///
/// This returns a future that awaits the initial response,
/// then you can stream the remaining buffered `WatchEvent` objects.
///
/// ```no_run
/// use kube::{api::{Api, ListParams, Meta, WatchEvent}, Client};
/// use k8s_openapi::api::batch::v1::Job;
/// use futures::{StreamExt, TryStreamExt};
/// #[tokio::main]
/// async fn main() -> Result<(), kube::Error> {
/// let client = Client::try_default().await?;
/// let jobs: Api<Job> = Api::namespaced(client, "apps");
/// let lp = ListParams::default()
/// .fields("metadata.name=my_job")
/// .timeout(20); // upper bound of how long we watch for
/// let mut stream = jobs.watch(&lp, "0").await?.boxed();
/// while let Some(status) = stream.try_next().await? {
/// match status {
/// WatchEvent::Added(s) => println!("Added {}", Meta::name(&s)),
/// WatchEvent::Modified(s) => println!("Modified: {}", Meta::name(&s)),
/// WatchEvent::Deleted(s) => println!("Deleted {}", Meta::name(&s)),
/// WatchEvent::Bookmark(s) => {},
/// WatchEvent::Error(s) => println!("{}", s),
/// }
/// }
/// Ok(())
/// }
/// ```
pub async fn watch(
&self,
lp: &ListParams,
version: &str,
) -> Result<impl Stream<Item = Result<WatchEvent<K>>>> {
let req = self.resource.watch(&lp, &version)?;
self.client.request_events::<K>(req).await
}
}
impl<K> From<Api<K>> for Client {
fn from(api: Api<K>) -> Self {
api.client
}
}