forked from influxdata/kapacitor
-
Notifications
You must be signed in to change notification settings - Fork 0
/
udf.go
154 lines (140 loc) · 4.09 KB
/
udf.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
package pipeline
import (
"fmt"
"time"
"github.com/influxdata/kapacitor/command"
"github.com/influxdata/kapacitor/tick"
"github.com/influxdata/kapacitor/udf"
)
// A UDFNode is a node that can run a User Defined Function (UDF) in a separate process.
//
// A UDF is a custom script or binary that can communicate via Kapacitor's UDF RPC protocol.
// The path and arguments to the UDF program are specified in Kapacitor's configuration.
// Using TICKscripts you can invoke and configure your UDF for each task.
//
// See the [README.md](https://github.com/influxdata/kapacitor/tree/master/udf/agent/)
// for details on how to write your own UDF.
//
// UDFs are configured via Kapacitor's main configuration file.
//
// Example:
// [udf]
// [udf.functions]
// # Example moving average UDF.
// [udf.functions.movingAverage]
// prog = "/path/to/executable/moving_avg"
// args = []
// timeout = "10s"
//
// UDFs are first class objects in TICKscripts and are referenced via their configuration name.
//
// Example:
// // Given you have a UDF that computes a moving average
// // The UDF can define what its options are and then can be
// // invoked via a TICKscript like so:
// stream
// .from()...
// .movingAverage()
// .field('value')
// .size(100)
// .as('mavg')
// .httpOut('movingaverage')
//
// NOTE: The UDF process runs as the same user as the Kapacitor daemon.
// As a result make the user is properly secured as well as the configuration file.
type UDFNode struct {
chainnode
desc string
//tick:ignore
Commander command.Commander
// tick:ignore
Timeout time.Duration
options map[string]*udf.OptionInfo
// Options that were set on the node
// tick:ignore
Options []*udf.Option
describer *tick.ReflectionDescriber
}
func NewUDF(
parent Node,
name string,
commander command.Commander,
timeout time.Duration,
wants,
provides EdgeType,
options map[string]*udf.OptionInfo,
) *UDFNode {
udf := &UDFNode{
chainnode: newBasicChainNode(name, wants, provides),
desc: name,
Commander: commander,
Timeout: timeout,
options: options,
}
udf.describer = tick.NewReflectionDescriber(udf)
parent.linkChild(udf)
return udf
}
// tick:ignore
func (u *UDFNode) Desc() string {
return u.desc
}
// tick:ignore
func (u *UDFNode) HasMethod(name string) bool {
_, ok := u.options[name]
if ok {
return ok
}
return u.describer.HasMethod(name)
}
// tick:ignore
func (u *UDFNode) CallMethod(name string, args ...interface{}) (interface{}, error) {
opt, ok := u.options[name]
if ok {
if got, exp := len(args), len(opt.ValueTypes); got != exp {
return nil, fmt.Errorf("unexpected number of args to %s, got %d expected %d", name, got, exp)
}
values := make([]*udf.OptionValue, len(args))
for i, arg := range args {
values[i] = &udf.OptionValue{}
switch v := arg.(type) {
case bool:
values[i].Type = udf.ValueType_BOOL
values[i].Value = &udf.OptionValue_BoolValue{v}
case int64:
values[i].Type = udf.ValueType_INT
values[i].Value = &udf.OptionValue_IntValue{v}
case float64:
values[i].Type = udf.ValueType_DOUBLE
values[i].Value = &udf.OptionValue_DoubleValue{v}
case string:
values[i].Type = udf.ValueType_STRING
values[i].Value = &udf.OptionValue_StringValue{v}
case time.Duration:
values[i].Type = udf.ValueType_DURATION
values[i].Value = &udf.OptionValue_DurationValue{int64(v)}
}
if values[i].Type != opt.ValueTypes[i] {
return nil, fmt.Errorf("unexpected arg to %s, got %v expected %v", name, values[i].Type, opt.ValueTypes[i])
}
}
u.Options = append(u.Options, &udf.Option{
Name: name,
Values: values,
})
return u, nil
}
return u.describer.CallMethod(name, args...)
}
// tick:ignore
func (u *UDFNode) HasProperty(name string) bool {
return u.describer.HasProperty(name)
}
// tick:ignore
func (u *UDFNode) Property(name string) interface{} {
return u.describer.Property(name)
}
// tick:ignore
func (u *UDFNode) SetProperty(name string, value interface{}) error {
return u.describer.SetProperty(name, value)
}