elasticctl_api/fleet/integration_policy_ops/
import_shared.rs

1use super::common::*;
2use super::read::collect;
3
4const IMPORT_RACE_WARNING: &str =
5    "warning  Fleet can change after the final recheck and before the write";
6
7pub struct IntegrationPolicyImportArtifact {
8    pub(super) source: PathBuf,
9    pub(super) canonical: Vec<IntegrationPolicySpec>,
10}
11
12impl fmt::Debug for IntegrationPolicyImportArtifact {
13    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
14        formatter
15            .debug_struct("IntegrationPolicyImportArtifact")
16            .field("policy_count", &self.canonical.len())
17            .finish()
18    }
19}
20
21#[derive(Clone, PartialEq)]
22pub struct IntegrationPolicyImportPlan {
23    pub preview: MutationPlan,
24    pub skipped: Vec<Value>,
25    pub package_installs: Vec<String>,
26    pub total: usize,
27    pub(super) source: PathBuf,
28    pub(super) host: String,
29    pub(super) space: String,
30    pub(super) canonical: Vec<IntegrationPolicySpec>,
31    pub(super) name_owners: BTreeMap<String, BTreeSet<String>>,
32    pub(super) name_owners_snapshot: BTreeMap<String, BTreeSet<String>>,
33    pub(super) parent_snapshots: BTreeMap<String, agent_policy_ops::AgentPolicyParentSnapshot>,
34    pub(super) skipped_snapshot: Vec<Value>,
35    pub(super) existing_snapshot: BTreeMap<String, Option<Map<String, Value>>>,
36    pub(super) targets: Vec<IntegrationPolicyImportTarget>,
37    pub(super) package_groups: BTreeMap<String, IntegrationPackageGroup>,
38    pub(super) overwrite: bool,
39    pub(super) skip_existing: bool,
40}
41
42impl fmt::Debug for IntegrationPolicyImportPlan {
43    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
44        formatter
45            .debug_struct("IntegrationPolicyImportPlan")
46            .field("target_count", &self.preview.targets.len())
47            .field("skipped_count", &self.skipped.len())
48            .field("package_install_count", &self.package_installs.len())
49            .field("existing_snapshot_count", &self.existing_snapshot.len())
50            .field("total", &self.total)
51            .finish()
52    }
53}
54
55#[derive(Debug, Clone, PartialEq)]
56pub(super) struct IntegrationPolicyImportTarget {
57    pub(super) effective: IntegrationPolicySpec,
58    pub(super) current: Option<IntegrationPolicyCurrentSnapshot>,
59    pub(super) parents: BTreeMap<String, agent_policy_ops::AgentPolicyParentSnapshot>,
60    pub(super) replacement_body: Option<Value>,
61}
62
63#[derive(Debug, Clone, PartialEq)]
64pub(super) struct IntegrationPolicyCurrentSnapshot {
65    pub(super) item: Map<String, Value>,
66    pub(super) spec: IntegrationPolicySpec,
67    pub(super) parent_ids: Vec<String>,
68}
69
70#[derive(Debug, Clone, PartialEq)]
71pub(super) struct IntegrationPackageGroup {
72    pub(super) package: IntegrationPackageSpec,
73    pub(super) state: PackageDependencySnapshot,
74    pub(super) state_snapshot: PackageDependencySnapshot,
75    pub(super) metadata: Map<String, Value>,
76    pub(super) metadata_snapshot: Map<String, Value>,
77}
78
79pub(super) fn validate_requested_package_versions(specs: &[IntegrationPolicySpec]) -> Result<()> {
80    let mut versions = BTreeMap::new();
81    for spec in specs {
82        match versions.entry(spec.package.name.as_str()) {
83            std::collections::btree_map::Entry::Vacant(entry) => {
84                entry.insert(spec.package.version.as_str());
85            }
86            std::collections::btree_map::Entry::Occupied(entry)
87                if entry.get() != &spec.package.version.as_str() =>
88            {
89                return Err(Error::new(
90                    ErrorKind::Conflict,
91                    format!(
92                        "integration package '{}' is requested at more than one version",
93                        spec.package.name
94                    ),
95                ));
96            }
97            std::collections::btree_map::Entry::Occupied(_) => {}
98        }
99    }
100    Ok(())
101}
102
103pub(super) async fn relevant_name_owners(
104    transport: &Transport,
105    names: &BTreeSet<String>,
106) -> Result<BTreeMap<String, BTreeSet<String>>> {
107    let mut owners = names
108        .iter()
109        .map(|name| (name.clone(), BTreeSet::new()))
110        .collect::<BTreeMap<_, _>>();
111    if names.is_empty() {
112        return Ok(owners);
113    }
114    for item in collect(transport).await? {
115        let Some(name) = item.get("name").and_then(Value::as_str) else {
116            continue;
117        };
118        let Some(owners) = owners.get_mut(name) else {
119            continue;
120        };
121        owners.insert(required_string(&item, "id", "integration policies list")?);
122    }
123    Ok(owners)
124}
125
126pub(super) fn target_name_owners_match(
127    target: &IntegrationPolicyImportTarget,
128    owners: &BTreeMap<String, BTreeSet<String>>,
129) -> bool {
130    let Some(actual) = owners.get(&target.effective.name) else {
131        return false;
132    };
133    let expected = match &target.current {
134        None => BTreeSet::new(),
135        Some(current) if current.spec.name == target.effective.name => {
136            BTreeSet::from([target.effective.id.clone()])
137        }
138        Some(_) => BTreeSet::new(),
139    };
140    actual == &expected
141}
142
143pub(super) async fn read_import_parent_snapshots(
144    transport: &Transport,
145    integration_id: &str,
146    current_parent_ids: &[String],
147    desired_parent_ids: &[String],
148) -> Result<BTreeMap<String, agent_policy_ops::AgentPolicyParentSnapshot>> {
149    let parent_ids = current_parent_ids
150        .iter()
151        .chain(desired_parent_ids)
152        .cloned()
153        .collect::<BTreeSet<_>>();
154    let mut parents = BTreeMap::new();
155    for parent_id in parent_ids {
156        let parent = agent_policy_ops::read_parent_snapshot(transport, &parent_id).await?;
157        if current_parent_ids.binary_search(&parent_id).is_ok()
158            && parent
159                .attached_integrations
160                .binary_search_by(|attached| attached.as_str().cmp(integration_id))
161                .is_err()
162        {
163            return Err(http(format!(
164                "decoding integration policy '{integration_id}': parent '{parent_id}' is missing its attachment"
165            )));
166        }
167        parents.insert(parent_id, parent);
168    }
169    Ok(parents)
170}
171
172pub(super) fn effective_import_spec(
173    canonical: &IntegrationPolicySpec,
174    parents: &BTreeMap<String, agent_policy_ops::AgentPolicyParentSnapshot>,
175) -> Result<IntegrationPolicySpec> {
176    canonical.validate()?;
177    for parent in parents.values() {
178        if parent.platform_owned {
179            return unsupported(format!(
180                "integration policy '{}' is not portable: parent {} is platform-owned",
181                canonical.id, parent.id
182            ));
183        }
184        if parent.protected {
185            return unsupported(format!(
186                "integration policy '{}' is not portable: parent {} is_protected",
187                canonical.id, parent.id
188            ));
189        }
190    }
191    let selected = canonical
192        .policy_ids
193        .iter()
194        .map(|id| {
195            parents.get(id).ok_or_else(|| {
196                Error::new(
197                    ErrorKind::Error,
198                    format!(
199                        "integration policy '{}' has no parent snapshot for '{id}'",
200                        canonical.id
201                    ),
202                )
203            })
204        })
205        .collect::<Result<Vec<_>>>()?;
206    let mut effective = canonical.clone();
207    if let Some(namespace) = &effective.namespace {
208        if selected.iter().any(|parent| &parent.namespace != namespace) {
209            return unsupported(format!(
210                "integration policy '{}' is not portable: namespace does not match every parent",
211                canonical.id
212            ));
213        }
214    } else {
215        let namespaces = selected
216            .iter()
217            .map(|parent| parent.namespace.as_str())
218            .collect::<BTreeSet<_>>();
219        if namespaces.len() != 1 {
220            return Err(Error::new(
221                ErrorKind::Conflict,
222                format!(
223                    "integration policy '{}' is not portable: parents have different namespaces",
224                    canonical.id
225                ),
226            ));
227        }
228        effective.namespace = namespaces.into_iter().next().map(str::to_owned);
229    }
230    Ok(effective)
231}
232
233pub(super) fn replace_wire_body(spec: &IntegrationPolicySpec) -> Result<Value> {
234    spec.validate()?;
235    let mut body = serde_json::to_value(spec)
236        .map_err(|error| {
237            Error::new(
238                ErrorKind::Error,
239                format!("encoding integration policy: {error}"),
240            )
241        })?
242        .as_object()
243        .cloned()
244        .expect("integration policy specs serialize to objects");
245    body.remove("id");
246    Ok(Value::Object(body))
247}
248
249pub(super) fn planned_package_installs(
250    groups: &BTreeMap<String, IntegrationPackageGroup>,
251) -> Vec<String> {
252    groups
253        .values()
254        .filter_map(|group| match group.state.state {
255            PackageDependencyState::NotInstalled => {
256                Some(format!("{}@{}", group.package.name, group.package.version))
257            }
258            PackageDependencyState::Installed { .. } => None,
259        })
260        .collect()
261}
262
263pub(super) fn import_preview(
264    path: &Path,
265    targets: &[IntegrationPolicyImportTarget],
266    package_installs: &[String],
267) -> MutationPlan {
268    let mut details = targets
269        .iter()
270        .map(|target| {
271            let parents = target
272                .parents
273                .values()
274                .map(|parent| format!("{} ({})", parent.id, parent.name))
275                .collect::<Vec<_>>()
276                .join(", ");
277            let agents = target
278                .parents
279                .values()
280                .map(|parent| parent.agents)
281                .sum::<u64>();
282            let action = match &target.current {
283                None => "create".to_owned(),
284                Some(current) if current.spec == target.effective => "unchanged".to_owned(),
285                Some(current) if current.spec.name == target.effective.name => "replace".to_owned(),
286                Some(current) => format!(
287                    "replace  {} -> {}",
288                    current.spec.name, target.effective.name
289                ),
290            };
291            format!(
292                "{}  {action}  {}  parents {parents}  agents {agents}",
293                target.effective.id, target.effective.name
294            )
295        })
296        .collect::<Vec<_>>();
297    details.extend(
298        package_installs
299            .iter()
300            .map(|package| format!("package install  {package}")),
301    );
302    details.push(IMPORT_RACE_WARNING.to_owned());
303    MutationPlan {
304        preview_action: format!(
305            "Import {} integration policy(ies) from {}",
306            targets.len(),
307            path.display()
308        ),
309        preview_details: details,
310        targets: targets
311            .iter()
312            .map(|target| target.effective.id.clone())
313            .collect(),
314    }
315}
316
317pub(super) fn import_remote_error(error: Error, context: &str) -> Error {
318    let message = format!("integration-policy import {context} failed");
319    match error.http_status {
320        Some(status) => Error::with_status(error.kind, status, message),
321        None => Error::new(error.kind, message),
322    }
323}
elasticctl_api/fleet/integration_policy_ops/import_shared.rs

elasticctl_api/fleet/integration_policy_ops/
import_shared.rs
