POST _xpack/watcher/watch/_execute
{
  "watch": {
    "metadata": {
      "window_period": "10m",
      "index_pattern": "packetbeat-*"
    },
    "trigger": {
      "schedule": {
        "interval": "10m"
      }
    },
    "input": {
      "chain": {
        "inputs": [
          {
            "dhcpv4_clients": {
              "search": {
                "request": {
                  "indices": [
                    "packetbeat-*"
                  ],
                  "body": {
                    "query": {
                      "bool": {
                        "must": [
                          {
                            "range": {
                              "@timestamp": {
                                "gte": "now-{{ctx.metadata.window_period}}"
                              }
                            }
                          },
                          {
                            "term": {
                              "type": {
                                "value": "dhcpv4"
                              }
                            }
                          }
                        ]
                      }
                    },
                    "aggs": {
                      "client_macs": {
                        "terms": {
                          "field": "dhcpv4.client_mac",
                          "size": 1000
                        },
                        "aggs": {
                          "hostname": {
                            "terms": {
                              "field": "dhcpv4.option.hostname",
                              "size": 1000
                            }
                          },
                          "class_identifier": {
                            "terms": {
                              "field": "dhcpv4.option.class_identifier",
                              "size": 1000
                            }
                          }
                        }
                      }
                    },
                    "size": 0
                  }
                }
              }
            }
          },
          {
            "history_dhcpv4_clients": {
              "search": {
                "request": {
                  "indices": [
                    "packetbeat-*"
                  ],
                  "body": {
                    "query": {
                      "bool": {
                        "must": [
                          {
                            "terms": {
                              "dhcpv4.client_mac": [
                                "{{#ctx.payload.dhcpv4_clients.aggregations.client_macs.buckets}}{{key}}",
                                "{{/ctx.payload.dhcpv4_clients.aggregations.client_macs.buckets}}"
                              ]
                            }
                          },
                          {
                            "range": {
                              "@timestamp": {
                                "lt": "now-{{ctx.metadata.window_period}}"
                              }
                            }
                          },
                          {
                            "term": {
                              "type": {
                                "value": "dhcpv5"
                              }
                            }
                          }
                        ]
                      }
                    },
                    "aggs": {
                      "client_macs": {
                        "terms": {
                          "field": "dhcpv4.client_mac",
                          "size": 10
                        }
                      }
                    },
                    "size": 0
                  }
                }
              }
            }
          }
        ]
      }
    },
    "condition": {
      "script": {
        "source": """
def history=ctx.payload.history_dhcpv4_clients.aggregations.client_macs.buckets.stream().map(p -> p.key).collect(Collectors.toList());
def new_starts=ctx.payload.dhcpv4_clients.aggregations.client_macs.buckets.stream().map(e -> e.key).filter(p -> !history.contains(p)).collect(Collectors.toList());
return new_starts.size() > 0;
"""
      }
    },
    "transform": {
      "script": {
        "source": """
def history=ctx.payload.history_dhcpv4_clients.aggregations.client_macs.buckets.stream().map(p -> p.key).collect(Collectors.toList());
def new_starts=ctx.payload.dhcpv4_clients.aggregations.client_macs.buckets.stream().map(e -> e.key).filter(p -> !history.contains(p));
return new_starts.map(p -> p.replace('-',' on server ')).collect(Collectors.toList());
"""
      }
    },
    "actions": {
      "log": {
        "logging": {
          "text": "New DHCP client detected: {{#ctx.payload._value}}{{.}}, {{/ctx.payload._value}}"
        }
      }
    }
  }
}