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

elasticctl_api/fleet/integration_policy_ops/
import_shared.rs
