How to put the value of {{ctx.payload.aggregations.bucketAgg.buckets.key}} into Watcher body

Elastic Stack Elasticsearch
1

Hello :slight_smile:

What i am trying to do is to hand over the value of

"{{ctx.payload.aggregations.bucketAgg.buckets.key}}"

to the body of my Watcher.

However, the value is always empty. But in truth it is not!

Here you can see the relevant part of my request:

  "actions": {
    "mobots_webhook": {
      "webhook": {
        "scheme": "https",
        "host": "www.xyz.com",
        "port": 443,
        "method": "post",
        "path": "/block",
        "params": {},
        "headers": {
          "Host": "www.xyz.com",
          "Content-Type": "application/json"
        },
        "body": {
          "source": {
            "WorkerId": "{{ctx.payload.aggregations.bucketAgg.buckets.key}}",
            "Reason": "for test"
          },
          "lang": "mustache",
          "options": {
            "content_type": "application/json; charset=UTF-8"
          }
        }
      }
    }
  }

Here you can see the relevant execution output of my Watcher:
(You can see the "key": "A27CQJNO9GMFTK" in the results and at the very end you can see the body of the webhook and the empty "WorkerID")

  "result": {
    "execution_time": "2019-08-04T20:13:03.163Z",
    "execution_duration": 130,
    "input": {
      "type": "search",
      "status": "success",
      "payload": {
        "_shards": {
          "total": 2,
          "failed": 0,
          "successful": 2,
          "skipped": 0
        },
        "hits": {
          "hits": [],
          "total": 1,
          "max_score": null
        },
        "took": 2,
        "timed_out": false,
        "aggregations": {
          "bucketAgg": {
            "doc_count_error_upper_bound": 0,
            "sum_other_doc_count": 0,
            "buckets": [
              {
                "doc_count": 1,
                "metricAgg": {
                  "value": 0
                },
                "key": "A27CQJNO9GMFTK"
              }
            ]
          }
        }
      },
      "search": {
        "request": {
          "search_type": "query_then_fetch",
          "indices": [
            "mobots_assignments*"
          ],
          "rest_total_hits_as_int": true,
          "body": {
            "size": 0,
            "query": {
              "bool": {
                "must": [
                  {
                    "match_phrase": {
                      "assignment_completed": {
                        "query": true
                      }
                    }
                  }
                ],
                "filter": {
                  "range": {
                    "start_time": {
                      "gte": "2019-08-04T20:13:03.087Z||-75m",
                      "lte": "2019-08-04T20:13:03.087Z",
                      "format": "strict_date_optional_time||epoch_millis"
                    }
                  }
                }
              }
            },
            "aggs": {
              "bucketAgg": {
                "terms": {
                  "field": "workerId.keyword",
                  "size": 1,
                  "order": {
                    "metricAgg": "asc"
                  }
                },
                "aggs": {
                  "metricAgg": {
                    "min": {
                      "field": "f1score"
                    }
                  }
                }
              }
            }
          }
        }
      }
    },
    "condition": {
      "type": "script",
      "status": "success",
      "met": true
    },
    "actions": [
      {
        "id": "mobots_webhook",
        "type": "webhook",
        "status": "failure",
        "reason": "received [500] status code",
        "webhook": {
          "request": {
            "host": "www.themobots.com",
            "port": 443,
            "scheme": "https",
            "method": "post",
            "path": "/create_worker_block",
            "params": {
              "p": ""
            },
            "headers": {
              "Host": "www.themobots.com",
              "Content-Type": "application/json"
            },
            "body": "{\"WorkerId\":\"\",\"Reason\":\"for test\"}"
          },
          "response": {
            "status": 500,
            },
            .....
          }
        }
      }
    ]
  },
  "messages": []
}

Can someone please tell me what I can do to hand over the "key": "A27CQJNO9GMFTK" to my Action body ?

Thank you!

2

if you look at the JSON, you can see that aggregations.bucketAgg.buckets is an array, and thus you cannot just use the key field as it is inside an array.

This example in the docs should help to show how to loop through such an array using mustache.

--Alex

3

Thank you very much Alex!
I read about it but I did not take it seriously enough.
I can perfectly get the IDs (keys) now.

"WorkerId": "{{#ctx.payload.aggregations.bucketAgg.buckets}}{{key}} {{/ctx.payload.aggregations.bucketAgg.buckets}}"

Unfortunately I have a new problem now. Maybe you could help me with that as well.

For every ID (key) I get from the array I need do send a separate POST request. Do you think this is possible or do I have to modify my endpoint to deal with more than one value at the same time.

4

This is possible with Elasticsearch 7.3. See https://www.elastic.co/guide/en/elastic-stack-overview/7.3/action-foreach.html

5

Ahh.. I use version 7.2. That's why I did not immediately find it in the documentation. I will probably need an update.

I have one last question know.
How can I run the Watcher only if the number of "'assignment_completed' = true" is greater than 5? I just can not get it. I am actually just asking for "assignment_completed": {"query": true }

  "input": {
    "search": {
      "request": {
        "search_type": "query_then_fetch",
        "indices": [
          "mobots_assignments*"
        ],
        "rest_total_hits_as_int": true,
        "body": {
          "size": 0,
          "query": {
            "bool": {
              "must": [
                {
                  "match_phrase": {
                    "assignment_completed": {
                      "query": true
                    }
                  }
                },
                {
                    "match_phrase": {
                        "qt": {
                            "query": false
                            
                        }
                    }
                }
              ],
              "filter": {
                "range": {
                  "start_time": {
                    "gte": "{{ctx.trigger.scheduled_time}}||-60m",
                    "lte": "{{ctx.trigger.scheduled_time}}",
                    "format": "strict_date_optional_time||epoch_millis"
                  }
                }
              }
            }
          },
          "aggs": {
            "bucketAgg": {
              "terms": {
                "field": "workerId.keyword",
                "size": 10,
                "order": {
                  "metricAgg": "asc"
                }
              },
              "aggs": {
                "metricAgg": {
                  "avg": {
                    "field": "f1score"
                  }
                }
              }
            }
          }
        }
      }
    }
  },
  "condition": {
    "script": {
      "source": "ArrayList arr = ctx.payload.aggregations.bucketAgg.buckets; for (int i = 0; i < arr.length; i++) { if (arr[i]['metricAgg'].value < params.threshold) { return true; } } return false;",
      "lang": "painless",
      "params": {
        "threshold": 0.7
      }
    }
  },

Thank you for your help so far! :innocent:

6

@isabel you can use "min_doc_count ": 5 in your first aggregation.
That would make sure that each worker_id has minimum occurrence of 5, i.e. each of the resulting workers in that bucket have atleast 5 completed assignments.

1 Like
7

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.