diff --git a/docs/reference/rest-api.md b/docs/reference/rest-api.md index 54ac05c040b..6d0d4ce22e2 100644 --- a/docs/reference/rest-api.md +++ b/docs/reference/rest-api.md @@ -682,6 +682,40 @@ Resets checkpoints of source `source id` of index ID `index id`. It returns an empty body. +### Get source checkpoint + +``` +GET api/v1/indexes//sources//checkpoint +``` + +Gets the checkpoint of source `source id` of index ID `index id`. + +#### Response + +The response is a JSON object, and the content type is `application/json; charset=UTF-8.` + +| Field | Description | Type | +|----------------|-------------------------------------------------------------------|:----------:| +| `index_id` | The index ID of the source. | `String` | +| `source_id` | The source ID. | `String` | +| `checkpoint` | Map of partition ID to the last processed position. | `Object` | + +```json +{ + "index_id": "hdfs-logs", + "source_id": "kafka-source", + "checkpoint": { + "00000000000000000000": "00000000000000000042" + } +} +``` + +:::note + +Sources relying on shards (`ingest` source with ingest V2 and notification based `file` sources) do not store their positions in the index checkpoint, so their `checkpoint` is always empty. Their positions can be retrieved through the shards of the source. + +::: + ### Delete a source ``` diff --git a/quickwit/quickwit-serve/src/index_api/rest_handler.rs b/quickwit/quickwit-serve/src/index_api/rest_handler.rs index bc4c8d9b105..446de26beaf 100644 --- a/quickwit/quickwit-serve/src/index_api/rest_handler.rs +++ b/quickwit/quickwit-serve/src/index_api/rest_handler.rs @@ -31,9 +31,10 @@ use super::index_resource::{ list_indexes_metadata_handler, update_index_handler, }; use super::source_resource::{ - __path_create_source, __path_delete_source, __path_reset_source_checkpoint, - __path_toggle_source, __path_update_source, ToggleSource, create_source_handler, - delete_source_handler, get_source_handler, get_source_shards_handler, + __path_create_source, __path_delete_source, __path_get_source_checkpoint, + __path_reset_source_checkpoint, __path_toggle_source, __path_update_source, + SourceCheckpointResponse, ToggleSource, create_source_handler, delete_source_handler, + get_source_checkpoint_handler, get_source_handler, get_source_shards_handler, reset_source_checkpoint_handler, toggle_source_handler, update_source_handler, }; use super::split_resource::{ @@ -59,10 +60,11 @@ use crate::simple_list::from_simple_list; create_source, update_source, reset_source_checkpoint, + get_source_checkpoint, toggle_source, delete_source, ), - components(schemas(ToggleSource, SplitsForDeletion, IndexStats)) + components(schemas(ToggleSource, SplitsForDeletion, IndexStats, SourceCheckpointResponse)) )] pub struct IndexApi; @@ -104,6 +106,7 @@ pub fn index_management_handlers( .boxed() // Sources handlers. .or(reset_source_checkpoint_handler(index_service.metastore())) + .or(get_source_checkpoint_handler(index_service.metastore())) .or(toggle_source_handler(index_service.metastore())) .or(create_source_handler(index_service.clone())) .or(update_source_handler(index_service.clone())) @@ -214,9 +217,13 @@ mod tests { use quickwit_common::ServiceStream; use quickwit_common::uri::Uri; use quickwit_config::{ - CLI_SOURCE_ID, INGEST_API_SOURCE_ID, NodeConfig, SourceParams, VecSourceParams, + CLI_SOURCE_ID, INGEST_API_SOURCE_ID, NodeConfig, SourceConfig, SourceParams, + VecSourceParams, }; use quickwit_indexing::{MockSplitBuilder, mock_split}; + use quickwit_metastore::checkpoint::{ + IndexCheckpointDelta, PartitionId, SourceCheckpointDelta, + }; use quickwit_metastore::{ IndexMetadata, IndexMetadataResponseExt, ListIndexesMetadataResponseExt, ListSplitsRequestExt, ListSplitsResponseExt, SplitState, metastore_for_test, @@ -228,7 +235,7 @@ mod tests { MetastoreService, MetastoreServiceClient, MockMetastoreService, ResetSourceCheckpointRequest, SourceType, ToggleSourceRequest, }; - use quickwit_proto::types::IndexUid; + use quickwit_proto::types::{IndexUid, Position}; use quickwit_storage::StorageResolver; use serde_json::Value as JsonValue; @@ -1333,6 +1340,92 @@ mod tests { Ok(()) } + #[tokio::test] + async fn test_get_source_checkpoint() -> anyhow::Result<()> { + let mut index_metadata = IndexMetadata::for_test( + "quickwit-demo-index", + "file:///path/to/index/quickwit-demo-index", + ); + for source_id in ["source-with-checkpoint", "source-without-checkpoint"] { + index_metadata + .add_source(SourceConfig::for_test(source_id, SourceParams::void())) + .unwrap(); + } + let source_delta = SourceCheckpointDelta::from_partition_delta( + PartitionId::from(0u64), + Position::Beginning, + Position::offset(42u64), + ) + .unwrap(); + index_metadata + .checkpoint + .try_apply_delta(IndexCheckpointDelta { + source_id: "source-with-checkpoint".to_string(), + source_delta, + }) + .unwrap(); + let mut mock_metastore = MockMetastoreService::new(); + mock_metastore + .expect_index_metadata() + .returning(move |index_metadata_request: IndexMetadataRequest| { + if index_metadata_request.index_id.as_deref() != Some("quickwit-demo-index") { + return Err(MetastoreError::NotFound(EntityKind::Index { + index_id: "unknown-index".to_string(), + })); + } + Ok(IndexMetadataResponse::try_from_index_metadata(&index_metadata).unwrap()) + }) + .times(4); + let index_service = IndexService::new( + MetastoreServiceClient::from_mock(mock_metastore), + StorageResolver::unconfigured(), + ); + let index_management_handler = + super::index_management_handlers(index_service, Arc::new(NodeConfig::for_test())) + .recover(recover_fn); + + let resp = warp::test::request() + .path("/indexes/quickwit-demo-index/sources/source-with-checkpoint/checkpoint") + .reply(&index_management_handler) + .await; + assert_eq!(resp.status(), 200); + let actual_response_json: JsonValue = serde_json::from_slice(resp.body())?; + let expected_response_json = serde_json::json!({ + "index_id": "quickwit-demo-index", + "source_id": "source-with-checkpoint", + "checkpoint": { + "00000000000000000000": "00000000000000000042" + } + }); + assert_eq!(actual_response_json, expected_response_json); + + let resp = warp::test::request() + .path("/indexes/quickwit-demo-index/sources/source-without-checkpoint/checkpoint") + .reply(&index_management_handler) + .await; + assert_eq!(resp.status(), 200); + let actual_response_json: JsonValue = serde_json::from_slice(resp.body())?; + let expected_response_json = serde_json::json!({ + "index_id": "quickwit-demo-index", + "source_id": "source-without-checkpoint", + "checkpoint": {} + }); + assert_eq!(actual_response_json, expected_response_json); + + let resp = warp::test::request() + .path("/indexes/quickwit-demo-index/sources/unknown-source/checkpoint") + .reply(&index_management_handler) + .await; + assert_eq!(resp.status(), 404); + + let resp = warp::test::request() + .path("/indexes/unknown-index/sources/source-with-checkpoint/checkpoint") + .reply(&index_management_handler) + .await; + assert_eq!(resp.status(), 404); + Ok(()) + } + #[tokio::test] async fn test_source_toggle() -> anyhow::Result<()> { let mut mock_metastore = MockMetastoreService::new(); diff --git a/quickwit/quickwit-serve/src/index_api/source_resource.rs b/quickwit/quickwit-serve/src/index_api/source_resource.rs index 66f3069b228..10982f41341 100644 --- a/quickwit/quickwit-serve/src/index_api/source_resource.rs +++ b/quickwit/quickwit-serve/src/index_api/source_resource.rs @@ -19,6 +19,7 @@ use quickwit_config::{ }; use quickwit_index_management::{IndexService, IndexServiceError}; use quickwit_metastore::IndexMetadataResponseExt; +use quickwit_metastore::checkpoint::SourceCheckpoint; use quickwit_proto::ingest::Shard; use quickwit_proto::metastore::{ DeleteSourceRequest, EntityKind, IndexMetadataRequest, ListShardsRequest, ListShardsSubrequest, @@ -26,7 +27,7 @@ use quickwit_proto::metastore::{ ResetSourceCheckpointRequest, ToggleSourceRequest, }; use quickwit_proto::types::{IndexId, IndexUid, SourceId}; -use serde::Deserialize; +use serde::{Deserialize, Serialize}; use tracing::info; use warp::{Filter, Rejection}; @@ -279,6 +280,77 @@ pub async fn reset_source_checkpoint( Ok(()) } +/// Checkpoint of a single source. +#[derive(Debug, Serialize, utoipa::ToSchema)] +pub struct SourceCheckpointResponse { + /// The index ID of the source. + #[schema(value_type = String)] + pub index_id: IndexId, + /// The source ID. + #[schema(value_type = String)] + pub source_id: SourceId, + /// Map of partition ID to last processed position. + #[schema(value_type = std::collections::HashMap)] + pub checkpoint: SourceCheckpoint, +} + +pub fn get_source_checkpoint_handler( + metastore: MetastoreServiceClient, +) -> impl Filter + Clone { + warp::path!("indexes" / String / "sources" / String / "checkpoint") + .and(warp::get()) + .and(with_arg(metastore)) + .then(get_source_checkpoint) + .and(extract_format_from_qs()) + .map(into_rest_api_response) + .boxed() +} + +#[utoipa::path( + get, + tag = "Sources", + path = "/indexes/{index_id}/sources/{source_id}/checkpoint", + responses( + (status = 200, description = "Successfully fetched source checkpoint.", body = SourceCheckpointResponse) + ), + params( + ("index_id" = String, Path, description = "The index ID of the source."), + ("source_id" = String, Path, description = "The source ID whose checkpoint is fetched."), + ) +)] +/// Gets source checkpoint. +/// +/// Sources relying on the shard API (ingest v2, notification based file sources) track their +/// positions in the shards table, so their checkpoint is always empty. +pub async fn get_source_checkpoint( + index_id: IndexId, + source_id: SourceId, + metastore: MetastoreServiceClient, +) -> MetastoreResult { + info!(index_id = %index_id, source_id = %source_id, "get-source-checkpoint"); + let index_metadata_request = IndexMetadataRequest::for_index_id(index_id.to_string()); + let index_metadata = metastore + .index_metadata(index_metadata_request) + .await? + .deserialize_index_metadata()?; + if !index_metadata.sources.contains_key(&source_id) { + return Err(MetastoreError::NotFound(EntityKind::Source { + index_id, + source_id, + })); + } + let checkpoint = index_metadata + .checkpoint + .source_checkpoint(&source_id) + .cloned() + .unwrap_or_default(); + Ok(SourceCheckpointResponse { + index_id, + source_id, + checkpoint, + }) +} + pub fn toggle_source_handler( metastore: MetastoreServiceClient, ) -> impl Filter + Clone {