Watching event data

If you are indexing event data, such as log messages, network traffic, or a web feed, you can create a watch to email notifications when certain events occur. For example, if you index a feed of RSVPs for meetup events happening around the world, you can create a watch that alerts you to interesting events.

To index the meetup data, you can use Logstash to ingest live data from the Meetup.com streaming API, http://stream.meetup.com/2/rsvps.

To ingest this data with Logstash:

  1. Download Logstash and unpack the archive file.

  2. Create a Logstash configuration file that uses the Logstash standard input and the Logstash standard output and save it in logstash-{version} directory as livestream.conf:

    1. input {
    2.   stdin {
    3.     codec => json 
    4.   }
    5. }
    6. filter {
    7.   date {
    8.     match => [ "event.time", "UNIX_MS" ]
    9.     target => "event_time"
    10.   }
    11. }
    12. output { 
    13.   stdout {
    14.     codec => rubydebug
    15.   }
    16.   elasticsearch {
    17.     hosts => "http://localhost:9200"
    18.     user  => "elastic"
    19.     password  => "x-pack-test-password"
    20.   }
    21. }

    The meetup data is formatted in JSON.

    Index the meetup data into Elasticsearch.

  3. To start indexing the meetup data, pipe the RSVP stream into Logstash and specify your livestream.conf configuration file.

    1. curl http://stream.meetup.com/2/rsvps | bin/logstash -f livestream.conf

Now that you’re indexing the meetup RSVPs, you can set up a watch that lets you know about events you might be interested in. For example, let’s create a watch that runs every hour, looks for events that talk about about Open Source, and sends an email with information about the events.

To set up the watch:

  1. Specify how often you want to run the watch by adding a schedule trigger to the watch:

    1. {
    2.   "trigger": {
    3.     "schedule": {
    4.       "interval": "1h"
    5.     }
    6.   },

  2. Load data into the watch payload by creating an input that searches the meetup data for events that have Open Source as a topic. You can use aggregations to group the data by city, consolidate references to the same events, and sort the events by date.

    1. "input": {
    2.     "search": {
    3.       "request": {
    4.         "indices": [
    5.           "logstash" 
    6.         ],
    7.         "body": {
    8.           "size": 0,
    9.           "query": {
    10.             "bool": {
    11.               "filter": [
    12.                 {
    13.                   "range": {
    14.                     "@timestamp": {
    15.                       "gte": "now-3h"
    16.                     }
    17.                   }
    18.                 },
    19.                 {
    20.                   "match": {
    21.                     "group.group_topics.topic_name": "Open Source" 
    22.                   }
    23.                 }
    24.               ]
    25.             }
    26.           },
    27.           "aggs": {
    28.             "group_by_city": {
    29.               "terms": {
    30.                 "field": "group.group_city.keyword", 
    31.                 "size": 5
    32.               },
    33.               "aggs": {
    34.                 "group_by_event": {
    35.                   "terms": {
    36.                     "field": "event.event_url.keyword", 
    37.                     "size": 5
    38.                   },
    39.                   "aggs": {
    40.                     "get_latest": {
    41.                       "terms": {
    42.                         "field": "@timestamp", 
    43.                         "size": 1,
    44.                         "order": {
    45.                           "_key": "desc"
    46.                         }
    47.                       },
    48.                       "aggs": {
    49.                         "group_by_event_name": {
    50.                           "terms": {
    51.                             "field": "event.event_name.keyword" 
    52.                           }
    53.                         }
    54.                       }
    55.                     }
    56.                   }
    57.                 }
    58.               }
    59.             }
    60.           }
    61.         }
    62.       }
    63.     }
    64.   },

    logstash is the default index alias for the Logstash indices containing the meetup data. By default, the Logstash index lifecycle management (ILM) policy rolls this alias to a new index when the index size reaches 50GB or becomes 30 days old. For more information, see ILM defaults in Logstash.

    Find all of the RSVPs with Open Source as a topic.

    Group the RSVPs by city.

    Consolidate multiple RSVPs for the same event.

    Sort the events so the latest events are listed first.

    Group the events by name.

  3. To determine whether or not there are any Open Source events, add a compare condition that checks the watch payload to see if there were any search hits.

    1. "compare" : { "ctx.payload.hits.total" : { "gt" : 0 }}

  4. To send an email when Open Source events are found, add an email action:

    1. "actions": {
    2.     "email_me": {
    3.       "throttle_period": "10m",
    4.       "email": {
    5.         "from": "<from:email address>",
    6.         "to": "<to:email address>",
    7.         "subject": "Open Source Events",
    8.         "body": {
    9.           "html": "Found events matching Open Source: <ul>{{#ctx.payload.aggregations.group_by_city.buckets}}<          li>{{key}} ({{doc_count}})<ul>{{#group_by_event.buckets}}
    10.           <li><a href=\"{{key}}\">{{get_latest.buckets.0.group_by_event_name.buckets.0.key}}</a>
    11.           ({{doc_count}})</li>{{/group_by_event.buckets}}</ul></li>
    12.           {{/ctx.payload.aggregations.group_by_city.buckets}}</ul>"
    13.         }
    14.       }
    15.     }
    16.   }

To enable Watcher to send emails, you must configure an email account in elasticsearch.yml. For more information, see Configuring email accounts.

The complete watch looks like this:

  1. PUT _watcher/watch/meetup
  2. {
  3.   "trigger": {
  4.     "schedule": {
  5.       "interval": "1h"
  6.     }
  7.   },
  8.   "input": {
  9.     "search": {
  10.       "request": {
  11.         "indices": [
  12.           "logstash"
  13.         ],
  14.         "body": {
  15.           "size": 0,
  16.           "query": {
  17.             "bool": {
  18.               "filter": [
  19.                 {
  20.                   "range": {
  21.                     "@timestamp": {
  22.                       "gte": "now-3h"
  23.                     }
  24.                   }
  25.                 },
  26.                 {
  27.                   "match": {
  28.                     "group.group_topics.topic_name": "Open Source"
  29.                   }
  30.                 }
  31.               ]
  32.             }
  33.           },
  34.           "aggs": {
  35.             "group_by_city": {
  36.               "terms": {
  37.                 "field": "group.group_city.keyword",
  38.                 "size": 5
  39.               },
  40.               "aggs": {
  41.                 "group_by_event": {
  42.                   "terms": {
  43.                     "field": "event.event_url.keyword",
  44.                     "size": 5
  45.                   },
  46.                   "aggs": {
  47.                     "get_latest": {
  48.                       "terms": {
  49.                         "field": "@timestamp",
  50.                         "size": 1,
  51.                         "order": {
  52.                           "_key": "desc"
  53.                         }
  54.                       },
  55.                       "aggs": {
  56.                         "group_by_event_name": {
  57.                           "terms": {
  58.                             "field": "event.event_name.keyword"
  59.                           }
  60.                         }
  61.                       }
  62.                     }
  63.                   }
  64.                 }
  65.               }
  66.             }
  67.           }
  68.         }
  69.       }
  70.     }
  71.   },
  72.   "condition": {
  73.     "compare": {
  74.       "ctx.payload.hits.total": {
  75.         "gt": 0
  76.       }
  77.     }
  78.   },
  79.   "actions": {  
  80.     "email_me": {
  81.       "throttle_period": "10m",
  82.       "email": {
  83.         "from": "username@example.org",  
  84.         "to": "recipient@example.org",   
  85.         "subject": "Open Source events",
  86.         "body": {
  87.           "html": "Found events matching Open Source: <ul>{{#ctx.payload.aggregations.group_by_city.buckets}}<li>{{key}} ({{doc_count}})<ul>{{#group_by_event.buckets}}<li><a href=\"{{key}}\">{{get_latest.buckets.0.group_by_event_name.buckets.0.key}}</a> ({{doc_count}})</li>{{/group_by_event.buckets}}</ul></li>{{/ctx.payload.aggregations.group_by_city.buckets}}</ul>"
  88.          }
  89.       }
  90.     }
  91.   }
  92. }

The email body can include Mustache templates to reference data in the watch payload. By default,it will be sanitized to block dangerous content.

Replace the from address with the email address you configured in elasticsearch.yml.

Replace the to address with your email address to receive notifications.

Now that you’ve created your watch, you can use the _execute API to run it without waiting for the schedule to trigger execution:

  1. POST _watcher/watch/meetup/_execute