feat: Endpoint to retrieve the checkpoint of a given source

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/<index id>/sources/<source id>/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
 
 ```

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();

quickwit/quickwit-serve/src/index_api/source_resource.rs

@@ -19,14 +19,15 @@ 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,
     MetastoreError, MetastoreResult, MetastoreService, MetastoreServiceClient,
     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<String, String>)]
+    pub checkpoint: SourceCheckpoint,
+}
+
+pub fn get_source_checkpoint_handler(
+    metastore: MetastoreServiceClient,
+) -> impl Filter<Extract = (impl warp::Reply,), Error = Rejection> + 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<SourceCheckpointResponse> {
+    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<Extract = (impl warp::Reply,), Error = Rejection> + Clone {