elasticctl_api/dashboards_ops/
import.rs

1//! Dashboard artifact import planning and apply.
2
3use super::{
4    BTreeMap, BTreeSet, DashboardSpec, Error, ErrorKind, Path, Result, Serialize, Transport, Value,
5    dashboards, data_views, invalid_plan, validate,
6};
7
8/// The immutable, guard-ready import work computed from one portable
9/// dashboard artifact.
10#[derive(Debug, Clone, PartialEq)]
11pub struct DashboardImportPlan {
12    pub preview: crate::ops::MutationPlan,
13    pub specs: Vec<DashboardSpec>,
14    pub before: BTreeMap<String, Option<DashboardSpec>>,
15    pub skipped: Vec<Value>,
16    pub total: usize,
17    pub overwrite: bool,
18}
19
20/// The per-object result of applying a guarded dashboard import.
21///
22/// Field order is rendered JSON order and is contractual.
23#[derive(Debug, Clone, PartialEq, Serialize)]
24pub struct DashboardImportReport {
25    pub applied: bool,
26    pub succeeded: Vec<Value>,
27    pub skipped: Vec<Value>,
28    pub failed: Vec<Value>,
29    pub lossy: Vec<Value>,
30    pub total: usize,
31}
32/// Fully validate an artifact and collect the precise import work to show in
33/// the mutation guard. The supplied transport is optional only for a local
34/// no-conflict plan with no data-view references.
35pub async fn plan_import(
36    transport: Option<&Transport>,
37    path: &Path,
38    overwrite: bool,
39    skip_existing: bool,
40) -> Result<DashboardImportPlan> {
41    let mut specs = validate(path)?;
42    if specs.is_empty() {
43        return Err(Error::new(
44            ErrorKind::Error,
45            "dashboard import needs at least one dashboard",
46        ));
47    }
48    if overwrite && skip_existing {
49        return Err(Error::new(
50            ErrorKind::Error,
51            "--overwrite and --skip-existing cannot be used together",
52        ));
53    }
54
55    let references: BTreeSet<_> = specs
56        .iter()
57        .flat_map(|spec| dashboards::collect_data_view_refs(&Value::Object(spec.data.clone())))
58        .collect();
59    let requires_server =
60        overwrite || skip_existing || transport.is_some() || !references.is_empty();
61    let transport = if requires_server { transport } else { None };
62    if (overwrite || skip_existing || !references.is_empty()) && transport.is_none() {
63        return Err(Error::new(
64            ErrorKind::Error,
65            "dashboard import preflight needs a transport",
66        ));
67    }
68
69    let total = specs.len();
70    let mut before = BTreeMap::new();
71    let mut conflicts = Vec::new();
72    if let Some(transport) = transport {
73        for spec in &specs {
74            match read_spec(transport, &spec.id).await {
75                Ok(current) => {
76                    if !overwrite && !skip_existing {
77                        conflicts.push(spec.id.clone());
78                    }
79                    before.insert(spec.id.clone(), Some(current));
80                }
81                Err(error) if error.kind == ErrorKind::NotFound => {
82                    before.insert(spec.id.clone(), None);
83                }
84                Err(error) => return Err(error),
85            }
86        }
87
88        let mut missing = Vec::new();
89        for id in references {
90            match data_views::get(transport, &id).await {
91                Ok(data_view) => match data_view.data_view.get("id").and_then(Value::as_str) {
92                    Some(actual) if actual == id => {}
93                    Some(actual) => {
94                        return Err(Error::new(
95                            ErrorKind::Http,
96                            format!("decoding data view: expected id '{id}', got '{actual}'"),
97                        ));
98                    }
99                    None => {
100                        return Err(Error::new(
101                            ErrorKind::Http,
102                            format!(
103                                "decoding data view: expected id '{id}', got missing or non-string id"
104                            ),
105                        ));
106                    }
107                },
108                Err(error) if error.kind == ErrorKind::NotFound => missing.push(id),
109                Err(error) => return Err(error),
110            }
111        }
112        if !missing.is_empty() {
113            return Err(Error::new(
114                ErrorKind::NotFound,
115                format!("referenced data views do not exist: {}", missing.join(", ")),
116            ));
117        }
118    } else {
119        before.extend(specs.iter().map(|spec| (spec.id.clone(), None)));
120    }
121
122    if !conflicts.is_empty() {
123        return Err(Error::new(
124            ErrorKind::Conflict,
125            format!("dashboards already exist: {}", conflicts.join(", ")),
126        ));
127    }
128
129    let mut skipped = Vec::new();
130    if skip_existing {
131        specs.retain(|spec| match before.get(&spec.id) {
132            Some(Some(_)) => {
133                skipped.push(serde_json::json!({"id": spec.id, "reason": "exists"}));
134                false
135            }
136            _ => true,
137        });
138        before.retain(|id, _| specs.iter().any(|spec| spec.id == *id));
139    }
140
141    let preview = crate::ops::MutationPlan {
142        preview_action: format!("Import {} dashboard(s)", specs.len()),
143        preview_details: import_preview_details(&specs, &before),
144        targets: specs.iter().map(|spec| spec.id.clone()).collect(),
145    };
146    Ok(DashboardImportPlan {
147        preview,
148        specs,
149        before,
150        skipped,
151        total,
152        overwrite,
153    })
154}
155
156/// Apply a guard-approved dashboard import without rereading its source file.
157///
158/// The final dashboard GET immediately precedes each possible PUT. Kibana has
159/// no conditional-write token, so the interval after that read is the smallest
160/// unavoidable race window. Independent objects continue after failures, and
161/// successful earlier writes are never rolled back.
162pub async fn apply_import(
163    transport: &Transport,
164    plan: &DashboardImportPlan,
165) -> Result<DashboardImportReport> {
166    validate_import_plan(plan)?;
167    let mut succeeded = Vec::new();
168    let mut failed = Vec::new();
169    let mut lossy = Vec::new();
170
171    for desired in &plan.specs {
172        let Some(before) = plan.before.get(&desired.id) else {
173            failed.push(failed_row(&desired.id, false, "missing preflight snapshot"));
174            continue;
175        };
176        let current = match read_spec(transport, &desired.id).await {
177            Ok(current) => Some(current),
178            Err(error) if error.kind == ErrorKind::NotFound => None,
179            Err(error) => {
180                failed.push(failed_row(&desired.id, false, error.message));
181                continue;
182            }
183        };
184
185        match (before, current) {
186            (None, Some(_)) => {
187                failed.push(failed_row(
188                    &desired.id,
189                    false,
190                    "dashboard appeared since preview",
191                ));
192            }
193            (Some(_), None) => {
194                failed.push(failed_row(
195                    &desired.id,
196                    false,
197                    "dashboard disappeared since preview",
198                ));
199            }
200            (Some(before), Some(current)) if before != &current => {
201                failed.push(failed_row(
202                    &desired.id,
203                    false,
204                    "dashboard changed since preview",
205                ));
206            }
207            (Some(_), Some(current)) if current == *desired => {
208                succeeded.push(serde_json::json!({"id": desired.id, "action": "unchanged"}));
209            }
210            (None, None) => {
211                apply_put(
212                    transport,
213                    desired,
214                    "created",
215                    &mut succeeded,
216                    &mut failed,
217                    &mut lossy,
218                )
219                .await;
220            }
221            (Some(_), Some(_)) => {
222                apply_put(
223                    transport,
224                    desired,
225                    "replaced",
226                    &mut succeeded,
227                    &mut failed,
228                    &mut lossy,
229                )
230                .await;
231            }
232        }
233    }
234
235    Ok(DashboardImportReport {
236        applied: true,
237        succeeded,
238        skipped: plan.skipped.clone(),
239        failed,
240        lossy,
241        total: plan.total,
242    })
243}
244async fn apply_put(
245    transport: &Transport,
246    desired: &DashboardSpec,
247    action: &str,
248    succeeded: &mut Vec<Value>,
249    failed: &mut Vec<Value>,
250    lossy: &mut Vec<Value>,
251) {
252    let response = match dashboards::put(transport, desired).await {
253        Ok(response) => response,
254        Err(error) => {
255            failed.push(failed_row(&desired.id, false, error.message));
256            return;
257        }
258    };
259    if response.id != desired.id {
260        failed.push(failed_row(
261            &desired.id,
262            true,
263            format!(
264                "dashboard PUT returned id '{}' instead of '{}'",
265                response.id, desired.id
266            ),
267        ));
268        return;
269    }
270
271    let mut paths: Vec<_> = dashboards::subset_losses(
272        &Value::Object(desired.data.clone()),
273        &Value::Object(response.data),
274    )
275    .into_iter()
276    .map(|loss| loss.path)
277    .collect();
278    if paths.is_empty() {
279        succeeded.push(serde_json::json!({"id": desired.id, "action": action}));
280        return;
281    }
282    paths.sort();
283    let warnings = match dashboards::get(transport, &desired.id).await {
284        Ok(dashboard) if dashboard.id == desired.id => dashboard
285            .warnings
286            .into_iter()
287            .map(|warning| warning.message)
288            .collect(),
289        Ok(dashboard) => {
290            failed.push(failed_row(
291                &desired.id,
292                true,
293                format!(
294                    "dashboard loss audit returned id '{}' instead of '{}'",
295                    dashboard.id, desired.id
296                ),
297            ));
298            Vec::new()
299        }
300        Err(error) => {
301            failed.push(failed_row(
302                &desired.id,
303                true,
304                format!("dashboard loss audit failed: {}", error.message),
305            ));
306            Vec::new()
307        }
308    };
309    lossy.push(serde_json::json!({
310        "id": desired.id,
311        "applied": true,
312        "paths": paths,
313        "warnings": warnings,
314    }));
315}
316
317fn validate_import_plan(plan: &DashboardImportPlan) -> Result<()> {
318    if plan.total == 0 {
319        return invalid_plan("total must be greater than zero");
320    }
321    if plan.total != plan.specs.len() + plan.skipped.len() {
322        return invalid_plan("total does not equal pending and skipped dashboards");
323    }
324    if !plan.skipped.is_empty() && plan.overwrite {
325        return invalid_plan("skipped dashboards require skip-existing mode");
326    }
327
328    let mut ids = Vec::with_capacity(plan.specs.len());
329    for spec in &plan.specs {
330        dashboards::validate_spec(spec)?;
331        if ids
332            .last()
333            .is_some_and(|previous: &String| previous >= &spec.id)
334        {
335            return invalid_plan("pending dashboards must be unique and sorted by id");
336        }
337        ids.push(spec.id.clone());
338    }
339    let pending: BTreeSet<_> = ids.iter().cloned().collect();
340    if plan.preview.targets != ids {
341        return invalid_plan("preview targets do not match pending dashboards");
342    }
343    if plan.preview.preview_action != format!("Import {} dashboard(s)", plan.specs.len()) {
344        return invalid_plan("preview action does not match pending dashboards");
345    }
346    if plan.preview.preview_details != import_preview_details(&plan.specs, &plan.before) {
347        return invalid_plan("preview details do not match pending dashboards");
348    }
349
350    if plan.before.keys().cloned().collect::<BTreeSet<_>>() != pending {
351        return invalid_plan("preflight snapshots do not match pending dashboards");
352    }
353    for spec in &plan.specs {
354        match plan.before.get(&spec.id).expect("checked key set") {
355            None => {}
356            Some(snapshot) => {
357                dashboards::validate_spec(snapshot)?;
358                if snapshot.id != spec.id {
359                    return invalid_plan("preflight snapshot id does not match its target");
360                }
361                if !plan.overwrite {
362                    return invalid_plan("replacement plan requires overwrite");
363                }
364            }
365        }
366    }
367
368    let mut skipped_ids = BTreeSet::new();
369    let mut previous = None;
370    for row in &plan.skipped {
371        let object = row
372            .as_object()
373            .filter(|object| object.len() == 2)
374            .ok_or_else(|| Error::new(ErrorKind::Error, "invalid dashboard import skipped row"))?;
375        if !object.keys().map(String::as_str).eq(["id", "reason"]) {
376            return invalid_plan("invalid dashboard import skipped row order");
377        }
378        let id = object
379            .get("id")
380            .and_then(Value::as_str)
381            .filter(|id| !id.trim().is_empty())
382            .ok_or_else(|| Error::new(ErrorKind::Error, "invalid dashboard import skipped row"))?;
383        if object.get("reason").and_then(Value::as_str) != Some("exists")
384            || previous.is_some_and(|previous: &str| previous >= id)
385            || !skipped_ids.insert(id.to_owned())
386            || pending.contains(id)
387        {
388            return invalid_plan("invalid dashboard import skipped rows");
389        }
390        previous = Some(id);
391    }
392    Ok(())
393}
394
395fn failed_row(id: &str, applied: bool, error: impl Into<String>) -> Value {
396    serde_json::json!({"id": id, "applied": applied, "error": error.into()})
397}
398/// Read one live dashboard and rebuild the portable spec used for snapshot
399/// comparisons. The response id is checked separately from its request path.
400async fn read_spec(transport: &Transport, id: &str) -> Result<DashboardSpec> {
401    let dashboard = dashboards::get(transport, id).await?;
402    if dashboard.id != id {
403        return Err(Error::new(
404            ErrorKind::Http,
405            format!(
406                "decoding dashboard: expected id '{id}', got '{}'",
407                dashboard.id
408            ),
409        ));
410    }
411    let spec = DashboardSpec {
412        id: dashboard.id,
413        data: dashboard.data,
414    };
415    dashboards::validate_spec(&spec)?;
416    Ok(spec)
417}
418
419fn import_preview_details(
420    specs: &[DashboardSpec],
421    before: &BTreeMap<String, Option<DashboardSpec>>,
422) -> Vec<String> {
423    specs
424        .iter()
425        .filter_map(|spec| match before.get(&spec.id) {
426            Some(None) => Some(format!("{}  create  {}", spec.id, dashboard_title(spec))),
427            Some(Some(current)) if current == spec => {
428                Some(format!("{}  no-op  {}", spec.id, dashboard_title(spec)))
429            }
430            Some(Some(current)) => Some(format!(
431                "{}  replace  {} -> {}",
432                spec.id,
433                dashboard_title(current),
434                dashboard_title(spec)
435            )),
436            None => None,
437        })
438        .collect()
439}
440
441fn dashboard_title(spec: &DashboardSpec) -> &str {
442    spec.data
443        .get("title")
444        .and_then(Value::as_str)
445        .expect("validated dashboard specs have a title")
446}
elasticctl_api/dashboards_ops/import.rs

elasticctl_api/dashboards_ops/
import.rs
