elasticctl_api/fleet/
integration_policies.rs

1use crate::compose::require_path_segment;
2use elasticctl_core::{Error, ErrorKind, Feature, Result, Transport, urlencode};
3use serde::{Deserialize, Deserializer, Serialize};
4use serde_json::{Map, Value};
5
6const BASE: &str = "/api/fleet/package_policies";
7const PACKAGES: &str = "/api/fleet/epm/packages";
8
9#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
10#[serde(deny_unknown_fields)]
11pub struct IntegrationPackageSpec {
12    pub name: String,
13    pub version: String,
14}
15
16#[derive(Debug, Clone, PartialEq, Serialize)]
17pub struct IntegrationPolicySpec {
18    pub id: String,
19    pub name: String,
20    #[serde(default, skip_serializing_if = "Option::is_none")]
21    pub description: Option<String>,
22    #[serde(default, skip_serializing_if = "Option::is_none")]
23    pub namespace: Option<String>,
24    pub policy_ids: Vec<String>,
25    pub package: IntegrationPackageSpec,
26    #[serde(default, skip_serializing_if = "Option::is_none")]
27    pub vars: Option<Map<String, Value>>,
28    #[serde(default, skip_serializing_if = "Option::is_none")]
29    pub var_group_selections: Option<Map<String, Value>>,
30    pub inputs: Map<String, Value>,
31    #[serde(default, skip_serializing_if = "Option::is_none")]
32    pub condition: Option<String>,
33    #[serde(default, skip_serializing_if = "Option::is_none")]
34    pub additional_datastreams_permissions: Option<Vec<String>>,
35}
36
37#[derive(Deserialize)]
38#[serde(deny_unknown_fields)]
39struct RawIntegrationPolicySpec {
40    id: String,
41    name: String,
42    #[serde(default)]
43    description: Option<String>,
44    #[serde(default)]
45    namespace: Option<String>,
46    policy_ids: Vec<String>,
47    package: IntegrationPackageSpec,
48    #[serde(default)]
49    vars: Option<Map<String, Value>>,
50    #[serde(default)]
51    var_group_selections: Option<Map<String, Value>>,
52    inputs: Map<String, Value>,
53    #[serde(default)]
54    condition: Option<String>,
55    #[serde(default)]
56    additional_datastreams_permissions: Option<Vec<String>>,
57}
58
59impl IntegrationPolicySpec {
60    fn from_raw(raw: RawIntegrationPolicySpec) -> Result<Self> {
61        let spec = Self {
62            id: raw.id,
63            name: raw.name,
64            description: raw.description,
65            namespace: raw.namespace,
66            policy_ids: raw.policy_ids,
67            package: raw.package,
68            vars: raw.vars,
69            var_group_selections: raw.var_group_selections,
70            inputs: raw.inputs,
71            condition: raw.condition,
72            additional_datastreams_permissions: raw.additional_datastreams_permissions,
73        };
74        spec.validate()?;
75        Ok(spec)
76    }
77
78    pub fn validate(&self) -> Result<()> {
79        for (field, value) in [
80            ("id", &self.id),
81            ("name", &self.name),
82            ("package.name", &self.package.name),
83            ("package.version", &self.package.version),
84        ] {
85            if value.trim().is_empty() {
86                return Err(Error::new(
87                    ErrorKind::Error,
88                    format!("integration policy {field} must not be empty"),
89                ));
90            }
91        }
92        if self.policy_ids.is_empty() {
93            return Err(Error::new(
94                ErrorKind::Error,
95                "integration policy policy_ids must not be empty",
96            ));
97        }
98        let mut previous = None;
99        for policy_id in &self.policy_ids {
100            if policy_id.trim().is_empty() {
101                return Err(Error::new(
102                    ErrorKind::Error,
103                    "integration policy policy_ids must not contain an empty id",
104                ));
105            }
106            if previous.is_some_and(|previous: &String| previous >= policy_id) {
107                return Err(Error::new(
108                    ErrorKind::Error,
109                    "integration policy policy_ids must be sorted and duplicate-free",
110                ));
111            }
112            previous = Some(policy_id);
113        }
114        if let Some(selections) = &self.var_group_selections {
115            for (name, selection) in selections {
116                if !selection.is_string() {
117                    return Err(Error::new(
118                        ErrorKind::Error,
119                        format!("integration policy var_group_selections.{name} must be a string"),
120                    ));
121                }
122            }
123        }
124        Ok(())
125    }
126}
127
128impl TryFrom<Value> for IntegrationPolicySpec {
129    type Error = Error;
130
131    fn try_from(value: Value) -> Result<Self> {
132        serde_json::from_value(value).map_err(|error| {
133            Error::new(
134                ErrorKind::Error,
135                format!("decoding integration policy: {error}"),
136            )
137        })
138    }
139}
140
141impl<'de> Deserialize<'de> for IntegrationPolicySpec {
142    fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
143    where
144        D: Deserializer<'de>,
145    {
146        let raw = RawIntegrationPolicySpec::deserialize(deserializer)?;
147        Self::from_raw(raw).map_err(serde::de::Error::custom)
148    }
149}
150
151#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
152pub struct IntegrationPolicySummary {
153    pub id: String,
154    pub name: String,
155    pub namespace: String,
156    pub description: Option<String>,
157    pub policy_ids: Vec<String>,
158    pub package: IntegrationPackageSpec,
159}
160
161#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
162pub struct IntegrationPolicyDetail {
163    pub id: String,
164    pub name: String,
165    pub namespace: String,
166    pub description: Option<String>,
167    pub policy_ids: Vec<String>,
168    pub package: IntegrationPackageSpec,
169    pub affected_agents: u64,
170    pub blocked_by: Vec<String>,
171}
172
173#[derive(Debug, Clone, PartialEq)]
174pub struct IntegrationPolicyPage {
175    pub items: Vec<Map<String, Value>>,
176    pub total: u64,
177    pub page: u64,
178    pub per_page: u64,
179}
180
181#[derive(Debug, Clone, PartialEq)]
182pub struct IntegrationPolicy {
183    pub item: Map<String, Value>,
184}
185
186#[derive(Debug, Clone, PartialEq)]
187pub struct PackageMetadata {
188    pub(crate) item: Map<String, Value>,
189}
190
191pub async fn list_page(transport: &Transport, page: u64) -> Result<IntegrationPolicyPage> {
192    transport.require_feature(Feature::FleetPolicies).await?;
193    let body = transport
194        .get(&format!(
195            "{BASE}?page={page}&perPage=1000&sortField=created_at&sortOrder=asc&format=simplified"
196        ))
197        .await?;
198    let envelope: PageEnvelope = decode(&body, "integration policies list")?;
199    Ok(IntegrationPolicyPage {
200        items: envelope.items,
201        total: envelope.total,
202        page: envelope.page,
203        per_page: envelope.per_page,
204    })
205}
206
207pub async fn get(transport: &Transport, id: &str) -> Result<IntegrationPolicy> {
208    get_selector(transport, require_path_segment(id)?).await
209}
210
211pub(crate) async fn get_selector(
212    transport: &Transport,
213    selector: &str,
214) -> Result<IntegrationPolicy> {
215    transport.require_feature(Feature::FleetPolicies).await?;
216    decode_item(
217        &transport
218            .get(&format!("{}?format=simplified", policy_path(selector)))
219            .await?,
220        "integration policy get",
221    )
222}
223
224pub async fn create(
225    transport: &Transport,
226    spec: &IntegrationPolicySpec,
227) -> Result<IntegrationPolicy> {
228    transport.require_feature(Feature::FleetPolicies).await?;
229    spec.validate()?;
230    let body = encode_spec(spec, "create")?;
231    decode_item(
232        &transport.post(BASE, Some(&body)).await?,
233        "integration policy create",
234    )
235}
236
237pub async fn update(
238    transport: &Transport,
239    id: &str,
240    spec: &IntegrationPolicySpec,
241) -> Result<IntegrationPolicy> {
242    let id = require_path_segment(id)?;
243    transport.require_feature(Feature::FleetPolicies).await?;
244    spec.validate()?;
245    let mut body = encode_spec(spec, "update")?;
246    let object = body
247        .as_object_mut()
248        .expect("integration policy serialization is an object");
249    object.remove("id");
250    decode_item(
251        &transport.put(&policy_path(id), &body).await?,
252        "integration policy update",
253    )
254}
255
256pub async fn delete(transport: &Transport, id: &str) -> Result<()> {
257    let id = require_path_segment(id)?;
258    transport.require_feature(Feature::FleetPolicies).await?;
259    let response = transport.delete(&policy_path(id)).await?;
260    let deleted: DeleteEnvelope = decode(&response, "integration policy delete")?;
261    if deleted.id != id {
262        return Err(Error::with_status(
263            ErrorKind::Http,
264            200,
265            format!(
266                "decoding integration policy delete: expected id '{id}', got '{}'",
267                deleted.id
268            ),
269        ));
270    }
271    Ok(())
272}
273
274pub async fn package_metadata(
275    transport: &Transport,
276    name: &str,
277    version: &str,
278) -> Result<PackageMetadata> {
279    transport.require_feature(Feature::FleetPolicies).await?;
280    for (field, value) in [("name", name), ("version", version)] {
281        if value.trim().is_empty() {
282            return Err(Error::new(
283                ErrorKind::Error,
284                format!("integration package metadata {field} must not be empty"),
285            ));
286        }
287        require_path_segment(value)?;
288    }
289    let body = transport
290        .get(&format!(
291            "{PACKAGES}/{}/{}",
292            urlencode(name),
293            urlencode(version)
294        ))
295        .await?;
296    let envelope: ItemEnvelope = decode(&body, "integration package metadata")?;
297    let returned_name = required_string(&envelope.item, "name", "integration package metadata")?;
298    let returned_version =
299        required_string(&envelope.item, "version", "integration package metadata")?;
300    if returned_name != name || returned_version != version {
301        return Err(Error::new(
302            ErrorKind::Http,
303            format!(
304                "decoding integration package metadata: expected {name}@{version}, got {returned_name}@{returned_version}"
305            ),
306        ));
307    }
308    Ok(PackageMetadata {
309        item: envelope.item,
310    })
311}
312
313fn encode_spec(spec: &IntegrationPolicySpec, context: &str) -> Result<Value> {
314    serde_json::to_value(spec).map_err(|error| {
315        Error::new(
316            ErrorKind::Error,
317            format!("encoding integration policy {context}: {error}"),
318        )
319    })
320}
321
322fn policy_path(id: &str) -> String {
323    format!("{BASE}/{}", urlencode(id))
324}
325
326fn decode_item(body: &Value, context: &str) -> Result<IntegrationPolicy> {
327    let envelope: ItemEnvelope = decode(body, context)?;
328    Ok(IntegrationPolicy {
329        item: envelope.item,
330    })
331}
332
333fn decode<T: serde::de::DeserializeOwned>(body: &Value, context: &str) -> Result<T> {
334    serde_json::from_value(body.clone())
335        .map_err(|error| Error::new(ErrorKind::Http, format!("decoding {context}: {error}")))
336}
337
338fn required_string(item: &Map<String, Value>, field: &str, context: &str) -> Result<String> {
339    item.get(field)
340        .and_then(Value::as_str)
341        .filter(|value| !value.trim().is_empty())
342        .map(str::to_owned)
343        .ok_or_else(|| {
344            Error::new(
345                ErrorKind::Http,
346                format!("decoding {context}: {field} must be a non-empty string"),
347            )
348        })
349}
350
351#[derive(Deserialize)]
352#[serde(deny_unknown_fields)]
353struct PageEnvelope {
354    items: Vec<Map<String, Value>>,
355    total: u64,
356    page: u64,
357    #[serde(rename = "perPage")]
358    per_page: u64,
359}
360
361#[derive(Deserialize)]
362#[serde(deny_unknown_fields)]
363struct ItemEnvelope {
364    item: Map<String, Value>,
365}
366
367#[derive(Deserialize)]
368#[serde(deny_unknown_fields)]
369struct DeleteEnvelope {
370    id: String,
371}
elasticctl_api/fleet/integration_policies.rs

elasticctl_api/fleet/
integration_policies.rs
