kibana_sync/kibana/saved_objects/
loader.rs

1//! Saved Objects API loader
2//!
3//! Loads saved objects to Kibana via POST /api/saved_objects/_import
4
5use crate::client::KibanaClient;
6use crate::etl::Loader;
7
8use crate::{Error, Result, ResultContext};
9use serde::Deserialize;
10use serde_json::Value;
11
12const LEGACY_JSON_FIELDS: &[&str] = &[
13    "attributes.panelsJSON",
14    "attributes.fieldFormatMap",
15    "attributes.controlGroupInput.ignoreParentSettingsJSON",
16    "attributes.controlGroupInput.panelsJSON",
17    "attributes.kibanaSavedObjectMeta.searchSourceJSON",
18    "attributes.optionsJSON",
19    "attributes.visState",
20    "attributes.fieldAttrs",
21];
22
23#[derive(Debug, Deserialize)]
24#[serde(rename_all = "camelCase")]
25struct ImportResponse {
26    #[serde(default)]
27    success: bool,
28    #[serde(default)]
29    errors: Vec<Value>,
30}
31
32/// Loader for Kibana saved objects
33///
34/// Imports saved objects into Kibana using the import API.
35/// Objects are sent as NDJSON in a multipart form.
36///
37/// # Example
38/// ```no_run
39/// use kibana_sync::kibana::saved_objects::SavedObjectsLoader;
40/// use kibana_sync::client::{Auth, KibanaClient};
41/// use kibana_sync::etl::Loader;
42/// use serde_json::json;
43/// use url::Url;
44///
45/// # async fn example() -> kibana_sync::Result<()> {
46/// let url = Url::parse("http://localhost:5601")?;
47/// let client = KibanaClient::new(url, Auth::None)?;
48/// let space_client = client.space("default")?;
49/// let loader = SavedObjectsLoader::new(space_client);
50///
51/// let objects = vec![
52///     json!({
53///         "type": "dashboard",
54///         "id": "my-dashboard",
55///         "attributes": {"title": "My Dashboard"}
56///     })
57/// ];
58///
59/// let count = loader.load(objects).await?;
60/// # Ok(())
61/// # }
62/// ```
63pub struct SavedObjectsLoader {
64    client: KibanaClient,
65    overwrite: bool,
66}
67
68impl SavedObjectsLoader {
69    /// Create a new saved objects loader
70    ///
71    /// # Arguments
72    /// * `client` - Space-scoped Kibana client
73    pub fn new(client: KibanaClient) -> Self {
74        Self {
75            client,
76            overwrite: true,
77        }
78    }
79
80    /// Set whether to overwrite existing objects (default: true)
81    pub fn with_overwrite(mut self, overwrite: bool) -> Self {
82        self.overwrite = overwrite;
83        self
84    }
85
86    /// Import saved objects into Kibana
87    ///
88    /// Converts objects to NDJSON and uploads via multipart form.
89    async fn import_objects(&self, objects: &[Value]) -> Result<()> {
90        // Kibana requires legacy structured fields to be serialized JSON strings.
91        let ndjson = objects
92            .iter()
93            .cloned()
94            .map(serialize_legacy_json_fields)
95            .collect::<Result<Vec<_>>>()?
96            .iter()
97            .map(serde_json::to_string)
98            .collect::<std::result::Result<Vec<_>, serde_json::Error>>()
99            .with_context(|| "Failed to serialize objects to NDJSON")?
100            .join("\n");
101
102        let path = format!("api/saved_objects/_import?overwrite={}", self.overwrite);
103
104        tracing::debug!(
105            "Importing {} object(s) to space '{}'",
106            objects.len(),
107            self.client.space_id()
108        );
109
110        let response = self
111            .client
112            .post_form(&path, ndjson.as_bytes())
113            .await
114            .with_context(|| "Failed to import saved objects to Kibana")?;
115
116        let status = response.status();
117        if !status.is_success() {
118            let body = response.text().await.unwrap_or_default();
119            return Err(Error::api_response(status, body));
120        }
121
122        let body = response
123            .text()
124            .await
125            .with_context(|| "Failed to read saved object import response")?;
126        let result: ImportResponse = serde_json::from_str(&body)
127            .with_context(|| "Failed to parse saved object import response")?;
128        if !result.success || !result.errors.is_empty() {
129            return Err(Error::api_response(status, body));
130        }
131
132        tracing::info!("Imported {} object(s) to Kibana", objects.len());
133
134        Ok(())
135    }
136}
137
138fn serialize_legacy_json_fields(mut object: Value) -> Result<Value> {
139    for field_path in LEGACY_JSON_FIELDS {
140        if let Some(field) = get_nested_mut(&mut object, field_path)
141            && (field.is_object() || field.is_array())
142        {
143            *field = Value::String(
144                serde_json::to_string(field)
145                    .with_context(|| format!("Failed to serialize legacy field: {field_path}"))?,
146            );
147        }
148    }
149    Ok(object)
150}
151
152fn get_nested_mut<'a>(object: &'a mut Value, path: &str) -> Option<&'a mut Value> {
153    let mut current = object;
154    for field in path.split('.') {
155        current = current.get_mut(field)?;
156    }
157    Some(current)
158}
159
160impl Loader for SavedObjectsLoader {
161    type Item = Value;
162
163    async fn load(&self, items: Vec<Self::Item>) -> Result<usize> {
164        if items.is_empty() {
165            tracing::info!("No saved objects to import");
166            return Ok(0);
167        }
168
169        self.import_objects(&items).await?;
170
171        tracing::info!("Loaded {} saved object(s) to Kibana", items.len());
172        Ok(items.len())
173    }
174}
175
176#[cfg(test)]
177mod tests {
178    use super::*;
179    use crate::client::{Auth, KibanaClient};
180    use crate::etl::Loader;
181    use crate::test_support::{MockResponse, TestServer};
182    use serde_json::json;
183    use url::Url;
184
185    #[test]
186    fn test_loader_creation() {
187        let url = Url::parse("http://localhost:5601").unwrap();
188        let client = KibanaClient::new(url, Auth::None).unwrap();
189        let space_client = client.space("default").unwrap();
190        let loader = SavedObjectsLoader::new(space_client);
191        assert!(loader.overwrite);
192    }
193
194    #[test]
195    fn test_with_overwrite() {
196        let url = Url::parse("http://localhost:5601").unwrap();
197        let client = KibanaClient::new(url, Auth::None).unwrap();
198        let space_client = client.space("default").unwrap();
199        let loader = SavedObjectsLoader::new(space_client).with_overwrite(false);
200        assert!(!loader.overwrite);
201    }
202
203    #[test]
204    fn test_custom_space() {
205        let url = Url::parse("http://localhost:5601").unwrap();
206        let client = KibanaClient::builder(url)
207            .spaces([
208                ("default".to_string(), "Default".to_string()),
209                ("marketing".to_string(), "Marketing".to_string()),
210            ])
211            .build()
212            .unwrap();
213        let space_client = client.space("marketing").unwrap();
214        let loader = SavedObjectsLoader::new(space_client);
215        assert_eq!(loader.client.space_id(), "marketing");
216    }
217
218    #[test]
219    fn serializes_structured_legacy_fields_without_reescaping_strings() {
220        let object = json!({
221            "attributes": {
222                "panelsJSON": [{"panelIndex": "1"}],
223                "kibanaSavedObjectMeta": {
224                    "searchSourceJSON": {"query": {"language": "kuery"}}
225                },
226                "visState": {"type": "pie"},
227                "optionsJSON": "{\"useMargins\":true}"
228            }
229        });
230
231        let serialized = serialize_legacy_json_fields(object).unwrap();
232        let attributes = &serialized["attributes"];
233        assert!(attributes["panelsJSON"].is_string());
234        assert!(attributes["kibanaSavedObjectMeta"]["searchSourceJSON"].is_string());
235        assert!(attributes["visState"].is_string());
236        assert_eq!(attributes["optionsJSON"], "{\"useMargins\":true}");
237        assert_eq!(
238            serde_json::from_str::<Value>(attributes["panelsJSON"].as_str().unwrap()).unwrap(),
239            json!([{"panelIndex": "1"}])
240        );
241    }
242
243    #[tokio::test]
244    async fn import_serializes_legacy_fields_in_ndjson() {
245        let server = TestServer::new(vec![MockResponse {
246            method: "POST",
247            path: "/api/saved_objects/_import?overwrite=true",
248            status: 200,
249            body: json!({"success": true, "successCount": 1, "errors": []}),
250        }]);
251        let loader = SavedObjectsLoader::new(server.client().unwrap().space("default").unwrap());
252
253        loader
254            .load(vec![json!({
255                "type": "dashboard",
256                "id": "dashboard-1",
257                "attributes": {
258                    "panelsJSON": [{"panelIndex": "1"}],
259                    "kibanaSavedObjectMeta": {
260                        "searchSourceJSON": {"query": {"language": "kuery"}}
261                    },
262                    "visState": {"type": "pie"}
263                }
264            })])
265            .await
266            .unwrap();
267
268        let request = &server.requests()[0];
269        let (_, ndjson) = request.body.split_once("\r\n\r\n").unwrap();
270        let (ndjson, _) = ndjson.split_once("\r\n--").unwrap();
271        let imported: Value = serde_json::from_str(ndjson).unwrap();
272        assert!(imported["attributes"]["panelsJSON"].is_string());
273        assert!(imported["attributes"]["kibanaSavedObjectMeta"]["searchSourceJSON"].is_string());
274        assert!(imported["attributes"]["visState"].is_string());
275    }
276
277    #[tokio::test]
278    async fn import_fails_when_kibana_reports_item_errors() {
279        let server = TestServer::new(vec![MockResponse {
280            method: "POST",
281            path: "/api/saved_objects/_import?overwrite=true",
282            status: 200,
283            body: json!({
284                "success": true,
285                "successCount": 1,
286                "errors": [{"id": "dashboard-1", "type": "dashboard", "error": {"message": "invalid panelsJSON"}}]
287            }),
288        }]);
289        let loader = SavedObjectsLoader::new(server.client().unwrap().space("default").unwrap());
290
291        let error = loader
292            .load(vec![json!({"type": "dashboard", "id": "dashboard-1"})])
293            .await
294            .unwrap_err();
295
296        assert!(error.to_string().contains("invalid panelsJSON"));
297    }
298
299    #[tokio::test]
300    async fn import_fails_when_kibana_reports_unsuccessful_response() {
301        let server = TestServer::new(vec![MockResponse {
302            method: "POST",
303            path: "/api/saved_objects/_import?overwrite=true",
304            status: 200,
305            body: json!({"success": false, "successCount": 0, "errors": []}),
306        }]);
307        let loader = SavedObjectsLoader::new(server.client().unwrap().space("default").unwrap());
308
309        let error = loader
310            .load(vec![json!({"type": "dashboard", "id": "dashboard-1"})])
311            .await
312            .unwrap_err();
313
314        assert!(error.to_string().contains("\"success\":false"));
315    }
316}
kibana_sync/kibana/saved_objects/loader.rs

kibana_sync/kibana/saved_objects/
loader.rs
