Elasticsearch 7.16 引入了一個新的豐富策略：range， range 策略允許將傳入檔案中的數字、日期或 IP 地址與豐富索引中相同型別的范圍相匹配，能夠與 IP 范圍進行匹配在安全用例中特別有用，其中額外的元資料可用于進一步細化檢測規則，由于我們已經在檔案中添加了一個使用 IP 范圍的示例，因此我們將在此處使用 date_range 型別進行示例，

在之前我的文章 “Elasticsearch：enrich processor （7.5發行版新功能）” 已經詳細描述了 geo_match 及 match 的豐富策略，詳細使用，請閱讀那篇文章，

我們虛構的例子：事件和待命時間表

假設我們有許多待命（隨傳隨到）時間表，我們希望將它們添加到 Elasticsearch，以便每個連續班次都是一個檔案，讓我們介紹一下我們虛構的測驗用例：Bob、Alice、Dan、Matt 和 Lizzie，

Bob 喜歡朝九晚六的作業，中午午休一小時，我們可以像這樣添加他在 11 月 29 日星期一的日程安排：

PUT /on_call_schedules
{
  "mappings": {
    "properties": {
      "shift": { 
        "type": "date_range", 
        "format": "yyyy-MM-dd HH:mm:ss||yyyy-MM-dd||epoch_millis"
      }
    }
  }
}
POST on_call_schedules/_doc
{
  "engineer" : { 
    "name" : "Bob"
  },
  "shift" : {
    "gte" : "2021-11-29 08:00:00", 
    "lte" : "2021-11-29 12:00:00"
  }
}
POST on_call_schedules/_doc
{
  "engineer" : { 
    "name" : "Bob"
  },
  "shift" : {
    "gte" : "2021-11-29 13:00:00", 
    "lte" : "2021-11-29 17:00:00"
  }
}

對于其他工程師，他們的故事如下：Alice 有類似的時間表，但從 13:00 開始吃午飯，Dan 和 Matt 在不同的時區，Matt 作業半天，0:00 - 4:00，Dan 作業 3 :00-8:00 午休至 9:00，12:00 下班，Lizzie 晚上 16:00 作業至午夜 20:00 休息，

填充索引的其余請求如下所示（為簡潔起見，將物件放在同一行）：

POST on_call_schedules/_doc
{
  "engineer" : {  "name" : "Alice"  },
  "shift" : { "gte" : "2021-11-29 09:00:00", "lte" : "2021-11-29 13:00:00"  }
}
POST on_call_schedules/_doc
{
  "engineer" : {  "name" : "Alice"  }, 
  "shift" : {  "gte" : "2021-11-29 14:00:00", "lte" : "2021-11-29 18:00:00"  }
}
POST on_call_schedules/_doc
{
  "engineer" : {  "name" : "Dan"  },
  "shift" : { "gte" : "2021-11-29 03:00:00", "lte" : "2021-11-29 08:00:00"  }
}
POST on_call_schedules/_doc
{
  "engineer" : {  "name" : "Dan"  }, 
  "shift" : {  "gte" : "2021-11-29 09:00:00", "lte" : "2021-11-29 12:00:00"  }
}
POST on_call_schedules/_doc
{
  "engineer" : {  "name" : "Matt"  }, 
  "shift" : {  "gte" : "2021-11-29 00:00:00", "lte" : "2021-11-29 04:00:00"  }
}
POST on_call_schedules/_doc
{
  "engineer" : {  "name" : "Lizzie"  }, 
  "shift" : {  "gte" : "2021-11-29 16:00:00", "lte" : "2021-11-29 20:00:00"  }
}
POST on_call_schedules/_doc
{
  "engineer" : {  "name" : "Lizzie"  }, 
  "shift" : {  "gte" : "2021-11-29 21:00:00", "lte" : "2021-11-30 00:00:00"  }
}

現在我們有了一個包含所有時間表的索引，我們可以繼續創建一個豐富的策略，以便在我們提供日期時通過將其與包含 date_range 的 shift 欄位進行匹配來查找待命工程師：

PUT /_enrich/policy/add-oncall-engineers-policy
{
  "range": {
    "indices": "on_call_schedules",
    "match_field": "shift",
    "enrich_fields": ["engineer.name"]
  }
}

上面的 range 策略的意思是：當 shift 欄位的值是在輸入檔案中匹配的時間范圍內，那么 engineer.name 的值將被豐富進來，有了策略，我們可以執行它，以便可以準備源索引中的資料以供使用：

POST /_enrich/policy/add-oncall-engineers-policy/_execute?wait_for_completion=true

現在我們將創建一個攝入管道，以便我們可以處理傳入的檔案：

PUT /_ingest/pipeline/engineer_lookup
{
  "processors" : [
    {
      "enrich" : {
        "description": "Add on-call engineer based on 'date'",
        "policy_name": "add-oncall-engineers-policy",
        "field" : "@timestamp",
        "target_field": "oncall_engineers",
        "max_matches": "25"
      }
    }
  ]
}

在這一點上，我們都準備好記錄一些事件，并用預定的工程師來豐富它們，這個攝入管道的意思是：當一個檔案 @timestamp 的時間落在策略中的所定義的索引 on_call_schedules 中的 shift 范圍內，那么相應的 engineer.name 將被豐富于新的檔案中，

讓我們使用一個事件來測驗

讓我們記錄一下 Dan 在早上 6:12 處理的事件一：

PUT /incidents/_doc/incident1?pipeline=engineer_lookup
{
  "@timestamp": "2021-11-29 06:12:33",
  "severity": "high",
  "handled_by": "Dan"
}

在上面，我們輸入一個事件，我們可以看到它里面還有一個欄位叫做 @timestamp，我們可以用這個值和之前輸入的檔案中的 shift 進行比較，如果這個 @timestamp 的值的范圍落于其中的一個檔案時間范圍，那么我們可以把它的 name 這欄位豐富過來，并寫入到 target_field 中，

當我們檢索檔案時，我們可以看到 Dan 是唯一安排好的人：

GET /incidents/_doc/incident1

上面的命令回應是：

{
  "_index" : "incidents",
  "_type" : "_doc",
  "_id" : "incident1",
  "_version" : 8,
  "_seq_no" : 7,
  "_primary_term" : 1,
  "found" : true,
  "_source" : {
    "severity" : "high",
    "@timestamp" : "2021-11-29 06:12:33",
    "handled_by" : "Dan",
    "oncall_engineers" : [
      {
        "shift" : {
          "gte" : "2021-11-29 03:00:00",
          "lte" : "2021-11-29 08:00:00"
        },
        "engineer" : {
          "name" : "Dan"
        }
      }
    ]
  }
}

在上面我們可以看到被豐富的欄位 engineer.name 是 Dan，也就是如下的檔案：

POST on_call_schedules/_doc
{
  "engineer" : {  "name" : "Dan"  },
  "shift" : { "gte" : "2021-11-29 03:00:00", "lte" : "2021-11-29 08:00:00"  }
}

中的 shift 時間范圍包含之前的 @timestamp 時間 2021-11-29 06:12:33，從而 engineer.name 這個欄位被添加到新的檔案中，

更多事件

讓我們再記錄三個事件，兩個由 Dan 在 11:12 和 14:08 處理，一個由 Alice 在 16:12 處理：

PUT /incidents/_doc/incident2?pipeline=engineer_lookup
{
  "@timestamp": "2021-11-29 11:12:52",  "severity": "high",  "handled_by": "Dan"
}

PUT /incidents/_doc/incident3?pipeline=engineer_lookup
{
  "@timestamp": "2021-11-29 14:08:06",   "severity": "high",  "handled_by": "Dan"
}

PUT /incidents/_doc/incident4?pipeline=engineer_lookup&refresh=wait_for
{
  "@timestamp": "2021-11-29 16:12:16",  "severity": "high",  "handled_by": "Alice"
}

根據我們的日程安排，當第二次事故發生時，我們應該有 3 名工程師隨叫隨到，讓我們驗證一下：

GET /incidents/_doc/incident2

回應是：

{
  "_index" : "incidents",
  "_type" : "_doc",
  "_id" : "incident2",
  "_version" : 1,
  "_seq_no" : 8,
  "_primary_term" : 1,
  "found" : true,
  "_source" : {
    "severity" : "high",
    "@timestamp" : "2021-11-29 11:12:52",
    "handled_by" : "Dan",
    "oncall_engineers" : [
      {
        "shift" : {
          "gte" : "2021-11-29 08:00:00",
          "lte" : "2021-11-29 12:00:00"
        },
        "engineer" : {
          "name" : "Bob"
        }
      },
      {
        "shift" : {
          "gte" : "2021-11-29 09:00:00",
          "lte" : "2021-11-29 13:00:00"
        },
        "engineer" : {
          "name" : "Alice"
        }
      },
      {
        "shift" : {
          "gte" : "2021-11-29 09:00:00",
          "lte" : "2021-11-29 12:00:00"
        },
        "engineer" : {
          "name" : "Dan"
        }
      }
    ]
  }
}

如我們所見，shift 匹配正確，

事件三有點奇怪，Dan 處理了這件事，但沒有被安排做這件事！ Dan 作業太辛苦了，與其直接檢索事件，不如搜索 Dan 在未安排時間時處理的所有事件：

GET incidents/_search
{
  "query": {
    "bool": {
      "must_not": [
        {
          "term": {
            "oncall_engineers.engineer.name.keyword": "Dan"
          }
        }
      ], 
      "filter": [
        {
          "term": {
            "handled_by.keyword": "Dan"
          }
        }
      ]
    }
  }
}

事實上，我們將事件三作為命中：

{
  "took" : 0,
  "timed_out" : false,
  "_shards" : {
    "total" : 1,
    "successful" : 1,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : {
      "value" : 1,
      "relation" : "eq"
    },
    "max_score" : 0.0,
    "hits" : [
      {
        "_index" : "incidents",
        "_type" : "_doc",
        "_id" : "incident3",
        "_score" : 0.0,
        "_source" : {
          "severity" : "high",
          "@timestamp" : "2021-11-29 14:08:06",
          "handled_by" : "Dan",
          "oncall_engineers" : [
            {
              "shift" : {
                "gte" : "2021-11-29 13:00:00",
                "lte" : "2021-11-29 17:00:00"
              },
              "engineer" : {
                "name" : "Bob"
              }
            },
            {
              "shift" : {
                "gte" : "2021-11-29 14:00:00",
                "lte" : "2021-11-29 18:00:00"
              },
              "engineer" : {
                "name" : "Alice"
              }
            }
          ]
        }
      }
    ]
  }
}

除了搜索之外，我們還可以運行聚合，讓我們按照每個工程師處理事件的頻率以及他們待命的事件數量來分解事情：

GET incidents/_search
{
  "aggs": {
    "on_call_per_incident": {
      "terms": {
        "field": "oncall_engineers.engineer.name.keyword",
        "size": 10
      }
    },
    "handled_incidents": {
      "terms": {
        "field": "handled_by.keyword",
        "size": 10
      }
    }
  }
  , "size": 0
}

回應為：

{
  "took" : 2,
  "timed_out" : false,
  "_shards" : {
    "total" : 1,
    "successful" : 1,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : {
      "value" : 4,
      "relation" : "eq"
    },
    "max_score" : null,
    "hits" : [ ]
  },
  "aggregations" : {
    "handled_incidents" : {
      "doc_count_error_upper_bound" : 0,
      "sum_other_doc_count" : 0,
      "buckets" : [
        {
          "key" : "Dan",
          "doc_count" : 3
        },
        {
          "key" : "Alice",
          "doc_count" : 1
        }
      ]
    },
    "on_call_per_incident" : {
      "doc_count_error_upper_bound" : 0,
      "sum_other_doc_count" : 0,
      "buckets" : [
        {
          "key" : "Alice",
          "doc_count" : 3
        },
        {
          "key" : "Bob",
          "doc_count" : 3
        },
        {
          "key" : "Dan",
          "doc_count" : 2
        },
        {
          "key" : "Lizzie",
          "doc_count" : 1
        }
      ]
    }
  }
}

作為最后一個例子，讓我們也分解一下每個工程師以及當該工程師處理事件時誰也在待命：

GET incidents/_search
{
  "aggs": {
    "incidents_handled_by": {
      "terms": {
        "field": "handled_by.keyword",
        "size": 10
      }
      , "aggs": {
        "supporting": {
          "terms": {
            "field": "oncall_engineers.engineer.name.keyword",
            "size": 10
          }
        }
      }
    }
  }
  , "size": 0
}

更多例子

以下示例創建一個 range 豐富策略，該策略根據 IP 地址將描述性網路名稱和負責部門添加到傳入檔案，然后，它將豐富策略添加到攝取管道中的處理器，

使用帶有適當映射的 create index API 創建源索引，

PUT /networks
{
  "mappings": {
    "properties": {
      "range": { "type": "ip_range" },
      "name": { "type": "keyword" },
      "department": { "type": "keyword" }
    }
  }
}

以下索引 API 請求將新檔案索引到該索引，

PUT /networks/_doc/1?refresh=wait_for
{
  "range": "10.100.0.0/16",
  "name": "production",
  "department": "OPS"
}

使用創建豐富策略 API 創建具有 range 策略型別的豐富策略，該政策必須包括：

一個或多個源索引
match_field，源索引中用于匹配傳入檔案的欄位
豐富你想要附加到傳入檔案的源索引中的欄位

由于我們計劃根據 IP 地址豐富檔案，因此策略的 match_field 必須是 ip_range 欄位，

PUT /_enrich/policy/networks-policy
{
  "range": {
    "indices": "networks",
    "match_field": "range",
    "enrich_fields": ["name", "department"]
  }
}

使用 execute enrich policy API 為策略創建豐富索引，

POST /_enrich/policy/networks-policy/_execute

使用創建或更新管道 API 創建攝取管道，在管道中，添加一個豐富的處理器，其中包括：

你的豐富策略，
用于匹配來自豐富索引的檔案的傳入檔案的 field，
target_field 用于存盤傳入檔案的附加豐富資料，此欄位包含你的豐富策略中指定的 match_field 和enrich_fields，

PUT /_ingest/pipeline/networks_lookup
{
  "processors" : [
    {
      "enrich" : {
        "description": "Add 'network' data based on 'ip'",
        "policy_name": "networks-policy",
        "field" : "ip",
        "target_field": "network",
        "max_matches": "10"
      }
    }
  ]
}

使用攝入管道來索引檔案，傳入的檔案應包括你的豐富處理器中指定的欄位，

PUT /my-index-000001/_doc/my_id?pipeline=networks_lookup
{
  "ip": "10.100.34.1"
}

在上面，由于 "ip": "10.100.34.1" 這個 IP 地址是在 10.100.0.0/16 范圍內，所以當我們攝入資料時 networks_lookup 攝入管道會起作用，它會把豐富的內容添加到 target_field 欄位里，

要驗證豐富處理器匹配并附加了適當的欄位資料，請使用 get API 查看索引檔案，

GET /my-index-000001/_doc/my_id

API 回傳以下回應：

{
  "_index" : "my-index-000001",
  "_type" : "_doc",
  "_id" : "my_id",
  "_version" : 1,
  "_seq_no" : 1,
  "_primary_term" : 8,
  "found" : true,
  "_source" : {
    "ip" : "10.100.34.1",
    "network" : [
      {
        "name" : "production",
        "range" : "10.100.0.0/16",
        "department" : "OPS"
      }
    ]
  }
}

從上面我們可以看出來，除了我們已經輸入的欄位 ip 以外，我們還可以看到被豐富的欄位 network 作為起 target_field，在它里面含有我們之前在 networks-policy 定義的 name 及 department 欄位，

Range豐富政策的好處

Range 豐富政策開辟了新的匹配選項和豐富檔案的新方法，在這篇博文中，我們展示了一家虛構的公司，其中包含預定的工程師和記錄的事件，使用 Elasticsearch 的功能，我們可以記錄事件，豐富工程師安排的事件，并分析資料，

轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/421428.html

標籤：其他

上一篇：2021年影響會計人員的十大資訊技術榜單出爐，RPA再次登榜

下一篇：ElasticSearch對標Mysql，誰能拔得頭籌？

Elasticsearch 的新 range 豐富策略使背景關系資料分析更上一層樓 - 7.16

我們虛構的例子：事件和待命時間表

讓我們使用一個事件來測驗

更多事件

更多例子

Range豐富政策的好處