Fetching new documents on insertion in ElasticSearch with Java

Question

I have been looking for a solution to create a sort of alert when new documents are added to ES via Logstash. I have seen some threads on here such as : stackoverflow.com/a/51980618/4604579, but that does not really serve my purposes as the plug-ins mentioned do not work with the newest version of ELK and there is no Changes API out yet.

So I have resorted to trying 2 different approaches:

Create a Scroll and run over all the documents in a given index using the Search API, retain the last document's ID and use it after a given timeout period to get all documents that were added after it
Creating a Watcher that checks after a given interval (for example 5 minutes) if new documents have been added to an index.

I have advanced on approach 1, where I can scroll through about 50k documents that are currently in ES and retrieve the last documents id (i sort the query based on timestamp in ascending order, that way I know that the last document will be the latest that was inserted). But I don't know how efficient this approach is and I know that a scroller may time out after a given delay, so if no new documents are inserted, that means the scroll will be removed.

I was looking also into using a Watcher, but I don't really understand how I can set up the condition to check if a new document was inserted in a given index. I imagine I can do something of the genre:

PUT _watcher/watch/new_docs
{
  "trigger" : { 
    "schedule" : {
      "interval" : "5s"
    }
  },
  "input" : { 
    "search" : {
      "request" : {
        "indices" : "logstash",
        "body" : {
          "size" : 0,
          "query" : { "match" : { "@timestamp" : "now-5s" } }
        }
      }
    }
  },
  "condition" : { 
    "compare" : { ?? }
  },
  "actions" : { 
    "my_webhook" : {
      "webhook" : {
        "method" : "POST",
        "host" : "mylisteninghost",
        "port" : 9200,
        "path" : "/{{watch_id}}",
        "body" : "New document {{document ID}} errors"
      }
    }

I am not exactly sure how to define or use the Watcher and if it would even work. Can anyone let me know what the best course of action would be? Thank you

EDIT:

For those interested I found a way to poll the ES REST API using Search After. The difference is that using Scroll, there is a snapshot taken of the documents in the ES DB, so any new documents added wont be in this snapshot. Contrary to that, Search After is state-less, which means that it will use unique sorting parameters (in my case timestamp/id) and hold the last one fetched, afterwards we query all documents that come after the held parameters. This way if any new documents are added, they will come after the held timestamp and will be fetched by the query.

Code:

public static void searchAfterElasticData()
        throws FileNotFoundException, IOException, InterruptedException {

    //create a search request for a given index
    SearchRequest search_request = new SearchRequest(elastic_index);
    SearchSourceBuilder source_builder =
            getSearchSourceBuilder("@timestamp", "_id", 100);
    search_request.source(source_builder);
    SearchResponse search_response = null;

    try {
        search_response = client.search(search_request, RequestOptions.DEFAULT);
    } catch (ElasticsearchException | ConnectException ex) {
        log.info("Error while querying Elastic API: {}", ex.toString()); 
    }
    
    if (search_response != null) {
        SearchHit[] search_hits = search_response.getHits().getHits();
        Object[] sort_values = null;

        
        while (search_hits != null) { 

            if (search_hits.length > 0) {
                //if there are records retrieved, parse them
                for (SearchHit hit: search_hits) {
                    Map source_map = hit.getSourceAsMap();
                    try {
                    parse((String)source_map.get("message"));
                    } catch (Exception ex) {
                        log.error("Error while parsing: {}",
                                (String)source_map.get("message"));
                    }
                }
                //get sorting value of last record and do new request
                log.info("Getting sorting values");
                sort_values = search_response.getHits()
                        .getAt(search_hits.length-1).getSortValues();
            } else {
                log.info("Waiting 1 minute for new entries");          
                Thread.sleep(60000);
            }
            source_builder.searchAfter(sort_values);
            search_request.source(source_builder);
            search_response =
                client.search(search_request, RequestOptions.DEFAULT);

            search_hits = search_response.getHits().getHits();

            log.info("Fetched hits: {}", search_hits.length);
            log.info("Searching after for new hits");
        }
    }
}

I still would like to know if it is possible to do the same using a Watcher, also if anyone has any suggestions to make the code more elegant, please share.

Thank you

Fetching new documents on insertion in ElasticSearch with Java

0 Answers