-
Notifications
You must be signed in to change notification settings - Fork 5.5k
/
webhooks.go
134 lines (107 loc) · 3.37 KB
/
webhooks.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
//go:generate ../../../tools/readme_config_includer/generator
package webhooks
import (
_ "embed"
"fmt"
"net"
"net/http"
"reflect"
"time"
"github.com/gorilla/mux"
"github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/config"
"github.com/influxdata/telegraf/plugins/inputs"
"github.com/influxdata/telegraf/plugins/inputs/webhooks/artifactory"
"github.com/influxdata/telegraf/plugins/inputs/webhooks/filestack"
"github.com/influxdata/telegraf/plugins/inputs/webhooks/github"
"github.com/influxdata/telegraf/plugins/inputs/webhooks/mandrill"
"github.com/influxdata/telegraf/plugins/inputs/webhooks/papertrail"
"github.com/influxdata/telegraf/plugins/inputs/webhooks/particle"
"github.com/influxdata/telegraf/plugins/inputs/webhooks/rollbar"
)
//go:embed sample.conf
var sampleConfig string
const (
defaultReadTimeout = 10 * time.Second
defaultWriteTimeout = 10 * time.Second
)
type Webhook interface {
Register(router *mux.Router, acc telegraf.Accumulator, log telegraf.Logger)
}
func init() {
inputs.Add("webhooks", func() telegraf.Input { return NewWebhooks() })
}
type Webhooks struct {
ServiceAddress string `toml:"service_address"`
ReadTimeout config.Duration `toml:"read_timeout"`
WriteTimeout config.Duration `toml:"write_timeout"`
Github *github.GithubWebhook `toml:"github"`
Filestack *filestack.FilestackWebhook `toml:"filestack"`
Mandrill *mandrill.MandrillWebhook `toml:"mandrill"`
Rollbar *rollbar.RollbarWebhook `toml:"rollbar"`
Papertrail *papertrail.PapertrailWebhook `toml:"papertrail"`
Particle *particle.ParticleWebhook `toml:"particle"`
Artifactory *artifactory.ArtifactoryWebhook `toml:"artifactory"`
Log telegraf.Logger `toml:"-"`
srv *http.Server
}
func NewWebhooks() *Webhooks {
return &Webhooks{}
}
func (*Webhooks) SampleConfig() string {
return sampleConfig
}
func (wb *Webhooks) Gather(_ telegraf.Accumulator) error {
return nil
}
// AvailableWebhooks Looks for fields which implement Webhook interface
func (wb *Webhooks) AvailableWebhooks() []Webhook {
webhooks := make([]Webhook, 0)
s := reflect.ValueOf(wb).Elem()
for i := 0; i < s.NumField(); i++ {
f := s.Field(i)
if !f.CanInterface() {
continue
}
if wbPlugin, ok := f.Interface().(Webhook); ok {
if !reflect.ValueOf(wbPlugin).IsNil() {
webhooks = append(webhooks, wbPlugin)
}
}
}
return webhooks
}
func (wb *Webhooks) Start(acc telegraf.Accumulator) error {
if wb.ReadTimeout < config.Duration(time.Second) {
wb.ReadTimeout = config.Duration(defaultReadTimeout)
}
if wb.WriteTimeout < config.Duration(time.Second) {
wb.WriteTimeout = config.Duration(defaultWriteTimeout)
}
r := mux.NewRouter()
for _, webhook := range wb.AvailableWebhooks() {
webhook.Register(r, acc, wb.Log)
}
wb.srv = &http.Server{
Handler: r,
ReadTimeout: time.Duration(wb.ReadTimeout),
WriteTimeout: time.Duration(wb.WriteTimeout),
}
ln, err := net.Listen("tcp", wb.ServiceAddress)
if err != nil {
return fmt.Errorf("error starting server: %w", err)
}
go func() {
if err := wb.srv.Serve(ln); err != nil {
if err != http.ErrServerClosed {
acc.AddError(fmt.Errorf("error listening: %w", err))
}
}
}()
wb.Log.Infof("Started the webhooks service on %s", wb.ServiceAddress)
return nil
}
func (wb *Webhooks) Stop() {
wb.srv.Close()
wb.Log.Infof("Stopping the Webhooks service")
}