# Detect newly inserted documents in an index

**URL:** <https://discuss.elastic.co/t/detect-newly-inserted-documents-in-an-index/368978>\
**Category:** Elasticsearch\
**Created:** [October 17, 2024, 7:42am UTC](https://discuss.elastic.co/t/detect-newly-inserted-documents-in-an-index/368978 "2024-10-17T07:42:17Z")\
**Posts on this page:** 9\
**Page:** 1

<div class="post-metadata">

**Author:** ![Saanchita\_V](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/saanchita_v/32/137428_2.png) [@Saanchita\_V](https://discuss.elastic.co/u/Saanchita_V)\
**Post date:** [October 17, 2024, 7:42am UTC](https://discuss.elastic.co/t/detect-newly-inserted-documents-in-an-index/368978/1 "2024-10-17T07:42:17Z")

</div>

Okay so here is my usecase  
I have an index say one with say 200000 documents( these documents are very bulky ) in a source elastic server  
Now I am migrating all the data to another server (we'll call it target elastic server)  
And everytime I perform some action in my application it gets logged in this index  
So the documents keep getting inserted very frequently into this index  
So I would have to ensure there is no data loss between the source and the target elastic index which is why I would have to keep doing multiple iterations for migrating this index  
And currently I am dropping the index in the target and creating it everytime I run my migration script  
But I don't want to do this  
I want to just see what documents were inserted newly in the source elastic index and migrate them from the source to target  
Also I am not using the @timestamp or any custom field to track the timestamp of inserting the document  
I read in the optimistic concurrency control doc that \_seq\_no is something that can be used to track a document and is unique and is not random like the \_id  
So is it possible for me to use the \_seq\_no for this use case

My source elastic server is 7.17 and target is 8.15

---

<div class="post-metadata">

**Author:** ![Quentin\_Pradet](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/quentin_pradet/32/94192_2.png) [@Quentin\_Pradet](https://discuss.elastic.co/u/Quentin_Pradet)\
**Post date:** [October 17, 2024, 12:28pm UTC](https://discuss.elastic.co/t/detect-newly-inserted-documents-in-an-index/368978/2 "2024-10-17T12:28:03Z")

</div>

Hello! Have you considered using the [reindex API](https://www.elastic.co/guide/en/elasticsearch/reference/8.15/docs-reindex.html) with `version_type` set to external (so that existing docs are updated) and [reindexing from a remote](https://www.elastic.co/guide/en/elasticsearch/reference/8.15/docs-reindex.html#reindex-from-remote)?

To do this automatically, you can use [cross-cluster replication](https://www.elastic.co/guide/en/elasticsearch/reference/current/xpack-ccr.html) which requires a license of Platinum or above.

---

<div class="post-metadata">

**Author:** ![Saanchita\_V](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/saanchita_v/32/137428_2.png) [@Saanchita\_V](https://discuss.elastic.co/u/Saanchita_V)\
**Post date:** [October 17, 2024, 1:32pm UTC](https://discuss.elastic.co/t/detect-newly-inserted-documents-in-an-index/368978/3 "2024-10-17T13:32:26Z")

</div>

Hey  
Can the reindex api be used between two elastic servers?  
Here I am trying to migrate from one server to another

---

<div class="post-metadata">

**Author:** ![Quentin\_Pradet](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/quentin_pradet/32/94192_2.png) [@Quentin\_Pradet](https://discuss.elastic.co/u/Quentin_Pradet)\
**Post date:** [October 17, 2024, 1:44pm UTC](https://discuss.elastic.co/t/detect-newly-inserted-documents-in-an-index/368978/4 "2024-10-17T13:44:42Z")

</div>

Yes, it is supported, it’s called a remote reindex, as I mentioned.

---

<div class="post-metadata">

**Author:** ![Saanchita\_V](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/saanchita_v/32/137428_2.png) [@Saanchita\_V](https://discuss.elastic.co/u/Saanchita_V)\
**Post date:** [October 17, 2024, 4:47pm UTC](https://discuss.elastic.co/t/detect-newly-inserted-documents-in-an-index/368978/5 "2024-10-17T16:47:40Z")

</div>

Thank you so much!  
I'll try that approach and let you know if it solves the problem

---

<div class="post-metadata">

**Author:** ![Saanchita\_V](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/saanchita_v/32/137428_2.png) [@Saanchita\_V](https://discuss.elastic.co/u/Saanchita_V)\
**Post date:** [October 17, 2024, 6:08pm UTC](https://discuss.elastic.co/t/detect-newly-inserted-documents-in-an-index/368978/6 "2024-10-17T18:08:43Z")

</div>

Hey okay so say below is my code using remote reindex

```auto
migrateNotifs = await targetElasticClient.reindex({
                      source: {
                        remote: {
                          host: deconstructURL.href,
                          username: deconstructURL.username,
                          password: deconstructURL.password,
                        },
                        index: "notifications",
                      },
                      dest: {
                        index: "notifications",
                      },
                    refresh: true, // Ensure documents are immediately searchable.
                    wait_for_completion: true, // Block until the operation completes.
                    slices: 5, // Optional: Parallelize the process.
                  });

```

I have followed a structure similar to that in the es docs

But I keep getting this error  
backend: =|ConfigurationError: Missing required parameter: body

Below is the exact structure in the doc

```auto
const response = await client.reindex({
  source: {
    remote: {
      host: "http://otherhost:9200",
      username: "user",
      password: "pass",
    },
    index: "my-index-000001",
    query: {
      match: {
        test: "data",
      },
    },
  },
  dest: {
    index: "my-new-index-000001",
  },
});

```

What is wrong in my code?

---

<div class="post-metadata">

**Author:** ![Quentin\_Pradet](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/quentin_pradet/32/94192_2.png) [@Quentin\_Pradet](https://discuss.elastic.co/u/Quentin_Pradet)\
**Post date:** [October 18, 2024, 5:49am UTC](https://discuss.elastic.co/t/detect-newly-inserted-documents-in-an-index/368978/7 "2024-10-18T05:49:49Z")

</div>

Hello,

Assuming you're using JavaScript, your code looks correct. I [started Elasticsearch with start-local](https://www.elastic.co/guide/en/elasticsearch/reference/current/run-elasticsearch-locally.html) and then used this code:

```js
const { Client } = require('@elastic/elasticsearch')
const client = new Client({
  node: 'http://localhost:9200/',
  auth: {
    apiKey: '...'
  }
})

async function main() {
  r = await client.info()
  console.log(r)

  const response = await client.reindex({
    source: {
      index: "my-index",
    },
    dest: {
      index: "my-dest-index",
    },
    refresh: true,
    wait_for_completion: true,
    slices: 5
  });
  console.log(response);
}

main()

```

The above prints:

```auto
{
  took: 17,
  timed_out: false,
  total: 1,
  updated: 1,
  created: 0,
  deleted: 0,
  batches: 1,
  version_conflicts: 0,
  noops: 0,
  retries: { bulk: 0, search: 0 },
  throttled_millis: 0,
  requests_per_second: -1,
  throttled_until_millis: 0,
  failures: []
}

```

I did not manage to make `version_type: external` work yet, however. And I'm not using remote reindex as I only tested with a single cluster, but your error does not seem related to that at all.

Note that you can also try in Kibana directly, and that all examples in the docs can be displayed as Kibana request but also in JavaScript and Python.

---

<div class="post-metadata">

**Author:** ![Saanchita\_V](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/saanchita_v/32/137428_2.png) [@Saanchita\_V](https://discuss.elastic.co/u/Saanchita_V)\
**Post date:** [October 18, 2024, 9:09am UTC](https://discuss.elastic.co/t/detect-newly-inserted-documents-in-an-index/368978/8 "2024-10-18T09:09:14Z")

</div>

Yeah it works if I use the reindex api within the same cluster.

I will be initiating the migration process on the click of a button in my ui which sends a request to the backend where I have my migration script so I don't think kibana can really help me out here

Here is how I am handling it currently

```auto
let indexExistsinTarget = await targetElasticClient.indices.exists({index: "notifications"});
            let migrateNotifs = false;
            if(indexExistsinTarget.body){
                              await targetElasticClient.indices.delete({index: "notifications"});
            }
            const {body: sourceMappingNotif} = await sourceElasticClient.indices.getMapping({index: "notifications"});
            let targetBodyNotif = {
                "settings":{
                    "index":{
                        "blocks": {
                            "read_only_allow_delete": "false"
                          }
                    }
                },
                "mappings": sourceMappingNotif["notifications"].mappings
            }
            await targetElasticClient.indices.create({
                index: "notifications",
                body: targetBodyNotif
            }) 
        
            migrateNotifs = await reindexIndexDatainBatches(sourceElasticClient,targetElasticClient,"notifications","notifications",null);

export async function reindexIndexDatainBatches(sourceElasticClient,targetElasticClient,sourceIndex,targetIndex,csvType){
    let scrollId = null;
    let moreRecordsExist = true;
    const batchSize = sourceIndex==="activity-log"?2000:9900;
    let totalIndexed = 0;
    let failedDocuments = [];

    while(moreRecordsExist){
        let body;
        if(!scrollId){
            console.log(`Starting search for index ${sourceIndex} `);
            const searchParams = {
            index: sourceIndex,
            scroll: '20m',
            size: batchSize
            };
            const response = await sourceElasticClient.search(searchParams);
            body = response.body;
            console.log("Initial search successful",body.hits.hits.length);
        }else{
            console.log(`Continuing scroll for index ${sourceIndex}`);
            const scrollResponse = await sourceElasticClient.scroll({
                scroll_id: scrollId,
                scroll: '20m'
            });
            body = scrollResponse.body;
            console.log(`Next batch retrieved ${body.hits.hits.length}`);
        }

        scrollId = body._scroll_id;
        const hits = body.hits.hits;

        if(hits.length === 0){
            console.log("No more records to scroll");
            moreRecordsExist = false;
            break;
        }

        const bulkBody = hits.flatMap(doc => {
            const baseDoc = {index: {_index:targetIndex,_id:doc._id}};
            const docSource = {...doc._source};
            if(docSource.relation_type){
                if(docSource.relation_type.parent){
                    baseDoc.index.routing = docSource.relation_type.parent;
                }
                else{
                    baseDoc.index.routing = doc._id;
                }
            }
            if(csvType){
                docSource.csv_type = csvType;
            }
            return [baseDoc,docSource];
        });    
        
        try{
            const bulkWriteResponse = await targetElasticClient.bulk({body: bulkBody,refresh: true});
            if(bulkWriteResponse.body.errors){
                bulkWriteResponse.body.items.forEach((item,idx)=>{
                    if(item.index && item.index.error){
                        console.error(`Error indexing document ${bulkBody[idx * 2].index._id}: ${JSON.stringify(item.index.error)}`);
                    }
                    return false;
                 })
            }

        }catch(err){
            if(err.meta && err.meta.statusCode === 413){
                console.log("Payload too large, storing it to push it in later");
                bulkBody.forEach((_,idx)=>{
                    if(idx%2===0){
                        failedDocuments.push({
                            metadata: bulkBody[idx],
                            source: bulkBody[idx+1]
                        });
                    }
                });
            }else{
                console.log("Error Indexing documents",err);
                return err;
            }
        }       
 
        totalIndexed = totalIndexed + hits.length;
        console.log("TOTAL INDEXED: " + totalIndexed);
    };

    if(failedDocuments.length>0){
        console.log("Retrying for failed documents.Number of documents: " + failedDocuments.length);
        for(let i=0; i<failedDocuments.length;i=i+30){
            console.log("============== BATCH ==============",i);
            const batch = failedDocuments.slice(i, i+30);
            const bulkRetryBody = batch.flatMap(doc => [
                doc.metadata,
                doc.source
            ]);

            try{
                const retryResponse = await targetElasticClient.bulk({
                    body: bulkRetryBody,
                    refresh: true
                });
                if(retryResponse.body.errors){
                    retryResponse.body.items.forEach((item,idx)=>{
                        if(item.index && item.index.error){
                            console.error(`Error indexing document ${bulkRetryBody[idx * 2].index._id}: ${JSON.stringify(item.index.error)}`);
                        }
                        return false;
                     })
                }
            }catch(retryErr){
                console.log(`Error retrying documents in batch : ${retryErr.message}`);
                return false;
            }
        }

    }

    return true;
}

```

---

<div class="post-metadata">

**Author:** ![Quentin\_Pradet](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/quentin_pradet/32/94192_2.png) [@Quentin\_Pradet](https://discuss.elastic.co/u/Quentin_Pradet)\
**Post date:** [October 18, 2024, 2:32pm UTC](https://discuss.elastic.co/t/detect-newly-inserted-documents-in-an-index/368978/9 "2024-10-18T14:32:33Z")

</div>

> [@Saanchita\_V](#):
>
> I don't think kibana can really help me out here

Right, I was only suggesting to avoid errors like the one below that cannot happen in Kibana:

> [@Saanchita\_V](#):
>
> backend: =|ConfigurationError: Missing required parameter: body

Regarding your new code snippet, yes this reimplements the reindex API (which also uses scrolls under the hood). It's more work and can be less efficient if the code runs far away from the two clusters. But this works too. 👍