refactor: extract ES-compat handler logic into reusable pub(crate) functions (#6271)

Extract JSON construction from warp handlers into standalone functions
that can be called directly without HTTP, enabling non-HTTP consumers
(e.g., gRPC passthrough) to reuse the ES-compatible logic.

Extracted functions:
- es_compat_cluster_info, es_compat_nodes_info, es_compat_search_shards,
  es_compat_aliases, es_compat_cluster_health_check, es_compat_delete_scroll

Made pub(crate): es_compat_index_mapping, es_compat_index_search,
es_compat_index_count, es_compat_index_stats, es_compat_cat_indices,
es_compat_index_cat_indices, es_compat_resolve_index,
es_compat_index_field_capabilities, es_compat_index_multi_search, es_scroll

Also made model and rest_handler modules pub(crate) in mod.rs.

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: congx4

quickwit/quickwit-serve/src/elasticsearch_api/mod.rs

@@ -15,8 +15,8 @@
 mod bulk;
 mod bulk_v2;
 mod filter;
-mod model;
-mod rest_handler;
+pub(crate) mod model;
+pub(crate) mod rest_handler;
 
 use std::sync::Arc;
 

quickwit/quickwit-serve/src/elasticsearch_api/rest_handler.rs

@@ -70,6 +70,28 @@ use crate::rest::recover_fn;
 use crate::rest_api_response::{RestApiError, RestApiResponse};
 use crate::{BuildInfo, with_arg};
 
+pub(crate) fn es_compat_cluster_info(
+    config: Arc<NodeConfig>,
+    build_info: &'static BuildInfo,
+) -> Value {
+    json!({
+        "name" : config.node_id,
+        "cluster_name" : config.cluster_id,
+        "cluster_uuid" : config.cluster_id,
+        "tagline" : "You Know, for Search",
+        "version" : {
+            "distribution" : "quickwit",
+            "number" : "7.17.0",
+            "build_hash" : build_info.commit_hash,
+            "build_date" : build_info.build_date,
+            "build_snapshot" : false,
+            "lucene_version" : "8.11.1",
+            "minimum_wire_compatibility_version" : "6.8.0",
+            "minimum_index_compatibility_version" : "6.0.0-beta1",
+        }
+    })
+}
+
 /// Elastic compatible cluster info handler.
 pub fn es_compat_cluster_info_handler(
     node_config: Arc<NodeConfig>,
@@ -80,76 +102,73 @@ pub fn es_compat_cluster_info_handler(
         .and(with_arg(build_info))
         .then(
             |config: Arc<NodeConfig>, build_info: &'static BuildInfo| async move {
-                warp::reply::json(&json!({
-                    "name" : config.node_id,
-                    "cluster_name" : config.cluster_id,
-                    "cluster_uuid" : config.cluster_id,
-                    "tagline" : "You Know, for Search",
-                    "version" : {
-                        "distribution" : "quickwit",
-                        "number" : "7.17.0",
-                        "build_hash" : build_info.commit_hash,
-                        "build_date" : build_info.build_date,
-                        "build_snapshot" : false,
-                        "lucene_version" : "8.11.1",
-                        "minimum_wire_compatibility_version" : "6.8.0",
-                        "minimum_index_compatibility_version" : "6.0.0-beta1",
-                    }
-                }))
+                warp::reply::json(&es_compat_cluster_info(config, build_info))
             },
         )
         .boxed()
 }
 
+pub(crate) fn es_compat_nodes_info(config: Arc<NodeConfig>) -> Value {
+    let advertise_addr = std::net::SocketAddr::new(
+        config.grpc_advertise_addr.ip(),
+        config.rest_config.listen_addr.port(),
+    );
+    json!({
+        "nodes": {
+            config.node_id.as_str(): {
+                "roles": ["data", "ingest"],
+                "http": {
+                    "publish_address": advertise_addr.to_string()
+                }
+            }
+        }
+    })
+}
+
 /// GET _elastic/_nodes/http
 pub fn es_compat_nodes_handler(
     node_config: Arc<NodeConfig>,
 ) -> impl Filter<Extract = (impl warp::Reply,), Error = Rejection> + Clone {
     elastic_nodes_filter()
         .and(with_arg(node_config))
         .then(|config: Arc<NodeConfig>| async move {
-            let advertise_addr = std::net::SocketAddr::new(
-                config.grpc_advertise_addr.ip(),
-                config.rest_config.listen_addr.port(),
-            );
-            warp::reply::json(&json!({
-                "nodes": {
-                    config.node_id.as_str(): {
-                        "roles": ["data", "ingest"],
-                        "http": {
-                            "publish_address": advertise_addr.to_string()
-                        }
-                    }
-                }
-            }))
+            warp::reply::json(&es_compat_nodes_info(config))
         })
         .boxed()
 }
 
+pub(crate) fn es_compat_search_shards(index_id: String, config: Arc<NodeConfig>) -> Value {
+    json!({
+        "shards": [[{
+            "index": index_id,
+            "shard": 0,
+            "primary": true,
+            "node": config.node_id.as_str()
+        }]]
+    })
+}
+
 /// GET _elastic/{index}/_search_shards
 pub fn es_compat_search_shards_handler(
     node_config: Arc<NodeConfig>,
 ) -> impl Filter<Extract = (impl warp::Reply,), Error = Rejection> + Clone {
     elastic_search_shards_filter()
         .and(with_arg(node_config))
         .then(|index_id: String, config: Arc<NodeConfig>| async move {
-            warp::reply::json(&json!({
-                "shards": [[{
-                    "index": index_id,
-                    "shard": 0,
-                    "primary": true,
-                    "node": config.node_id.as_str()
-                }]]
-            }))
+            warp::reply::json(&es_compat_search_shards(index_id, config))
         })
         .boxed()
 }
 
+pub(crate) fn es_compat_aliases() -> Value {
+    Value::Object(Map::new())
+}
+
 /// GET _elastic/_aliases
 pub fn es_compat_aliases_handler()
 -> impl Filter<Extract = (impl warp::Reply,), Error = Rejection> + Clone {
     elastic_aliases_filter()
-        .then(|| async { Ok(Value::Object(Map::new())) })
+        .then(|| async { Ok(es_compat_aliases()) })
         .map(|result| make_elastic_api_response(result, BodyFormat::default()))
         .recover(recover_fn)
         .boxed()
@@ -180,7 +199,7 @@ async fn get_index_metadata(
     Ok(index_metadata)
 }
 
-async fn es_compat_index_mapping(
+pub(crate) async fn es_compat_index_mapping(
     index_id: String,
     mut metastore: MetastoreServiceClient,
     search_service: Arc<dyn SearchService>,
@@ -296,6 +315,15 @@ pub fn es_compat_cluster_health_handler(
         (status = 503, description = "The cluster is unhealthy.", body = bool),
     ),
 )]
+pub(crate) async fn es_compat_cluster_health_check(cluster: &Cluster) -> (Value, StatusCode) {
+    let is_ready = cluster.is_self_node_ready().await;
+    if is_ready {
+        (json!({"status": "green"}), StatusCode::OK)
+    } else {
+        (json!({"status": "red"}), StatusCode::SERVICE_UNAVAILABLE)
+    }
+}
+
 /// Get Node Liveliness
 async fn es_compat_cluster_health(
     query_params: HashMap<String, String>,
@@ -308,18 +336,8 @@ async fn es_compat_cluster_health(
         }));
         return with_status(error_body, StatusCode::BAD_REQUEST);
     }
-    let is_ready = cluster.is_self_node_ready().await;
-    if is_ready {
-        with_status(
-            warp::reply::json(&json!({"status": "green"})),
-            StatusCode::OK,
-        )
-    } else {
-        with_status(
-            warp::reply::json(&json!({"status": "red"})),
-            StatusCode::SERVICE_UNAVAILABLE,
-        )
-    }
+    let (body, status) = es_compat_cluster_health_check(&cluster).await;
+    with_status(warp::reply::json(&body), status)
 }
 
 /// GET _elastic/{index}/_stats
@@ -423,19 +441,21 @@ pub fn es_compat_scroll_handler(
         .boxed()
 }
 
+pub(crate) fn es_compat_delete_scroll() -> Value {
+    json!({
+        "succeeded": true,
+        "num_freed": 0
+    })
+}
+
 /// DELETE _elastic/_search/scroll
 ///
 /// Clears a scroll context. Quickwit manages scroll lifetime via TTL,
 /// so this is a no-op that returns success.
 pub fn es_compat_delete_scroll_handler()
 -> impl Filter<Extract = (impl warp::Reply,), Error = Rejection> + Clone {
     elastic_delete_scroll_filter()
-        .then(|| async {
-            Ok::<_, ElasticsearchError>(json!({
-                "succeeded": true,
-                "num_freed": 0
-            }))
-        })
+        .then(|| async { Ok::<_, ElasticsearchError>(es_compat_delete_scroll()) })
         .map(|result| make_elastic_api_response(result, BodyFormat::default()))
         .recover(recover_fn)
         .boxed()
@@ -621,11 +641,11 @@ fn partial_hit_from_search_after_param(
 }
 
 #[derive(Debug, Serialize, Deserialize)]
-struct ElasticsearchCountResponse {
-    count: u64,
+pub(crate) struct ElasticsearchCountResponse {
+    pub(crate) count: u64,
 }
 
-async fn es_compat_index_count(
+pub(crate) async fn es_compat_index_count(
     index_id_patterns: Vec<String>,
     search_params: SearchQueryParamsCount,
     search_body: SearchBody,
@@ -642,7 +662,7 @@ async fn es_compat_index_count(
     Ok(search_response_rest)
 }
 
-async fn es_compat_index_search(
+pub(crate) async fn es_compat_index_search(
     index_id_patterns: Vec<String>,
     search_params: SearchQueryParams,
     search_body: SearchBody,
@@ -704,7 +724,7 @@ async fn es_compat_stats(
     es_compat_index_stats(vec!["*".to_string()], metastore).await
 }
 
-async fn es_compat_index_stats(
+pub(crate) async fn es_compat_index_stats(
     index_id_patterns: Vec<String>,
     mut metastore: MetastoreServiceClient,
 ) -> Result<ElasticsearchStatsResponse, ElasticsearchError> {
@@ -729,14 +749,14 @@ async fn es_compat_index_stats(
     Ok(search_response_rest)
 }
 
-async fn es_compat_cat_indices(
+pub(crate) async fn es_compat_cat_indices(
     query_params: CatIndexQueryParams,
     metastore: MetastoreServiceClient,
 ) -> Result<Vec<serde_json::Value>, ElasticsearchError> {
     es_compat_index_cat_indices(vec!["*".to_string()], query_params, metastore).await
 }
 
-async fn es_compat_index_cat_indices(
+pub(crate) async fn es_compat_index_cat_indices(
     index_id_patterns: Vec<String>,
     query_params: CatIndexQueryParams,
     mut metastore: MetastoreServiceClient,
@@ -783,7 +803,7 @@ async fn es_compat_index_cat_indices(
     Ok(search_response_rest)
 }
 
-async fn es_compat_resolve_index(
+pub(crate) async fn es_compat_resolve_index(
     index_id_patterns: Vec<String>,
     mut metastore: MetastoreServiceClient,
 ) -> Result<ElasticsearchResolveIndexResponse, ElasticsearchError> {
@@ -801,7 +821,7 @@ async fn es_compat_resolve_index(
     })
 }
 
-async fn es_compat_index_field_capabilities(
+pub(crate) async fn es_compat_index_field_capabilities(
     index_id_patterns: Vec<String>,
     search_params: FieldCapabilityQueryParams,
     search_body: FieldCapabilityRequestBody,
@@ -946,7 +966,7 @@ fn convert_hit(
     }
 }
 
-async fn es_compat_index_multi_search(
+pub(crate) async fn es_compat_index_multi_search(
     payload: Bytes,
     multi_search_params: MultiSearchQueryParams,
     search_service: Arc<dyn SearchService>,
@@ -1050,7 +1070,7 @@ async fn es_compat_index_multi_search(
     Ok(multi_search_response)
 }
 
-async fn es_scroll(
+pub(crate) async fn es_scroll(
     scroll_query_params: ScrollQueryParams,
     search_service: Arc<dyn SearchService>,
 ) -> Result<ElasticsearchResponse, ElasticsearchError> {