贡献新的 Kapacitor 输出节点
如果您尚未阅读，请查看 Kapacitor 贡献指南，了解如何开始贡献代码。
目标
在 Kapacitor 中添加一个新节点，用于将数据输出到自定义端点。在本指南中，假设我们要将数据输出到一个虚构的内部数据库 HouseDB。
概述
Kapacitor 通过流水线（pipeline）处理数据。流水线在形式上是一个有向无环图（DAG）。其基本思想是，图中的每个节点代表对数据进行某种形式的处理，每条边负责在节点间传递数据。为了添加一种新类型的节点，需要编写两个组件：
- 用于创建和配置节点的 API (TICKscript)，以及
- 数据处理步骤的实现。
在我们的示例中，数据处理步骤就是将数据输出到 HouseDB。
代码通过两个 Go 包来反映这些要求：
pipeline：该包定义了可用节点的类型及其配置方式。kapacitor：该包提供了pipeline包中定义的每个节点的具体实现。
为了使 API（即 TICKscript）简洁易读，节点的定义与节点的实现是分开的。
更新 TICKscript
首先，我们需要更新 TICKscript，以便用户可以定义我们的新节点。向 HouseDB 发送数据的 TICKscript 应该是什么样的？要连接到 HouseDB 实例，我们需要 URL 和数据库名称，因此我们需要一种方法来提供这些信息。这样如何？
node
|houseDBOut()
.url('house://housedb.example.com')
.database('metrics')为了更新 TICKscript 以支持这些新方法，我们需要编写一个实现了 pipeline.Node 接口的 Go 类型。该接口可以在这里找到，通过 pipeline.node 类型提供了完整的实现。由于 Node 的实现已经为我们准备好了，我们只需要使用它即可。首先我们需要一个名称。HouseDBOutNode 符合命名规范。让我们定义一个 Go struct，通过组合来实现该接口。在 pipeline 目录下创建一个名为 housedb_out.go 的文件，内容如下：
package pipeline
// A HouseDBOutNode will take the incoming data stream and store it in a
// HouseDB instance.
type HouseDBOutNode struct {
// Include the generic node implementation.
node
}就这样，我们在 Go 中拥有了一个实现所需接口的类型。为了实现我们需要的 .url 和 .database 方法，只需在类型上定义同名字段。首字母必须大写以便导出。重要的是字段必须导出，因为它们将被 kapacitor 包中的节点使用。名称的其余部分应与方法名称的大小写保持一致。TICKscript 会在运行时处理大小写匹配。更新 housedb_out.go 文件。
package pipeline
// A HouseDBOutNode will take the incoming data stream and store it in a
// HouseDB instance.
type HouseDBOutNode struct {
// Include the generic node implementation.
node
// URL for connecting to HouseDB
Url string
// Database name
Database string
}接下来，我们需要一种一致的方式来创建节点的新实例。但为此，我们需要考虑该节点如何连接到其他节点。就 Kapacitor 而言，这是一个输出节点，因此它是流水线的终点。我们不会提供任何出站边，图在当前节点结束。我们虚构的 HouseDB 很灵活，可以按批处理或作为单个数据点存储数据。因此，我们不关心 HouseDBOutNode 节点接收什么类型的数据。考虑到这些事实，我们可以定义一个函数来创建一个新的 HouseDBOutNode。将此函数添加到 housedb_out.go 文件的末尾。
// Create a new HouseDBOutNode that accepts any edge type.
func newHouseDBOutNode(wants EdgeType) *HouseDBOutNode {
return &HouseDBOutNode{
node: node{
desc: "housedb",
wants: wants,
provides: NoEdge,
}
}
}通过明确说明节点 wants（需要）和 provides（提供）的边类型，Kapacitor 将执行必要的类型检查，以防止无效的流水线。
最后，我们需要添加一个新的 链式方法 (chaining method)，以便用户可以将 HouseDBOutNode 连接到他们现有的流水线中。链式方法是指创建新节点并将其添加为调用节点子节点的方法。实际上，该方法将节点链接在一起。pipeline.chainnode 类型包含了所有可用于链接节点的方法。一旦我们将方法添加到该类型中，任何其他节点现在都可以与 HouseDBOutNode 链接。将此函数添加到 pipeline/node.go 文件的末尾。
// Create a new HouseDBOutNode as a child of the calling node.
func (c *chainnode) HouseDBOut() *HouseDBOutNode {
h := newHouseDBOutNode(c.Provides())
c.linkChild(h)
return h
}至此，我们已经定义了所有必要的组件，以便 TICKscript 可以定义 HouseDBOutNode。
node
|houseDBOut() // added as a method to the 'chainnode' type
.url('house://housedb.example.com') // added as a field to the HouseDBOutNode
.database('metrics') // added as a field to the HouseDBOutNode
实现 HouseDB 输出
现在 TICKscript 可以定义我们的新输出节点了，我们需要提供具体的实现，以便 Kapacitor 知道如何处理该节点。pipeline 包中的每个节点在 kapacitor 包中都有一个同名节点。创建一个名为 housedb_out.go 的文件，并将其放在仓库根目录下。在文件中放入以下内容。
package kapacitor
import (
"github.com/influxdb/kapacitor/pipeline"
)
type HouseDBOutNode struct {
// Include the generic node implementation
node
// Keep a reference to the pipeline node
h *pipeline.HouseDBOutNode
}kapacitor 包也定义了一个名为 Node 的接口，并提供了通过 kapacitor.node 类型实现的默认实现。同样，我们使用组合来实现该接口。注意，我们还有一个字段，用于包含我们刚刚定义好的 pipeline.HouseDBOutNode 实例。这个 pipeline.HouseDBOutNode 就像一个配置结构，告诉 kapacitor.HouseDBOutNode 它需要做什么。
现在我们有了结构体，让我们定义一个函数来创建我们新结构体的实例。kapacitor 包中的 new*Node 方法遵循以下约定：
func newNodeName(et *ExecutingTask, n *pipeline.NodeName) (*NodeName, error) {}在我们的例子中，我们想定义一个名为 newHouseDBOutNode 的函数。将以下方法添加到 housedb_out.go 文件中。
func newHouseDBOutNode(et *ExecutingTask, n *pipeline.HouseDBOutNode, d NodeDiagnostic) (*HouseDBOutNode, error) {
h := &HouseDBOutNode{
// pass in necessary fields to the 'node' struct
node: node{Node: n, et: et, diag: d},
// Keep a reference to the pipeline.HouseDBOutNode
h: n,
}
// Set the function to be called when running the node
// more on this in a bit.
h.node.runF = h.runOut
return h
}为了创建节点的实例，我们需要将其与 pipeline 包中的节点关联起来。这可以通过 task.go 文件中 createNode 方法的 switch 语句来完成。继续我们的示例：
// Create a node from a given pipeline node.
func (et *ExecutingTask) createNode(p pipeline.Node, d NodeDiagnostic) (n Node, err error) {
switch t := p.(type) {
...
case *pipeline.HouseDBOutNode:
n, err = newHouseDBOutNode(et, t, d)
...
}既然我们已经关联了这两个类型，让我们回到实现输出代码上来。注意 newHouseDBOutNode 函数中的 h.node.runF = h.runOut 行。这一行设置了当节点开始执行时调用的 kapacitor.HouseDBOutNode 方法。现在我们需要定义 runOut 方法。在 housedb_out.go 文件中添加此方法：
func (h *HouseDBOutNode) runOut(snapshot []byte) error {
return nil
}经过此修改，HouseDBOutNode 在语法上已经完整了，但还没有执行任何操作。让我们赋予它一些功能！
正如我们之前所了解的，节点通过边进行通信。有一个 Go 类型 edge.Edge 来处理这种通信。我们要做的只是从边读取数据并将其发送到 HouseDB。数据以 edge.Message 类型的形式呈现。节点使用 edge.Consumer 读取消息，并通过实现 edge.Receiver 接口来处理消息。Consumer 和 Receiver 接口都可以在这里找到。
我们在 HouseDBOutNode 中通过组合包含的 node 类型，在名为 ins 的字段中提供了一个边列表。由于 HouseDBOutNode 只能有一个父节点，我们关心的边是第 0 条边。我们可以使用 NewConsumerWithReceiver 函数消费和处理来自边的消息。
// NewConsumerWithReceiver creates a new consumer for the edge e and receiver r.
func NewConsumerWithReceiver(e Edge, r Receiver) Consumer {
return &consumer{
edge: e,
r: r,
}
}让我们更新 runOut 以使用此函数读取和处理消息。
func (h *HouseDBOutNode) runOut(snapshot []byte) error {
consumer := edge.NewConsumerWithReceiver(
n.ins[0],
h,
)
return consumer.Consume()
}剩下的就是让 HouseDBOutNode 实现 Receiver 接口，并编写一个接收一批点并将其写入 HouseDB 的函数。为了方便起见，我们可以使用 edge.BatchBuffer 来接收批处理消息。我们还可以将单个点消息转换为只包含一个点的批处理消息。
func (h *HouseDBOutNode) BeginBatch(begin edge.BeginBatchMessage) (edge.Message, error) {
return nil, h.batchBuffer.BeginBatch(begin)
}
func (h *HouseDBOutNode) BatchPoint(bp edge.BatchPointMessage) (edge.Message, error) {
return nil, h.batchBuffer.BatchPoint(bp)
}
func (h *HouseDBOutNode) EndBatch(end edge.EndBatchMessage) (edge.Message, error) {
msg := h.batchBuffer.BufferedBatchMessage(end)
return msg, h.write(msg)
}
func (h *HouseDBOutNode) Point(p edge.PointMessage) (edge.Message, error) {
batch := edge.NewBufferedBatchMessage(
edge.NewBeginBatchMessage(
p.Name(),
p.Tags(),
p.Dimensions().ByName,
p.Time(),
1,
),
[]edge.BatchPointMessage{
edge.NewBatchPointMessage(
p.Fields(),
p.Tags(),
p.Time(),
),
},
edge.NewEndBatchMessage(),
)
return p, h.write(batch)
}
func (h *HouseDBOutNode) Barrier(b edge.BarrierMessage) (edge.Message, error) {
return b, nil
}
func (h *HouseDBOutNode) DeleteGroup(d edge.DeleteGroupMessage) (edge.Message, error) {
return d, nil
}
func (h *HouseDBOutNode) Done() {}
// Write a batch of data to HouseDB
func (h *HouseDBOutNode) write(batch edge.BufferedBatchMessage) error {
// Implement writing to HouseDB here...
return nil
}一旦我们实现了 write 方法，我们就完成了。当数据到达 HouseDBOutNode 时，它将被写入指定的 HouseDB 实例。
总结
我们首先在 pipeline 包中编写了一个节点（文件路径：pipeline/housedb_out.go）来定义用于将数据发送到 HouseDB 实例的 TICKscript API。然后我们在 kapacitor 包（文件路径：housedb_out.go）中编写了该节点的实现。我们还更新了 pipeline/node.go 以添加新的链式方法，并更新了 task.go 以关联这两个类型。
以下是完整的文件内容：
pipeline/housedb_out.go
package pipeline
// A HouseDBOutNode will take the incoming data stream and store it in a
// HouseDB instance.
type HouseDBOutNode struct {
// Include the generic node implementation.
node
// URL for connecting to HouseDB
Url string
// Database name
Database string
}
// Create a new HouseDBOutNode that accepts any edge type.
func newHouseDBOutNode(wants EdgeType) *HouseDBOutNode {
return &HouseDBOutNode{
node: node{
desc: "housedb",
wants: wants,
provides: NoEdge,
}
}
}housedb_out.go
package kapacitor
import (
"github.com/influxdb/kapacitor/pipeline"
)
type HouseDBOutNode struct {
// Include the generic node implementation
node
// Keep a reference to the pipeline node
h *pipeline.HouseDBOutNode
// Buffer for a batch of points
batchBuffer *edge.BatchBuffer
}
func newHouseDBOutNode(et *ExecutingTask, n *pipeline.HouseDBOutNode, d NodeDiagnostic) (*HouseDBOutNode, error) {
h := &HouseDBOutNode{
// pass in necessary fields to the 'node' struct
node: node{Node: n, et: et, diag: d},
// Keep a reference to the pipeline.HouseDBOutNode
h: n,
// Buffer for a batch of points
batchBuffer: new(edge.BatchBuffer),
}
// Set the function to be called when running the node
h.node.runF = h.runOut
return h
}
func (h *HouseDBOutNode) runOut(snapshot []byte) error {
consumer := edge.NewConsumerWithReceiver(
n.ins[0],
h,
)
return consumer.Consume()
}
func (h *HouseDBOutNode) BeginBatch(begin edge.BeginBatchMessage) (edge.Message, error) {
return nil, h.batchBuffer.BeginBatch(begin)
}
func (h *HouseDBOutNode) BatchPoint(bp edge.BatchPointMessage) (edge.Message, error) {
return nil, h.batchBuffer.BatchPoint(bp)
}
func (h *HouseDBOutNode) EndBatch(end edge.EndBatchMessage) (edge.Message, error) {
msg := h.batchBuffer.BufferedBatchMessage(end)
return msg, h.write(msg)
}
func (h *HouseDBOutNode) Point(p edge.PointMessage) (edge.Message, error) {
batch := edge.NewBufferedBatchMessage(
edge.NewBeginBatchMessage(
p.Name(),
p.Tags(),
p.Dimensions().ByName,
p.Time(),
1,
),
[]edge.BatchPointMessage{
edge.NewBatchPointMessage(
p.Fields(),
p.Tags(),
p.Time(),
),
},
edge.NewEndBatchMessage(),
)
return p, h.write(batch)
}
func (h *HouseDBOutNode) Barrier(b edge.BarrierMessage) (edge.Message, error) {
return b, nil
}
func (h *HouseDBOutNode) DeleteGroup(d edge.DeleteGroupMessage) (edge.Message, error) {
return d, nil
}
func (h *HouseDBOutNode) Done() {}
// Write a batch of data to HouseDB
func (h *HouseDBOutNode) write(batch edge.BufferedBatchMessage) error {
// Implement writing to HouseDB here...
return nil
}pipeline/node.go (仅显示新的链式方法)
...
// Create a new HouseDBOutNode as a child of the calling node.
func (c *chainnode) HouseDBOut() *HouseDBOutNode {
h := newHouseDBOutNode(c.Provides())
c.linkChild(h)
return h
}
...task.go (仅显示新的 case)
...
// Create a node from a given pipeline node.
func (et *ExecutingTask) createNode(p pipeline.Node, d NodeDiagnostic) (n Node, err error) {
switch t := p.(type) {
...
case *pipeline.HouseDBOutNode:
n, err = newHouseDBOutNode(et, t, d)
...
}
...记录你的新节点
由于 TICKscript 是一门独立的语言，我们构建了一个类似于 godoc 的小型实用程序，名为 tickdoc。tickdoc 会根据代码中的注释生成文档。tickdoc 能够识别两条特殊的注释，以帮助其生成清晰的文档。
tick:ignore：可以添加到任何字段、方法、函数或结构体中。tickdoc将跳过它，不会为其生成任何文档。这对于忽略通过属性方法设置的字段最有用。tick:property：仅添加到方法中。通知tickdoc该方法是一个属性方法而不是链式方法。
将这些注释之一单独放在一行上，tickdoc 就会找到它并相应地处理。否则，请按照常规方式记录你的代码，tickdoc 会完成剩下的工作。
贡献非输出节点。
编写任何节点（不仅是输出节点）的过程非常相似，这里留给读者自行练习。只有几点可能有所不同：
第一个区别是，如果你的新节点可以将数据发送给子节点，它将需要在 pipeline 包中使用 pipeline.Node 接口的 pipeline.chainnode 实现。例如：
package pipeline
type MyCustomNode struct {
// Include pipeline.chainnode so we have all the chaining methods available
// to our new node
chainnode
}
func newMyCustomNode(e EdgeType, n Node) *MyCustomNode {
m := &MyCustomNode{
chainnode: newBasicChainNode("mycustom", e, e),
}
n.linkChild(m)
return m
}第二个区别是，可以定义一个设置流水线节点字段并返回相同实例的方法，以创建一个 属性方法。例如：
package pipeline
type MyCustomNode struct {
// Include pipeline.chainnode so we have all the chaining methods available
// to our new node
chainnode
// Mark this field as ignored for docs
// Since it is set via the Names method below
// tick:ignore
NameList []string `tick:"Names"`
}
func newMyCustomNode(e EdgeType, n Node) *MyCustomNode {
m := &MyCustomNode{
chainnode: newBasicChainNode("mycustom", e, e),
}
n.linkChild(m)
return m
}
// Set the NameList field on the node via this method.
//
// Example:
// node.names('name0', 'name1')
//
// Use the tickdoc comment 'tick:property' to mark this method
// as a 'property method'
// tick:property
func (m *MyCustomNode) Names(name ...string) *MyCustomNode {
m.NameList = name
return m
}此页面是否有帮助？
感谢您的反馈！
支持和反馈
感谢您成为我们社区的一员！我们欢迎并鼓励您对 Kapacitor 和本文档提供反馈和错误报告。要获取支持，请使用以下资源：