使用Kapacitor的自定义异常检测
每个人都有自己独特的异常检测算法，因此我们构建了Kapacitor，以便与适合您领域的算法轻松集成。Kapacitor称这些自定义算法为用户定义的函数（UDF）。本指南将介绍在Kapacitor中编写和使用您自己的UDF所需的步骤。
如果您还没有这样做，我们建议在继续之前遵循入门指南来了解Kapacitor。
三维打印
如果您拥有或最近购买了3D打印机，您可能知道3D打印要求环境在特定的温度下以确保打印质量。打印也可能需要很长时间（有些可能超过24小时），因此您不能一直关注温度曲线来确保打印顺利进行。此外，如果打印早期出现问题，您希望确保停止它，以便可以重新启动，而不是浪费材料继续一个不好的打印。
由于3D打印的物理限制，打印机软件通常设计为保持温度在某些容差范围内。为了方便讨论，假设您不信任软件执行其工作（或想要自己创建），并希望在温度达到异常水平时收到警报。
在3D打印时有三种温度：
- 热端的温度（在打印之前塑料被熔化的地方）。
- 床的温度（零件正在打印的地方）。
- 环境空气的温度（打印机周围的空气）。
这三种温度都会影响打印的质量（有些比其他的更重要），但我们想确保跟踪它们全部。
为了保持我们的异常检测算法简单，我们将为每个接收到的数据窗口计算一个
p-value，然后发出一个具有该p-value的单一数据点。为了计算p-value，我们将使用
Welch’s t-test。对于
一个零假设，我们将声明一个新窗口来自于与历史窗口相同的人群。如果p-value足够低
，我们可以拒绝零假设并得出结论，窗口必须来自于与历史数据人群不同的某种情况，或
一个异常。这是一种过于简化的方法，但我们正在学习如何编写UDF，而不是统计学。
编写用户自定义函数 (UDF)
现在我们已经初步了解了我们想要做什么，让我们理解一下
Kapacitor 如何与我们的进程进行通信。从 UDF
README 中，我们了解到 Kapacitor 将会生成一个叫做 agent 的进程。
agent 负责描述它有哪些选项，然后用一组选项初始化自己。当 UDF 接收到数据时，
agent 执行其计算，然后将结果数据返回给 Kapacitor。所有这些通信都通过
STDIN 和 STDOUT 使用协议缓冲区进行。截止到本文撰写时，Kapacitor 已经在 Go 和 Python 中实现了
处理通信细节的代理，并提供了进行实际工作的接口。对于本指南，我们将使用 Python 代理。
处理程序接口
这是代理的Python处理程序接口：
# The Agent calls the appropriate methods on the Handler as requests are read off STDIN.
#
# Throwing an exception will cause the Agent to stop and an ErrorResponse to be sent.
# Some *Response objects (like SnapshotResponse) allow for returning their own error within the object itself.
# These types of errors will not stop the Agent and Kapacitor will deal with them appropriately.
#
# The Handler is called from a single thread, meaning methods will not be called concurrently.
#
# To write Points/Batches back to the Agent/Kapacitor use the Agent.write_response method, which is thread safe.
class Handler(object):
def info(self):
pass
def init(self, init_req):
pass
def snapshot(self):
pass
def restore(self, restore_req):
pass
def begin_batch(self):
pass
def point(self):
pass
def end_batch(self, end_req):
pass
信息方法
让我们从info方法开始。当Kapacitor启动时，它会调用info并期待返回一些关于这个UDF如何执行的信息。具体来说，Kapacitor期待UDF想要和提供的边缘类型。
记住: 在Kapacitor中，数据是以流或批量方式传输的，因此UDF必须声明它所期望的内容。
此外，UDF可以接受某些选项，以便它们可以单独配置。info响应可以包含一组选项、它们的名称和期望的参数。
对于我们的示例 UDF，我们需要知道三件事：
- 要操作的字段。
- 要保留的历史窗口的大小。
- 正在使用的显著性水平或
alpha。
下面是我们处理程序的info方法的实现，该方法定义了可用的边缘类型和选项：
...
def info(self):
"""
Respond with which type of edges we want/provide and any options we have.
"""
response = udf_pb2.Response()
# We will consume batch edges aka windows of data.
response.info.wants = udf_pb2.BATCH
# We will produce single points of data aka stream.
response.info.provides = udf_pb2.STREAM
# Here we can define options for the UDF.
# Define which field we should process.
response.info.options['field'].valueTypes.append(udf_pb2.STRING)
# Since we will be computing a moving average let's make the size configurable.
# Define an option 'size' that takes one integer argument.
response.info.options['size'].valueTypes.append(udf_pb2.INT)
# We need to know the alpha level so that we can ignore bad windows.
# Define an option 'alpha' that takes one double valued argument.
response.info.options['alpha'].valueTypes.append(udf_pb2.DOUBLE)
return response
...
当Kapacitor启动时，它会生成我们的UDF进程并请求info数据，然后关闭该进程。Kapacitor会记住每个UDF的信息。通过这种方式，Kapacitor可以在任务内部执行之前理解特定UDF的可用选项。
初始化方法
接下来让我们实现 init 方法，该方法在任务开始执行时被调用。 init 方法接收一个选择的选项列表，然后用于适当地配置处理程序。作为响应，我们指示 init 请求是否成功，如果没有，任何选项无效时的错误消息。
...
def init(self, init_req):
"""
Given a list of options initialize this instance of the handler
"""
success = True
msg = ''
size = 0
for opt in init_req.options:
if opt.name == 'field':
self._field = opt.values[0].stringValue
elif opt.name == 'size':
size = opt.values[0].intValue
elif opt.name == 'alpha':
self._alpha = opt.values[0].doubleValue
if size <= 1:
success = False
msg += ' must supply window size > 1'
if self._field == '':
success = False
msg += ' must supply a field name'
if self._alpha == 0:
success = False
msg += ' must supply an alpha value'
# Initialize our historical window
# We will define MovingStats in the next step
self._history = MovingStats(size)
response = udf_pb2.Response()
response.init.success = success
response.init.error = msg[1:]
return response
...
当任务开始时，Kapacitor 为 UDF 启动一个新进程并调用 init，传递来自 TICKscript 的任何指定选项。初始化完成后，进程将保持运行，Kapacitor 将开始在数据到达时发送数据。
批处理和点方法
我们的任务需要一个 batch 边缘，这意味着它期望以批次或窗口的形式接收数据。要将一批数据发送到 UDF 过程，Kapacitor 首先调用 begin_batch 方法，这表明所有后续点都属于一个批次。一旦批次完成，end_batch 方法将被调用，并附带一些关于批次的元数据。
从高层次来看，以下是我们的 UDF 代码在每个 begin_batch、point 和 end_batch 调用时将会执行的操作：
begin_batch: 标记新批次的开始并为其初始化一个结构point: 存储点end_batch: 执行t-test然后更新历史数据
完整的UDF脚本
接下来是完整的UDF实现，包括我们的 info,
init 和批处理方法（以及我们需要的其他所有内容）。
from kapacitor.udf.agent import Agent, Handler
from scipy import stats
import math
from kapacitor.udf import udf_pb2
import sys
class TTestHandler(Handler):
"""
Keep a rolling window of historically normal data
When a new window arrives use a two-sided t-test to determine
if the new window is statistically significantly different.
"""
def __init__(self, agent):
self._agent = agent
self._field = ''
self._history = None
self._batch = None
self._alpha = 0.0
def info(self):
"""
Respond with which type of edges we want/provide and any options we have.
"""
response = udf_pb2.Response()
# We will consume batch edges aka windows of data.
response.info.wants = udf_pb2.BATCH
# We will produce single points of data aka stream.
response.info.provides = udf_pb2.STREAM
# Here we can define options for the UDF.
# Define which field we should process
response.info.options['field'].valueTypes.append(udf_pb2.STRING)
# Since we will be computing a moving average let's make the size configurable.
# Define an option 'size' that takes one integer argument.
response.info.options['size'].valueTypes.append(udf_pb2.INT)
# We need to know the alpha level so that we can ignore bad windows
# Define an option 'alpha' that takes one double argument.
response.info.options['alpha'].valueTypes.append(udf_pb2.DOUBLE)
return response
def init(self, init_req):
"""
Given a list of options initialize this instance of the handler
"""
success = True
msg = ''
size = 0
for opt in init_req.options:
if opt.name == 'field':
self._field = opt.values[0].stringValue
elif opt.name == 'size':
size = opt.values[0].intValue
elif opt.name == 'alpha':
self._alpha = opt.values[0].doubleValue
if size <= 1:
success = False
msg += ' must supply window size > 1'
if self._field == '':
success = False
msg += ' must supply a field name'
if self._alpha == 0:
success = False
msg += ' must supply an alpha value'
# Initialize our historical window
self._history = MovingStats(size)
response = udf_pb2.Response()
response.init.success = success
response.init.error = msg[1:]
return response
def begin_batch(self, begin_req):
# create new window for batch
self._batch = MovingStats(-1)
def point(self, point):
self._batch.update(point.fieldsDouble[self._field])
def end_batch(self, batch_meta):
pvalue = 1.0
if self._history.n != 0:
# Perform Welch's t test
t, pvalue = stats.ttest_ind_from_stats(
self._history.mean, self._history.stddev(), self._history.n,
self._batch.mean, self._batch.stddev(), self._batch.n,
equal_var=False)
# Send pvalue point back to Kapacitor
response = udf_pb2.Response()
response.point.time = batch_meta.tmax
response.point.name = batch_meta.name
response.point.group = batch_meta.group
response.point.tags.update(batch_meta.tags)
response.point.fieldsDouble["t"] = t
response.point.fieldsDouble["pvalue"] = pvalue
self._agent.write_response(response)
# Update historical stats with batch, but only if it was normal.
if pvalue > self._alpha:
for value in self._batch._window:
self._history.update(value)
class MovingStats(object):
"""
Calculate the moving mean and variance of a window.
Uses Welford's Algorithm.
"""
def __init__(self, size):
"""
Create new MovingStats object.
Size can be -1, infinite size or > 1 meaning static size
"""
self.size = size
if not (self.size == -1 or self.size > 1):
raise Exception("size must be -1 or > 1")
self._window = []
self.n = 0.0
self.mean = 0.0
self._s = 0.0
def stddev(self):
"""
Return the standard deviation
"""
if self.n == 1:
return 0.0
return math.sqrt(self._s / (self.n - 1))
def update(self, value):
# update stats for new value
self.n += 1.0
diff = (value - self.mean)
self.mean += diff / self.n
self._s += diff * (value - self.mean)
if self.n == self.size + 1:
# update stats for removing old value
old = self._window.pop(0)
oldM = (self.n * self.mean - old)/(self.n - 1)
self._s -= (old - self.mean) * (old - oldM)
self.mean = oldM
self.n -= 1
self._window.append(value)
if __name__ == '__main__':
# Create an agent
agent = Agent()
# Create a handler and pass it an agent so it can write points
h = TTestHandler(agent)
# Set the handler on the agent
agent.handler = h
# Anything printed to STDERR from a UDF process gets captured into the Kapacitor logs.
print >> sys.stderr, "Starting agent for TTestHandler"
agent.start()
agent.wait()
print >> sys.stderr, "Agent finished"
这很多，但现在我们准备好配置Kapacitor来运行我们的代码。创建一个临时目录以便继续该指南的其余部分：
mkdir /tmp/kapacitor_udf
cd /tmp/kapacitor_udf
将上述UDF python脚本保存到 /tmp/kapacitor_udf/ttest.py。
为我们的UDF配置Kapacitor
将此代码段添加到您的Kapacitor配置文件中（通常位于 /etc/kapacitor/kapacitor.conf）：
[udf]
[udf.functions]
[udf.functions.tTest]
# Run python
prog = "/usr/bin/python2"
# Pass args to python
# -u for unbuffered STDIN and STDOUT
# and the path to the script
args = ["-u", "/tmp/kapacitor_udf/ttest.py"]
# If the python process is unresponsive for 10s kill it
timeout = "10s"
# Define env vars for the process, in this case the PYTHONPATH
[udf.functions.tTest.env]
PYTHONPATH = "/tmp/kapacitor_udf/kapacitor/udf/agent/py"
在配置中，我们调用了函数 tTest。这也是我们在 TICKscript 中引用它的方式。
请注意，我们的Python脚本导入了Agent对象，并且我们在配置中设置了PYTHONPATH。将Kapacitor源代码克隆到临时目录，以便我们可以将PYTHONPATH指向必要的python代码。这通常是多余的，因为它只有两个Python文件，但这使得跟踪变得简单：
git clone https://github.com/influxdata/kapacitor.git /tmp/kapacitor_udf/kapacitor
使用UDF运行Kapacitor
重启Kapacitor守护进程以确保一切配置正确：
service kapacitor restart
检查日志 (/var/log/kapacitor/) 以确保您看到
正在监听信号 行，并且没有发生错误。如果您
没有看到该行，那是因为 UDF 进程已经挂起并且没有
响应。它应该在超时后被终止，所以请给它一些时间
以正确停止。一旦停止，您可以修复任何错误并再次尝试。
TICK脚本
如果一切都正确启动了，那么是时候编写我们的 TICKscript 来使用 tTest UDF 方法：
dbrp "printer"."autogen"
// This TICKscript monitors the three temperatures for a 3d printing job,
// and triggers alerts if the temperatures start to experience abnormal behavior.
// Define our desired significance level.
var alpha = 0.001
// Select the temperatures measurements
var data = stream
|from()
.measurement('temperatures')
|window()
.period(5m)
.every(5m)
data
//Run our tTest UDF on the hotend temperature
@tTest()
// specify the hotend field
.field('hotend')
// Keep a 1h rolling window
.size(3600)
// pass in the alpha value
.alpha(alpha)
|alert()
.id('hotend')
.crit(lambda: "pvalue" < alpha)
.log('/tmp/kapacitor_udf/hotend_failure.log')
// Do the same for the bed and air temperature.
data
@tTest()
.field('bed')
.size(3600)
.alpha(alpha)
|alert()
.id('bed')
.crit(lambda: "pvalue" < alpha)
.log('/tmp/kapacitor_udf/bed_failure.log')
data
@tTest()
.field('air')
.size(3600)
.alpha(alpha)
|alert()
.id('air')
.crit(lambda: "pvalue" < alpha)
.log('/tmp/kapacitor_udf/air_failure.log')
请注意，我们已经调用了 tTest 三次。这意味着
Kapacitor 将生成三个不同的 Python 进程，并将各自的初始化选项传递给每一个。
将此脚本保存为 /tmp/kapacitor_udf/print_temps.tick 并定义 Kapacitor 任务：
kapacitor define print_temps -tick print_temps.tick
生成测试数据
为了模拟我们的打印机进行测试，我们将编写一个简单的Python脚本来生成温度。这个脚本生成的温度是围绕目标温度正态分布的随机温度。在指定时间，温度的变化和偏移会发生变化，从而产生异常。
在这里不要过于担心细节。使用真实数据来测试我们的TICKscript和UDF会更好，但这样更快（并且比3D打印机便宜得多）。
#!/usr/bin/python2
from numpy import random
from datetime import timedelta, datetime
import sys
import time
import requests
# Target temperatures in C
hotend_t = 220
bed_t = 90
air_t = 70
# Connection info
write_url = 'http://localhost:9092/write?db=printer&rp=autogen&precision=s'
measurement = 'temperatures'
def temp(target, sigma):
"""
Pick a random temperature from a normal distribution
centered on target temperature.
"""
return random.normal(target, sigma)
def main():
hotend_sigma = 0
bed_sigma = 0
air_sigma = 0
hotend_offset = 0
bed_offset = 0
air_offset = 0
# Define some anomalies by changing sigma at certain times
# list of sigma values to start at a specified iteration
hotend_anomalies =[
(0, 0.5, 0), # normal sigma
(3600, 3.0, -1.5), # at one hour the hotend goes bad
(3900, 0.5, 0), # 5 minutes later recovers
]
bed_anomalies =[
(0, 1.0, 0), # normal sigma
(28800, 5.0, 2.0), # at 8 hours the bed goes bad
(29700, 1.0, 0), # 15 minutes later recovers
]
air_anomalies = [
(0, 3.0, 0), # normal sigma
(10800, 5.0, 0), # at 3 hours air starts to fluctuate more
(43200, 15.0, -5.0), # at 12 hours air goes really bad
(45000, 5.0, 0), # 30 minutes later recovers
(72000, 3.0, 0), # at 20 hours goes back to normal
]
# Start from 2016-01-01 00:00:00 UTC
# This makes it easy to reason about the data later
now = datetime(2016, 1, 1)
second = timedelta(seconds=1)
epoch = datetime(1970,1,1)
# 24 hours of temperatures once per second
points = []
for i in range(60*60*24+2):
# update sigma values
if len(hotend_anomalies) > 0 and i == hotend_anomalies[0][0]:
hotend_sigma = hotend_anomalies[0][1]
hotend_offset = hotend_anomalies[0][2]
hotend_anomalies = hotend_anomalies[1:]
if len(bed_anomalies) > 0 and i == bed_anomalies[0][0]:
bed_sigma = bed_anomalies[0][1]
bed_offset = bed_anomalies[0][2]
bed_anomalies = bed_anomalies[1:]
if len(air_anomalies) > 0 and i == air_anomalies[0][0]:
air_sigma = air_anomalies[0][1]
air_offset = air_anomalies[0][2]
air_anomalies = air_anomalies[1:]
# generate temps
hotend = temp(hotend_t+hotend_offset, hotend_sigma)
bed = temp(bed_t+bed_offset, bed_sigma)
air = temp(air_t+air_offset, air_sigma)
points.append("%s hotend=%f,bed=%f,air=%f %d" % (
measurement,
hotend,
bed,
air,
(now - epoch).total_seconds(),
))
now += second
# Write data to Kapacitor
r = requests.post(write_url, data='\n'.join(points))
if r.status_code != 204:
print >> sys.stderr, r.text
return 1
return 0
if __name__ == '__main__':
exit(main())
将上述脚本保存为 /tmp/kapacitor_udf/printer_data.py。
这个 Python 脚本有两个 Python 依赖：
requests和numpy。它们可以轻松地通过pip或者你的包管理器安装。
此时我们已经准备好一个任务，并有一个脚本来生成一些带有异常的虚假数据。现在我们可以创建一个虚假数据的记录，以便我们可以轻松地迭代这个任务：
# Start the recording in the background
kapacitor record stream -task print_temps -duration 24h -no-wait
# Grab the ID from the output and store it in a var
rid=7bd3ced5-5e95-4a67-a0e1-f00860b1af47
# Run our python script to generate data
chmod +x ./printer_data.py
./printer_data.py
我们可以通过列出关于录音的信息来验证它是否成功。我们的录音大小为 1.6MB，所以你的应该也接近这个大小：
$ kapacitor list recordings $rid
ID Type Status Size Date
7bd3ced5-5e95-4a67-a0e1-f00860b1af47 stream finished 1.6 MB 04 May 16 11:44 MDT
检测异常
最后，让我们运行这个任务的剧本，看看它是如何工作的：
kapacitor replay -task print_temps -recording $rid -rec-time
检查各种日志文件以查看算法是否捕捉到了异常：
cat /tmp/kapacitor_udf/{hotend,bed,air}_failure.log
根据上面的 printer_data.py 脚本，应该在以下位置存在异常：
- 1小时：热端
- 8小时：床
- 12小时：空气
也可能会存在一些误报，但是，由于我们希望这能够与真实数据（而不是我们干净的假数据）一起使用，因此此时进行调整并没有太大帮助。
好吧，我们有了它。现在我们可以在打印的温度偏离正常范围时得到警报。希望你现在对Kapacitor UDF的工作有了更好的理解，并且有了一个良好的工作示例作为进一步使用UDF的起点。
框架已经搭建完成，现在请插入一个适合你领域的真实异常检测算法！
扩展示例
有一些事情我们留给读者作为练习：
快照/恢复：Kapacitor将定期快照您的UDF过程的状态，以便在过程重新启动时可以恢复。示例 在这里 提供了
snapshot和restore方法的实现。作为练习，为TTestHandler处理器实现它们。将算法从t检验更改为更适合您领域的算法。
numpy和scipy都有丰富的算法。由
info请求返回的选项可以包含多个参数。修改field选项以接受三个字段名称，并将TTestHandler更改为维护每个字段的历史数据和批次，而不仅仅是一个。这样只需要运行一个ttest.py进程。