如何使用 ServiceNow 和 Elasticsearch 执行事件管理
欢迎回来！在上一篇博文中，我们设置了 ServiceNow 和 Elasticsearch 之间的双向通信。我们将大部分时间都花在了 ServiceNow 上，但从现在开始，我们将介绍在 Elasticsearch 和 Kibana 中如何操作。在按照本博文完成操作后，您将会看到这两个功能强大的应用程序协同工作，让事件管理变得轻而易举。至少，比您以前习惯的方式要容易得多！
与所有 Elasticsearch 项目一样，我们将使用适合我们需求的映射来创建索引。对于这个项目，我们需要可以保存以下数据的索引：
- ServiceNow 事件更新：这将存储所有从 ServiceNow 传入 Elasticsearch 的信息。ServiceNow 会将更新推送到这一索引。
- 便于使用的应用程序运行时间摘要：这将存储每个应用程序在线的总小时数。为了便于使用，可以将其视为中间数据状态。
- 应用程序事件摘要：这将存储每个应用程序发生的事件数、每个应用程序的运行时间，以及每个应用程序的当前 MTBF（平均故障间隔时间）。
后面两个索引是辅助索引，这样我们就不必在每次刷新 Canvas Workpad（我们将在第 3 部分中创建）时都要运行一大堆复杂的逻辑。我们将通过使用转换对它们进行连续更新。
创建上述三个索引
要创建这些索引，您可以按照下面的指导来操作。请注意，如果您使用的名称与下面使用的名称不同，则可能需要在 ServiceNow 设置中进行调整。
servicenow-incident-updates
按照最佳实践，我们将设置索引别名，然后设置索引生命周期管理 (ILM) 策略。此外，我们还将创建一个索引模板，以便将相同的映射应用于 ILM 策略之后创建的任何索引。一旦该索引中存储了 50GB 的数据，我们的 ILM 策略就将创建一个新索引，并在 1 年后将旧索引删除。我们还会使用索引别名，以便在创建新索引后可以轻松地指向该索引，而无需更新 ServiceNow 业务规则。
# 创建 ILM 策略
PUT _ilm/policy/servicenow-incident-updates-policy
{
"policy": {
"phases": {
"hot": {
"actions": {
"rollover": {
"max_size":"50GB"
}
}
},
"delete": {
"min_age":"360d",
"actions": {
"delete": {}
}
}
}
}
}
# 创建索引模板
PUT _template/servicenow-incident-updates-template
{
"index_patterns": [
"servicenow-incident-updates*"
],
"settings": {
"number_of_shards":1,
"index.lifecycle.name": "servicenow-incident-updates-policy",
"index.lifecycle.rollover_alias": "servicenow-incident-updates"
},
"mappings": {
"properties": {
"@timestamp": {
"type": "date",
"format": "yyyy-MM-dd HH:mm:ss"
},
"assignedTo": {
"type": "keyword"
},
"description": {
"type": "text",
"fields": {
"keyword": {
"type": "keyword",
"ignore_above":256
}
}
},
"incidentID": {
"type": "keyword"
},
"state": {
"type": "keyword"
},
"app_name": {
"type": "keyword"
},
"updatedDate": {
"type": "date",
"format": "yyyy-MM-dd HH:mm:ss"
},
"workNotes": {
"type": "text"
}
}
}
}
# 引导初始索引并创建别名
PUT servicenow-incident-updates-000001
{
"aliases": {
"servicenow-incident-updates": {
"is_write_index": true
}
}
}
app_uptime_summary 和 app_incident_summary
由于这两个索引都是以实体为中心的，因此不需要将 ILM 策略与其关联。这是因为在我们要监测的每个应用程序中将只有一个文档。要创建索引，请发出以下命令：
PUT app_uptime_summary
{
"mappings": {
"properties": {
"hours_online": {
"type": "float"
},
"app_name": {
"type": "keyword"
},
"up_count": {
"type": "long"
},
"last_updated": {
"type": "date"
}
}
}
}
PUT app_incident_summary
{
"mappings": {
"properties" : {
"hours_online" : {
"type" : "float"
},
"incident_count" : {
"type" : "integer"
},
"app_name" : {
"type" : "keyword"
},
"mtbf" : {
"type" : "float"
}
}
}
}
设置两个转换
转换是 Elastic Stack 中新添加的一项非常有用的功能。利用该功能，您可以将现有索引转换为以实体为中心的摘要，这对于执行分析和获取新见解都非常有用。转换有一个好处常常被人们所忽视，那就是其性能优势。例如，我们不需要通过查询和聚合（相当复杂的步骤）来计算每个应用程序的 MTBF，而是可以使用一个连续的转换来按照我们选择的节奏计算这个时间。例如，每分钟一次！如果不用转换，每个人在 Canvas Workpad 上进行每次刷新时，都会计算一次 MTBF。也就是说，如果有 50 个人在使用 Workpad，刷新间隔为 30 秒的话，我们每分钟就要运行 100 次昂贵的查询（听上去是不是有点过分）。虽然这在大多数情况下对 Elasticsearch 来说不是问题，但我想利用这个很棒的新功能让生活变得更简单一点。
我们将创建下面两个转换：
- calculate_uptime_hours_online_transform： 计算每个应用程序在线和响应的小时数。它利用 Heartbeat 中的运行时间数据来计算结果，并将结果数据存储在
app_uptime_summary索引中。
- app_incident_summary_transform：将 ServiceNow 数据与上述转换中的运行时间数据结合起来使用（是的，听起来有点像联接）。这一转换将获取运行时间数据，并计算出每个应用程序发生的事件数，然后代入在线小时数，最后根据这两个指标计算 MTBF。生成的索引叫作
app_incident_summary。
calculate_uptime_hours_online_transform
PUT _transform/calculate_uptime_hours_online_transform
{
"source": {
"index": [
"heartbeat*"
],
"query": {
"bool": {
"must": [
{
"match_phrase": {
"monitor.status": "up"
}
}
]
}
}
},
"dest": {
"index": "app_uptime_summary"
},
"sync": {
"time": {
"field": "@timestamp",
"delay":"60s"
}
},
"pivot": {
"group_by": {
"app_name": {
"terms": {
"field": "monitor.name"
}
}
},
"aggregations": {
"@timestamp": {
"max": {
"field": "@timestamp"
}
},
"up_count": {
"value_count": {
"field": "monitor.status"
}
},
"hours_online": {
"bucket_script": {
"buckets_path": {
"up_count": "up_count"
},
"script": "(params.up_count * 60.0) / 3600.0"
}
}
}
},
"description":"Calculate the hours online for each thing monitored by uptime"
}
app_incident_summary_transform
PUT _transform/app_incident_summary_transform
{
"source": {
"index": [
"app_uptime_summary",
"servicenow*"
]
},
"pivot": {
"group_by": {
"app_name": {
"terms": {
"field": "app_name"
}
}
},
"aggregations": {
"incident_count": {
"cardinality": {
"field": "incidentID"
}
},
"hours_online": {
"max": {
"field": "hours_online",
"missing":0
}
},
"mtbf": {
"bucket_script": {
"buckets_path": {
"hours_online": "hours_online",
"incident_count": "incident_count"
},
"script": "(float)params.hours_online / (float)params.incident_count"
}
}
}
},
"description":"Calculates the MTBF for apps by using the output from the calculate_uptime_hours_online transform",
"dest": {
"index": "app_incident_summary"
},
"sync": {
"time": {
"field": "@timestamp",
"delay":"1m"
}
}
}
现在确保下面的两个转换都在运行：
POST _transform/calculate_uptime_hours_online_transform/_start POST _transform/app_incident_summary_transform/_start
创建运行时间告警以在 ServiceNow 中创建工单
除了构建漂亮的 Canvas Workpad 之外，结束循环的最后一步是在 ServiceNow 中实际创建一个工单，前提是尚无针对此故障的工单。为此，我们将使用 Watcher 创建一个告警。此告警包含如下所示的步骤。就上下文而言，它每分钟运行一次。您可以在 Heartbeat 文档中看到所有运行时间字段。
1.检查在过去 5 分钟内有哪些应用程序发生中断
这很简单。我们将获得过去 5 分钟内（时间段筛选）按 monitor.name（术语聚合）分组的所有 down（术语筛选）的 Heartbeat 事件。monitor.name 字段遵循 Elastic Common Schema (ECS)，它将与应用程序名称字段中的值同义。这一切都是通过下面 Watcher 中的 down_check 输入来实现的。
2.获取每个应用程序的前 20 个工单，并获取每个工单的最新更新
这一步有些复杂。我们将搜索已采集的 ServiceNow 数据，这些数据是根据我们的 ServiceNow 业务规则自动采集的。existing_ticket_check 输入会使用多个聚合。首先是通过“apps”术语聚合将所有应用程序分组在一起。然后，对于每个应用，我们使用 incidents 术语聚合将 ServiceNow 工单事件 ID 分组在一起。最后，针对每个应用程序发现的每个事件，我们将使用按 @timestamp 字段排序的 top_hits 聚合来获取最新状态。
3.将两个数据源合并在一起，看看是否有工单需要创建
为此，我们使用脚本有效负载转换。简言之，它通过迭代 down_check 输出来检查中断原因，然后检查该特定应用程序是否具有未解决的工单。如果当前没有处理中、新建或挂起的工单，它会将应用程序添加到列表中，然后返回该列表并传递到行动阶段。
这一有效负载转换会在此过程中进行很多检查，以捕获我在下面概述的一些边缘情况，例如，如果应用之前没有任何事件历史记录，则创建一个工单。此转换的输出是一个由应用程序名称构成的数组。
4.如果是新的运行状态警报，则在 ServiceNow 中创建工单
我们使用 webhook 操作通过其 REST API 在 ServiceNow 中创建工单。为此，它使用 foreach 参数迭代上述数组中的应用程序名称，然后为每个应用程序运行 webhook 操作。只有当一个或多个应用程序需要工单时，它才会这样做。请确保为 ServiceNow 设置了正确的凭据和终端。
PUT _watcher/watch/e146d580-3de7-4a4c-a519-9598e47cbad1
{
"trigger": {
"schedule": {
"interval":"1m"
}
},
"input": {
"chain": {
"inputs": [
{
"down_check": {
"search": {
"request": {
"body": {
"query": {
"bool": {
"filter": [
{
"range": {
"@timestamp": {
"gte": "now-5m/m"
}
}
},
{
"term": {
"monitor.status": "down"
}
}
]
}
},
"size":0,
"aggs": {
"apps": {
"terms": {
"field": "monitor.name",
"size":100
}
}
}
},
"indices": [
"heartbeat-*"
],
"rest_total_hits_as_int": true,
"search_type": "query_then_fetch"
}
}
}
},
{
"existing_ticket_check": {
"search": {
"request": {
"body": {
"aggs": {
"apps": {
"aggs": {
"incidents": {
"aggs": {
"state": {
"top_hits": {
"_source": "state",
"size":1,
"sort": [
{
"@timestamp": {
"order": "desc"
}
}
]
}
}
},
"terms": {
"field": "incidentID",
"order": {
"_key": "desc"
},
"size":1
}
}
},
"terms": {
"field": "app_name",
"size":100
}
}
},
"size":0
},
"indices": [
"servicenow*"
],
"rest_total_hits_as_int": true,
"search_type": "query_then_fetch"
}
}
}
}
]
}
},
"transform": {
"script": """
List appsNeedingTicket = new ArrayList();
for (app_heartbeat in ctx.payload.down_check.aggregations.apps.buckets) {
boolean appFound = false;
List appsWithTickets = ctx.payload.existing_ticket_check.aggregations.apps.buckets;
if (appsWithTickets.size() == 0) {
appsNeedingTicket.add(app_heartbeat.key);
continue;
}
for (app in appsWithTickets) {
boolean needsTicket = false;
if (app.key == app_heartbeat.key) {
appFound = true;
for (incident in app.incidents.buckets) {
String state = incident.state.hits.hits[0]._source.state;
if (state == 'Resolved' || state == 'Closed' || state == 'Canceled') {
appsNeedingTicket.add(app.key);
}
}
}
}
if (appFound == false) {
appsNeedingTicket.add(app_heartbeat.key);
}
}
return appsNeedingTicket;
"""
},
"actions": {
"submit_servicenow_ticket": {
"condition": {
"script": {
"source": "return ctx.payload._value.size() > 0"
}
},
"foreach": "ctx.payload._value",
"max_iterations":500,
"webhook": {
"scheme": "https",
"host": "dev94721.service-now.com",
"port":443,
"method": "post",
"path": "/api/now/table/incident",
"params": {},
"headers": {
"Accept": "application/json",
"Content-Type": "application/json"
},
"auth": {
"basic": {
"username": "admin",
"password":"REDACTED"
}
},
"body": "{'description':'{{ctx.payload._value}} Offline','short_description':'{{ctx.payload._value}} Offline','caller_id': 'elastic_watcher','impact':'1','urgency':'1', 'u_application':'{{ctx.payload._value}}'}",
"read_timeout_millis":30000
}
}
},
"metadata": {
"xpack": {
"type": "json"
},
"name":"ApplicationDowntime Watcher"
}
}
结论
我们总结一下本项目的第二部分。在这一部分，我们为 Elasticsearch 创建了一些常规配置，以便创建索引和 ILM 策略。另外，我们还创建了转换，用于计算每个应用程序的运行时间和 MTBF，以及用于监测我们运行时间数据的 Watcher。这里要特别注意的是，如果 Watcher 发现某个应用出现故障，它会首先检查是否存在工单，如果没有，则会在 ServiceNow 中创建一个。
不妨按照上述步骤操作一下？最简单的方法是使用 Elastic Cloud。登录 Elastic Cloud 控制台或注册免费试用 14 天。您可以通过现有的 ServiceNow 实例按照上述步骤进行，或快速部署个人开发人员实例。
此外，如果想要搜索 ServiceNow 数据以及 GitHub、Google Drive 等其他资源，Elastic Workplace Search 提供有预先构建的 ServiceNow 连接器。Workplace Search 可为您的团队提供统一的搜索体验，以及所有内容来源的相关结果。Elastic Cloud 试用版中也包含 Workplace Search。
现在已万事俱备，但视觉上还不够吸引人。因此，在这个项目的第三部分（也是最后一部分），我们将介绍如何使用 Canvas 创建外观精美的前端来呈现所有这些数据，以及如何计算提到的平均确认时间 (MTTA)、平均解决时间 (MTTR) 等其他指标。第三部分，不见不散！