-
Notifications
You must be signed in to change notification settings - Fork 51
/
counterclient.go
98 lines (87 loc) · 2.47 KB
/
counterclient.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
package counterclient
import (
"context"
"errors"
"os"
"time"
"go.aporeto.io/trireme-lib/collector"
"go.aporeto.io/trireme-lib/controller/constants"
"go.aporeto.io/trireme-lib/controller/internal/enforcer/utils/rpcwrapper"
"go.aporeto.io/trireme-lib/controller/pkg/remoteenforcer/internal/statscollector"
"go.uber.org/zap"
)
const (
counterContextID = "UNUSED"
counterRPCCommand = "ProxyRPCServer.PostCounterEvent"
defaultCounterInterval = 10000 * time.Millisecond
)
type counterClient struct {
collector statscollector.Collector
rpchdl rpcwrapper.RPCClient
secret string
counterChannel string
counterInterval time.Duration
stop chan bool
}
// NewCounterClient returns an interface CounterClient
func NewCounterClient(cr statscollector.Collector) (CounterClient, error) {
c := &counterClient{
collector: cr,
rpchdl: rpcwrapper.NewRPCWrapper(),
secret: os.Getenv(constants.EnvStatsSecret),
counterChannel: os.Getenv(constants.EnvStatsChannel),
counterInterval: defaultCounterInterval,
stop: make(chan bool),
}
if c.counterChannel == "" {
return nil, errors.New("no path to socket provided")
}
if c.secret == "" {
return nil, errors.New("no secret provided for channel")
}
return c, nil
}
func (c *counterClient) sendData(records []*collector.CounterReport) error {
request := rpcwrapper.Request{
Payload: &rpcwrapper.CounterReportPayload{
CounterReports: records,
},
}
return c.rpchdl.RemoteCall(
counterContextID,
counterRPCCommand,
&request,
&rpcwrapper.Response{},
)
}
func (c *counterClient) sendCounterReports(ctx context.Context) {
ticker := time.NewTicker(c.counterInterval)
for {
select {
case <-ticker.C:
records := c.collector.GetAllCounterReports()
if len(records) > 0 {
if err := c.sendData(records); err != nil {
zap.L().Error("Unable to send counter report", zap.Error(err))
}
}
case <-ctx.Done():
records := c.collector.GetAllCounterReports()
if len(records) > 0 {
if err := c.sendData(records); err != nil {
zap.L().Error("Unable to send counter report", zap.Error(err))
}
}
return
}
}
}
// Run stats the counterClient
func (c *counterClient) Run(ctx context.Context) error {
if err := c.rpchdl.NewRPCClient(counterContextID, c.counterChannel, c.secret); err != nil {
zap.L().Error("CounterClient RPC client cannot connect", zap.Error(err))
return err
}
go c.sendCounterReports(ctx)
return nil
}