From 26d1564d719c08c0d44e979672e66da28ad0a0cd Mon Sep 17 00:00:00 2001 From: Pascal Seitz Date: Mon, 19 Feb 2024 23:52:08 +0800 Subject: [PATCH] add support delete index in es API closes #3841 --- .../quickwit-index-management/src/index.rs | 107 +++++++++++++++++- .../src/elasticsearch_api/bulk.rs | 24 ++++ .../src/elasticsearch_api/filter.rs | 8 ++ .../src/elasticsearch_api/mod.rs | 34 +++++- .../src/elasticsearch_api/model/error.rs | 21 ++++ .../src/elasticsearch_api/rest_handler.rs | 42 ++++++- quickwit/quickwit-serve/src/rest.rs | 1 + 7 files changed, 225 insertions(+), 12 deletions(-) diff --git a/quickwit/quickwit-index-management/src/index.rs b/quickwit/quickwit-index-management/src/index.rs index 901c520b8de..fdd372bb10c 100644 --- a/quickwit/quickwit-index-management/src/index.rs +++ b/quickwit/quickwit-index-management/src/index.rs @@ -17,21 +17,26 @@ // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . +use std::collections::{HashMap, HashSet}; use std::path::Path; use std::time::Duration; +use futures_util::StreamExt; +use itertools::Itertools; use quickwit_common::fs::{empty_dir, get_cache_directory_path}; +use quickwit_common::PrettySample; use quickwit_config::{validate_identifier, IndexConfig, SourceConfig}; use quickwit_indexing::check_source_connectivity; use quickwit_metastore::{ AddSourceRequestExt, CreateIndexResponseExt, IndexMetadata, IndexMetadataResponseExt, - ListSplitsQuery, ListSplitsRequestExt, MetastoreServiceStreamSplitsExt, SplitInfo, - SplitMetadata, SplitState, + ListIndexesMetadataResponseExt, ListSplitsQuery, ListSplitsRequestExt, + MetastoreServiceStreamSplitsExt, SplitInfo, SplitMetadata, SplitState, }; use quickwit_proto::metastore::{ serde_utils, AddSourceRequest, CreateIndexRequest, DeleteIndexRequest, EntityKind, - IndexMetadataRequest, ListSplitsRequest, MarkSplitsForDeletionRequest, MetastoreError, - MetastoreService, MetastoreServiceClient, ResetSourceCheckpointRequest, + IndexMetadataRequest, ListIndexesMetadataRequest, ListSplitsRequest, + MarkSplitsForDeletionRequest, MetastoreError, MetastoreService, MetastoreServiceClient, + ResetSourceCheckpointRequest, }; use quickwit_proto::types::{IndexUid, SplitId}; use quickwit_proto::{ServiceError, ServiceErrorCode}; @@ -216,6 +221,100 @@ impl IndexService { Ok(deleted_splits) } + /// Deletes the indexes specified with `index_id_patterns`. + /// This is a wrapper of delete_index, and support index delete with index pattern + /// + /// * `index_id_patterns` - The targeted index ID patterns. + /// * `dry_run` - Should this only return a list of affected files without performing deletion. + pub async fn delete_indexes( + &self, + index_id_patterns: Vec, + dry_run: bool, + ) -> Result, IndexServiceError> { + let list_indexes_metadatas_request = ListIndexesMetadataRequest { + index_id_patterns: index_id_patterns.to_owned(), + }; + // disallow index_id patterns containing * + for index_id_pattern in &index_id_patterns { + if index_id_pattern.contains('*') { + return Err(IndexServiceError::Metastore( + MetastoreError::InvalidArgument { + message: format!("index_id pattern {} contains *", index_id_pattern), + }, + )); + } + } + + let mut metastore = self.metastore.clone(); + let indexes_metadata = metastore + .list_indexes_metadata(list_indexes_metadatas_request) + .await? + .deserialize_indexes_metadata()?; + + if indexes_metadata.len() != index_id_patterns.len() { + let found_index_ids: HashSet<&str> = indexes_metadata + .iter() + .map(|index_metadata| index_metadata.index_id()) + .collect(); + let missing_index_ids: Vec = index_id_patterns + .iter() + .filter(|index_id| !found_index_ids.contains(index_id.as_str())) + .map(|index_id| index_id.to_string()) + .collect_vec(); + return Err(IndexServiceError::Metastore(MetastoreError::NotFound( + EntityKind::Indexes { + index_ids: missing_index_ids.to_vec(), + }, + ))); + } + let index_ids = indexes_metadata + .iter() + .map(|index_metadata| index_metadata.index_id()) + .collect_vec(); + info!(index_ids = ?PrettySample::new(&index_ids, 5), "delete indexes"); + + // setup delete index tasks + let mut delete_index_tasks = Vec::new(); + for index_id in index_ids { + let task = async move { + let result = self.clone().delete_index(index_id, dry_run).await; + (index_id, result) + }; + delete_index_tasks.push(task); + } + let mut delete_responses: HashMap> = HashMap::new(); + let mut delete_errors: HashMap = HashMap::new(); + let mut stream = futures::stream::iter(delete_index_tasks).buffer_unordered(100); + while let Some((index_id, delete_response)) = stream.next().await { + match delete_response { + Ok(split_infos) => { + delete_responses.insert(index_id.to_string(), split_infos); + } + Err(error) => { + delete_errors.insert(index_id.to_string(), error); + } + } + } + + if delete_errors.is_empty() { + let mut concatenated_split_infos = Vec::new(); + for (_, split_info_vec) in delete_responses.into_iter() { + concatenated_split_infos.extend(split_info_vec); + } + Ok(concatenated_split_infos) + } else { + Err(IndexServiceError::Metastore(MetastoreError::Internal { + message: format!( + "errors occurred when deleting indexes: {:?}", + index_id_patterns + ), + cause: format!( + "errors: {:?}\ndeleted indexes: {:?}", + delete_errors, delete_responses + ), + })) + } + } /// Detect all dangling splits and associated files from the index and removes them. /// /// * `index_id` - The target index Id. diff --git a/quickwit/quickwit-serve/src/elasticsearch_api/bulk.rs b/quickwit/quickwit-serve/src/elasticsearch_api/bulk.rs index be81ea32ad9..38a0e4efd68 100644 --- a/quickwit/quickwit-serve/src/elasticsearch_api/bulk.rs +++ b/quickwit/quickwit-serve/src/elasticsearch_api/bulk.rs @@ -148,10 +148,13 @@ mod tests { use hyper::StatusCode; use quickwit_config::{IngestApiConfig, NodeConfig}; + use quickwit_index_management::IndexService; use quickwit_ingest::{FetchRequest, IngestServiceClient, SuggestTruncateRequest}; + use quickwit_metastore::metastore_for_test; use quickwit_proto::ingest::router::IngestRouterServiceClient; use quickwit_proto::metastore::MetastoreServiceClient; use quickwit_search::MockSearchService; + use quickwit_storage::StorageResolver; use crate::elasticsearch_api::bulk_v2::ElasticBulkResponse; use crate::elasticsearch_api::elastic_api_handlers; @@ -166,12 +169,15 @@ mod tests { let (universe, _temp_dir, ingest_service, _) = setup_ingest_service(&["my-index"], &IngestApiConfig::default()).await; let ingest_router = IngestRouterServiceClient::from(IngestRouterServiceClient::mock()); + let index_service = + IndexService::new(metastore_for_test(), StorageResolver::unconfigured()); let elastic_api_handlers = elastic_api_handlers( config, search_service, ingest_service, ingest_router, metastore_service.into(), + index_service, ); let payload = r#" { "create" : { "_index" : "my-index", "_id" : "1"} } @@ -196,12 +202,15 @@ mod tests { let (universe, _temp_dir, ingest_service, _) = setup_ingest_service(&["my-index-1", "my-index-2"], &IngestApiConfig::default()).await; let ingest_router = IngestRouterServiceClient::from(IngestRouterServiceClient::mock()); + let index_service = + IndexService::new(metastore_for_test(), StorageResolver::unconfigured()); let elastic_api_handlers = elastic_api_handlers( config, search_service, ingest_service, ingest_router, metastore_service.into(), + index_service, ); let payload = r#" { "create" : { "_index" : "my-index-1", "_id" : "1"} } @@ -230,12 +239,15 @@ mod tests { let (universe, _temp_dir, ingest_service, _) = setup_ingest_service(&["my-index-1"], &IngestApiConfig::default()).await; let ingest_router = IngestRouterServiceClient::from(IngestRouterServiceClient::mock()); + let index_service = + IndexService::new(metastore_for_test(), StorageResolver::unconfigured()); let elastic_api_handlers = elastic_api_handlers( config, search_service, ingest_service, ingest_router, metastore_service.into(), + index_service, ); let payload = " {\"create\": {\"_index\": \"my-index-1\", \"_id\": \"1674834324802805760\"}} @@ -261,12 +273,15 @@ mod tests { let (universe, _temp_dir, ingest_service, _) = setup_ingest_service(&["my-index-1", "my-index-2"], &IngestApiConfig::default()).await; let ingest_router = IngestRouterServiceClient::from(IngestRouterServiceClient::mock()); + let index_service = + IndexService::new(metastore_for_test(), StorageResolver::unconfigured()); let elastic_api_handlers = elastic_api_handlers( config, search_service, ingest_service, ingest_router, metastore_service.into(), + index_service, ); let payload = r#" { "create" : { "_index" : "my-index-1", "_id" : "1"} } @@ -295,12 +310,15 @@ mod tests { let (universe, _temp_dir, ingest_service, ingest_service_mailbox) = setup_ingest_service(&["my-index-1", "my-index-2"], &IngestApiConfig::default()).await; let ingest_router = IngestRouterServiceClient::from(IngestRouterServiceClient::mock()); + let index_service = + IndexService::new(metastore_for_test(), StorageResolver::unconfigured()); let elastic_api_handlers = elastic_api_handlers( config, search_service, ingest_service, ingest_router, metastore_service.into(), + index_service, ); let payload = r#" { "create" : { "_index" : "my-index-1", "_id" : "1"} } @@ -380,12 +398,15 @@ mod tests { let (universe, _temp_dir, ingest_service, ingest_service_mailbox) = setup_ingest_service(&["my-index-1", "my-index-2"], &IngestApiConfig::default()).await; let ingest_router = IngestRouterServiceClient::from(IngestRouterServiceClient::mock()); + let index_service = + IndexService::new(metastore_for_test(), StorageResolver::unconfigured()); let elastic_api_handlers = elastic_api_handlers( config, search_service, ingest_service, ingest_router, metastore_service.into(), + index_service, ); let payload = r#" { "create" : { "_index" : "my-index-1", "_id" : "1"} } @@ -463,12 +484,15 @@ mod tests { let metastore_service = MetastoreServiceClient::mock(); let ingest_service = IngestServiceClient::from(IngestServiceClient::mock()); let ingest_router = IngestRouterServiceClient::from(IngestRouterServiceClient::mock()); + let index_service = + IndexService::new(metastore_for_test(), StorageResolver::unconfigured()); let elastic_api_handlers = elastic_api_handlers( config, search_service, ingest_service, ingest_router, metastore_service.into(), + index_service, ); let payload = r#" {"create": {"_index": "my-index", "_id": "1"},} diff --git a/quickwit/quickwit-serve/src/elasticsearch_api/filter.rs b/quickwit/quickwit-serve/src/elasticsearch_api/filter.rs index 96ef7c3ae00..ca3f524d2fe 100644 --- a/quickwit/quickwit-serve/src/elasticsearch_api/filter.rs +++ b/quickwit/quickwit-serve/src/elasticsearch_api/filter.rs @@ -172,6 +172,14 @@ pub(crate) fn elastic_index_count_filter( .and(json_or_empty()) } +#[utoipa::path(delete, tag = "Indexes", path = "/{index}")] +pub(crate) fn elastic_delete_index_filter( +) -> impl Filter,), Error = Rejection> + Clone { + warp::path!("_elastic" / String) + .and_then(extract_index_id_patterns) + .and(warp::get()) +} + // No support for any query parameters for now. #[utoipa::path(get, tag = "Search", path = "/{index}/_stats")] pub(crate) fn elastic_index_stats_filter( diff --git a/quickwit/quickwit-serve/src/elasticsearch_api/mod.rs b/quickwit/quickwit-serve/src/elasticsearch_api/mod.rs index 02dbf03e1db..94c8bedd8fb 100644 --- a/quickwit/quickwit-serve/src/elasticsearch_api/mod.rs +++ b/quickwit/quickwit-serve/src/elasticsearch_api/mod.rs @@ -29,6 +29,7 @@ use bulk::{es_compat_bulk_handler, es_compat_index_bulk_handler}; pub use filter::ElasticCompatibleApi; use hyper::StatusCode; use quickwit_config::NodeConfig; +use quickwit_index_management::IndexService; use quickwit_ingest::IngestServiceClient; use quickwit_proto::ingest::router::IngestRouterServiceClient; use quickwit_proto::metastore::MetastoreServiceClient; @@ -41,9 +42,10 @@ use serde::{Deserialize, Serialize}; use warp::{Filter, Rejection}; use self::rest_handler::{ - es_compat_cat_indices_handler, es_compat_index_cat_indices_handler, - es_compat_index_count_handler, es_compat_index_field_capabilities_handler, - es_compat_index_stats_handler, es_compat_stats_handler, + es_compat_cat_indices_handler, es_compat_delete_index_handler, + es_compat_index_cat_indices_handler, es_compat_index_count_handler, + es_compat_index_field_capabilities_handler, es_compat_index_stats_handler, + es_compat_stats_handler, }; use crate::elasticsearch_api::model::ElasticsearchError; use crate::rest_api_response::RestApiResponse; @@ -59,6 +61,7 @@ pub fn elastic_api_handlers( ingest_service: IngestServiceClient, ingest_router: IngestRouterServiceClient, metastore: MetastoreServiceClient, + index_service: IndexService, ) -> impl Filter + Clone { es_compat_cluster_info_handler(node_config, BuildInfo::get()) .or(es_compat_search_handler(search_service.clone())) @@ -75,6 +78,7 @@ pub fn elastic_api_handlers( )) .or(es_compat_index_bulk_handler(ingest_service, ingest_router)) .or(es_compat_index_stats_handler(metastore.clone())) + .or(es_compat_delete_index_handler(index_service)) .or(es_compat_stats_handler(metastore.clone())) .or(es_compat_index_cat_indices_handler(metastore.clone())) .or(es_compat_cat_indices_handler(metastore.clone())) @@ -128,10 +132,13 @@ mod tests { use assert_json_diff::assert_json_include; use mockall::predicate; use quickwit_config::NodeConfig; + use quickwit_index_management::IndexService; use quickwit_ingest::{IngestApiService, IngestServiceClient}; + use quickwit_metastore::metastore_for_test; use quickwit_proto::ingest::router::IngestRouterServiceClient; use quickwit_proto::metastore::MetastoreServiceClient; use quickwit_search::MockSearchService; + use quickwit_storage::StorageResolver; use serde_json::Value as JsonValue; use warp::Filter; @@ -166,12 +173,15 @@ mod tests { )) .returning(|_| Ok(Default::default())); let ingest_router = IngestRouterServiceClient::from(IngestRouterServiceClient::mock()); + let index_service = + IndexService::new(metastore_for_test(), StorageResolver::unconfigured()); let es_search_api_handler = super::elastic_api_handlers( config, Arc::new(mock_search_service), ingest_service_client(), ingest_router, MetastoreServiceClient::mock().into(), + index_service, ); let msearch_payload = r#" {"index":"index-1"} @@ -217,12 +227,15 @@ mod tests { }); let ingest_router = IngestRouterServiceClient::from(IngestRouterServiceClient::mock()); + let index_service = + IndexService::new(metastore_for_test(), StorageResolver::unconfigured()); let es_search_api_handler = super::elastic_api_handlers( config, Arc::new(mock_search_service), ingest_service_client(), ingest_router, MetastoreServiceClient::mock().into(), + index_service, ); let msearch_payload = r#" {"index":"index-1"} @@ -256,12 +269,15 @@ mod tests { let mock_search_service = MockSearchService::new(); let ingest_router = IngestRouterServiceClient::from(IngestRouterServiceClient::mock()); + let index_service = + IndexService::new(metastore_for_test(), StorageResolver::unconfigured()); let es_search_api_handler = super::elastic_api_handlers( config, Arc::new(mock_search_service), ingest_service_client(), ingest_router, MetastoreServiceClient::mock().into(), + index_service, ); let msearch_payload = r#" {"index":"index-1" @@ -288,12 +304,15 @@ mod tests { let mock_search_service = MockSearchService::new(); let ingest_router = IngestRouterServiceClient::from(IngestRouterServiceClient::mock()); + let index_service = + IndexService::new(metastore_for_test(), StorageResolver::unconfigured()); let es_search_api_handler = elastic_api_handlers( config, Arc::new(mock_search_service), ingest_service_client(), ingest_router, MetastoreServiceClient::mock().into(), + index_service, ); let msearch_payload = r#" {"index":"index-1"} @@ -320,12 +339,15 @@ mod tests { let mock_search_service = MockSearchService::new(); let ingest_router = IngestRouterServiceClient::from(IngestRouterServiceClient::mock()); + let index_service = + IndexService::new(metastore_for_test(), StorageResolver::unconfigured()); let es_search_api_handler = super::elastic_api_handlers( config, Arc::new(mock_search_service), ingest_service_client(), ingest_router, MetastoreServiceClient::mock().into(), + index_service, ); let msearch_payload = r#" {"index":"index-1"} @@ -351,12 +373,15 @@ mod tests { let mock_search_service = MockSearchService::new(); let ingest_router = IngestRouterServiceClient::from(IngestRouterServiceClient::mock()); + let index_service = + IndexService::new(metastore_for_test(), StorageResolver::unconfigured()); let es_search_api_handler = super::elastic_api_handlers( config, Arc::new(mock_search_service), ingest_service_client(), ingest_router, MetastoreServiceClient::mock().into(), + index_service, ); let msearch_payload = r#" {} @@ -394,12 +419,15 @@ mod tests { } }); let ingest_router = IngestRouterServiceClient::from(IngestRouterServiceClient::mock()); + let index_service = + IndexService::new(metastore_for_test(), StorageResolver::unconfigured()); let es_search_api_handler = super::elastic_api_handlers( config, Arc::new(mock_search_service), ingest_service_client(), ingest_router, MetastoreServiceClient::mock().into(), + index_service, ); let msearch_payload = r#" {"index": ["index-1", "index-2"]} diff --git a/quickwit/quickwit-serve/src/elasticsearch_api/model/error.rs b/quickwit/quickwit-serve/src/elasticsearch_api/model/error.rs index 674800a8509..f6571e9ec70 100644 --- a/quickwit/quickwit-serve/src/elasticsearch_api/model/error.rs +++ b/quickwit/quickwit-serve/src/elasticsearch_api/model/error.rs @@ -19,6 +19,7 @@ use elasticsearch_dsl::search::ErrorCause; use hyper::StatusCode; +use quickwit_index_management::IndexServiceError; use quickwit_ingest::IngestServiceError; use quickwit_proto::ingest::IngestV2Error; use quickwit_proto::ServiceError; @@ -108,3 +109,23 @@ impl From for ElasticsearchError { } } } + +impl From for ElasticsearchError { + fn from(ingest_error: IndexServiceError) -> Self { + let status = ingest_error.error_code().to_http_status_code(); + + let reason = ErrorCause { + reason: Some(ingest_error.to_string()), + caused_by: None, + root_cause: Vec::new(), + stack_trace: None, + suppressed: Vec::new(), + ty: None, + additional_details: Default::default(), + }; + ElasticsearchError { + status, + error: reason, + } + } +} diff --git a/quickwit/quickwit-serve/src/elasticsearch_api/rest_handler.rs b/quickwit/quickwit-serve/src/elasticsearch_api/rest_handler.rs index ed9baf5050b..85d39123253 100644 --- a/quickwit/quickwit-serve/src/elasticsearch_api/rest_handler.rs +++ b/quickwit/quickwit-serve/src/elasticsearch_api/rest_handler.rs @@ -30,6 +30,7 @@ use hyper::StatusCode; use itertools::Itertools; use quickwit_common::truncate_str; use quickwit_config::{validate_index_id_pattern, NodeConfig}; +use quickwit_index_management::IndexService; use quickwit_metastore::*; use quickwit_proto::metastore::MetastoreServiceClient; use quickwit_proto::search::{ @@ -46,11 +47,11 @@ use serde_json::json; use warp::{Filter, Rejection}; use super::filter::{ - elastic_cat_indices_filter, elastic_cluster_info_filter, elastic_field_capabilities_filter, - elastic_index_cat_indices_filter, elastic_index_count_filter, - elastic_index_field_capabilities_filter, elastic_index_search_filter, - elastic_index_stats_filter, elastic_multi_search_filter, elastic_scroll_filter, - elastic_stats_filter, elasticsearch_filter, + elastic_cat_indices_filter, elastic_cluster_info_filter, elastic_delete_index_filter, + elastic_field_capabilities_filter, elastic_index_cat_indices_filter, + elastic_index_count_filter, elastic_index_field_capabilities_filter, + elastic_index_search_filter, elastic_index_stats_filter, elastic_multi_search_filter, + elastic_scroll_filter, elastic_stats_filter, elasticsearch_filter, }; use super::model::{ build_list_field_request_for_es_api, convert_to_es_field_capabilities_response, @@ -117,6 +118,16 @@ pub fn es_compat_index_field_capabilities_handler( .map(|result| make_elastic_api_response(result, BodyFormat::default())) } +/// DELETE _elastic/{index} +pub fn es_compat_delete_index_handler( + index_service: IndexService, +) -> impl Filter + Clone { + elastic_delete_index_filter() + .and(with_arg(index_service)) + .then(es_compat_delete_index) + .map(|result| make_elastic_api_response(result, BodyFormat::default())) +} + /// GET _elastic/_stats pub fn es_compat_stats_handler( search_service: MetastoreServiceClient, @@ -126,6 +137,7 @@ pub fn es_compat_stats_handler( .then(es_compat_stats) .map(|result| make_elastic_api_response(result, BodyFormat::default())) } + /// GET _elastic/{index}/_stats pub fn es_compat_index_stats_handler( search_service: MetastoreServiceClient, @@ -410,6 +422,26 @@ async fn es_compat_index_search( Ok(search_response_rest) } +/// Returns JSON in the format: +/// +/// { +/// "acknowledged": true +/// } +#[derive(Clone, Serialize, Deserialize, Debug)] +pub struct ElasticsearchDeleteResponse { + pub acknowledged: bool, +} + +async fn es_compat_delete_index( + index_id_patterns: Vec, + index_service: IndexService, +) -> Result { + index_service + .delete_indexes(index_id_patterns, false) + .await?; + Ok(ElasticsearchDeleteResponse { acknowledged: true }) +} + async fn es_compat_stats( metastore: MetastoreServiceClient, ) -> Result { diff --git a/quickwit/quickwit-serve/src/rest.rs b/quickwit/quickwit-serve/src/rest.rs index d80c87377eb..71036164953 100644 --- a/quickwit/quickwit-serve/src/rest.rs +++ b/quickwit/quickwit-serve/src/rest.rs @@ -207,6 +207,7 @@ fn api_v1_routes( quickwit_services.ingest_service.clone(), quickwit_services.ingest_router_service.clone(), quickwit_services.metastore_client.clone(), + quickwit_services.index_manager.clone(), )) .or(index_template_api_handlers( quickwit_services.metastore_client.clone(),