Watcher chain results not in the ctx.payload for a condition

Elastic Stack Elasticsearch
1

Lately I ran into an issue I am trying to make a watcher. The main purpose is to see if the elastic agents we are running are still producing logs. The agents run on different hosts, therefore the query on the host names and then aggregated in buckets. But because those are the ones that are still sending logs, I also added a transform stage that has a list with all the hosts and checks wheter one is missing.
Now I debug the query and the script seperatly and they both work. Only together I get an error saying the missing_hosts is empty?

This is the watcher(obfuscated the names and lowered the trigger for testing):

{
  "trigger": {
    "schedule": {
      "interval": "1m"
    }
  },
  "input": {
    "chain": {
      "inputs": [
        {
          "first": {
            "search": {
              "request": {
                "search_type": "query_then_fetch",
                "indices": [
                  "index*"
                ],
                "rest_total_hits_as_int": true,
                "body": {
                  "size": 0,
                  "query": {
                    "bool": {
                      "must": [
                        {
                          "terms": {
                            "host.hostname": [
                              "name",
                              "name1",
                              "name2",
                              "name3"
                            ]
                          }
                        },
                        {
                          "range": {
                            "@timestamp": {
                              "gte": "now-15m",
                              "lte": "now"
                            }
                          }
                        }
                      ]
                    }
                  },
                  "aggs": {
                    "by_hostname": {
                      "terms": {
                        "field": "host.hostname",
                        "size": 10
                      }
                    }
                  }
                }
              }
            }
          }
        },
        {
          "second": {
            "transform": {
              "script": {
                "source": "def expected=['test','name1','name2','name3','name']; def seen=ctx.payload.first.aggregations.by_hostname.buckets.stream().map(b -> b.key).toList(); def missing=expected.stream().filter(e -> !seen.contains(e)).toList(); return ['missing_hosts': missing, 'seen_hosts': seen];",
                "lang": "painless"
              }
            }
          }
        }
      ]
    }
  },
  "condition": {
    "script": {
      "source": "return ctx.payload.second.missing_hosts.size() > 0;",
      "lang": "painless"
    }
  },
  "actions": {
    "notify": {
      "email": {
        "profile": "standard",
        "to": [
          "email"
        ],
        "subject": "ALERT: No logs from {NAME} in the last 15 minutes",
        "body": {
          "text": "The following hosts have NOT sent any data in the last 15 minutes: ALL missing hostst: {{#ctx.payload.second.missing_hosts}}{{.}} {{/ctx.payload.second.missing_hosts}} All seen hosts: {{#ctx.payload.second.seen_hosts}}{{.}} {{/ctx.payload.second.seen_hosts}}"
        }
      }
    }
  }
}

Here is the error:
"result": {
"execution_time": "2026-01-16T08:09:30.278Z",
"execution_duration": 82,
"input": {
"type": "chain",
"status": "failure",
"error": {
"root_cause": [
{
"type": "null_pointer_exception",
"reason": "Cannot invoke "org.elasticsearch.xpack.core.watcher.watch.Payload.data()" because the return value of "org.elasticsearch.xpack.core.watcher.input.Input$Result.payload()" is null"
}
],
"type": "null_pointer_exception",
"reason": "Cannot invoke "org.elasticsearch.xpack.core.watcher.watch.Payload.data()" because the return value of "org.elasticsearch.xpack.core.watcher.input.Input$Result.payload()" is null"
},
"chain": {}
},
"actions":
},
"messages": [
"failed to execute watch input"
]

2

Hello @Joey_Visbeen

We can use the chain input job without transform , the action part needs to be updated as per your requirement along with the time range as it was used as 5h incase below code is as per the requirement (script part generated using LLM) -

{
  "trigger": {
    "schedule": {
      "interval": "1m"
    }
  },
  "input": {
    "chain": {
      "inputs": [
        {
          "first": {
            "search": {
              "request": {
                "search_type": "query_then_fetch",
                "indices": [
                  "index*"
                ],
                "rest_total_hits_as_int": true,
                "body": {
                  "size": 0,
                  "query": {
                    "bool": {
                      "must": [
                        {
                          "terms": {
                            "host.hostname": [
                              "name",
                              "name1",
                              "name2",
                              "name3"
                            ]
                          }
                        },
                        {
                          "range": {
                            "@timestamp": {
                              "gte": "now-5h",
                              "lte": "now"
                            }
                          }
                        }
                      ]
                    }
                  },
                  "aggs": {
                    "by_hostname": {
                      "terms": {
                        "field": "host.hostname",
                        "size": 10
                      }
                    }
                  }
                }
              }
            }
          }
        }
      ]
    }
  },
  "condition": {
    "script": {
      "source": """
        def expected = ['name','name1','name2','name3'];

        if (ctx.payload.first == null || ctx.payload.first.aggregations == null) {
          return true;
        }

        def buckets = ctx.payload.first.aggregations.by_hostname.buckets;
        def seen = [];

        for (def b : buckets) {
          seen.add(b.key);
        }

        for (def h : expected) {
          if (!seen.contains(h)) {
            return true;
          }
        }

        return false;
      """,
      "lang": "painless"
    }
  },
  "actions": {
    "index_alert": {
      "transform": {
        "script": {
          "source": """
            def expected = ['name','name1','name2','name3'];
            def seen = [];

            if (ctx.payload.first != null && ctx.payload.first.aggregations != null) {
              for (def b : ctx.payload.first.aggregations.by_hostname.buckets) {
                seen.add(b.key);
              }
            }

            def missing = [];
            for (def h : expected) {
              if (!seen.contains(h)) {
                missing.add(h);
              }
            }

            return [
              '@timestamp': ctx.execution_time,
              'missing_hosts': missing,
              'seen_hosts': seen,
              'watch_id': ctx.watch_id
            ];
          """,
          "lang": "painless"
        }
      },
      "index": {
        "index": "abc"
      }
    }
  }
}

output :

  "seen_hosts": [
              "name",
              "name1",
              "name2"
            ],
            "missing_hosts": [
              "name3"
            ]
          }
        },

Thanks!!