forked from elastic/beats
-
Notifications
You must be signed in to change notification settings - Fork 0
/
redis.go
130 lines (110 loc) · 2.74 KB
/
redis.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
package redis
import (
"errors"
"fmt"
"time"
"github.com/elastic/beats/libbeat/common"
"github.com/elastic/beats/libbeat/common/op"
"github.com/elastic/beats/libbeat/logp"
"github.com/elastic/beats/libbeat/outputs"
"github.com/elastic/beats/libbeat/outputs/mode"
"github.com/elastic/beats/libbeat/outputs/mode/modeutil"
"github.com/elastic/beats/libbeat/outputs/transport"
)
type redisOut struct {
mode mode.ConnectionMode
topology
}
var (
debugf = logp.MakeDebug("redis")
)
const (
defaultWaitRetry = 1 * time.Second
defaultMaxWaitRetry = 60 * time.Second
)
func init() {
outputs.RegisterOutputPlugin("redis", new)
}
func new(cfg *common.Config, expireTopo int) (outputs.Outputer, error) {
r := &redisOut{}
if err := r.init(cfg, expireTopo); err != nil {
return nil, err
}
return r, nil
}
func (r *redisOut) init(cfg *common.Config, expireTopo int) error {
config := defaultConfig
if err := cfg.Unpack(&config); err != nil {
return err
}
sendRetries := config.MaxRetries
maxAttempts := config.MaxRetries + 1
if sendRetries < 0 {
maxAttempts = 0
}
var dataType redisDataType
switch config.DataType {
case "", "list":
dataType = redisListType
case "channel":
dataType = redisChannelType
default:
return errors.New("Bad Redis data type")
}
index := []byte(config.Index)
if len(index) == 0 {
return fmt.Errorf("missing %v", cfg.PathOf("index"))
}
tls, err := outputs.LoadTLSConfig(config.TLS)
if err != nil {
return err
}
transp := &transport.Config{
Timeout: config.Timeout,
Proxy: &config.Proxy,
TLS: tls,
}
// configure topology support
r.topology.init(transp, topoConfig{
host: config.HostTopology,
password: config.PasswordTopology,
db: config.DbTopology,
expire: time.Duration(expireTopo) * time.Second,
})
// configure publisher clients
clients, err := modeutil.MakeClients(cfg, func(host string) (mode.ProtocolClient, error) {
t, err := transport.NewClient(transp, "tcp", host, config.Port)
if err != nil {
return nil, err
}
return newClient(t, config.Password, config.Db, index, dataType), nil
})
if err != nil {
return err
}
logp.Info("Max Retries set to: %v", sendRetries)
m, err := modeutil.NewConnectionMode(clients, !config.LoadBalance,
maxAttempts, defaultWaitRetry, config.Timeout, defaultMaxWaitRetry)
if err != nil {
return err
}
r.mode = m
return nil
}
func (r *redisOut) Close() error {
return r.mode.Close()
}
func (r *redisOut) PublishEvent(
signaler op.Signaler,
opts outputs.Options,
event common.MapStr,
) error {
return r.mode.PublishEvent(signaler, opts, event)
}
func (r *redisOut) BulkPublish(
signaler op.Signaler,
opts outputs.Options,
events []common.MapStr,
) error {
return r.mode.PublishEvents(signaler, opts, events)
}