go.d pkg/socket: keep only one timeout option (#18633)
Ilya Mashchenko committed
Sep 28, 2024 at 19:25 UTC
e251707783df983f8e9a9f424691510194de8078
15 files changed
+107
-143
src/go/plugin/go.d/modules/beanstalk/client.go
+3
-5
@@ -88,11 +88,9 @@ func newBeanstalkConn(conf Config, log *logger.Logger) beanstalkConn {
88
return &beanstalkClient{
89
Logger: log,
90
client: socket.New(socket.Config{
91
- Address: conf.Address,
92
- ConnectTimeout: conf.Timeout.Duration(),
93
- ReadTimeout: conf.Timeout.Duration(),
94
- WriteTimeout: conf.Timeout.Duration(),
95
- TLSConf: nil,
91
+ Address: conf.Address,
92
+ Timeout: conf.Timeout.Duration(),
93
+ TLSConf: nil,
94
}),
95
}
96
}
src/go/plugin/go.d/modules/boinc/client.go
+2
-4
@@ -30,10 +30,8 @@ func newBoincConn(conf Config, log *logger.Logger) boincConn {
30
31
password: conf.Password,
32
conn: socket.New(socket.Config{
33
- Address: conf.Address,
34
- ConnectTimeout: conf.Timeout.Duration(),
35
- ReadTimeout: conf.Timeout.Duration(),
36
- WriteTimeout: conf.Timeout.Duration(),
33
+ Address: conf.Address,
34
+ Timeout: conf.Timeout.Duration(),
35
})}
36
}
37
src/go/plugin/go.d/modules/dovecot/client.go
+2
-4
@@ -16,10 +16,8 @@ type dovecotConn interface {
16
17
func newDovecotConn(conf Config) dovecotConn {
18
return &dovecotClient{conn: socket.New(socket.Config{
19
- Address: conf.Address,
20
- ConnectTimeout: conf.Timeout.Duration(),
21
- ReadTimeout: conf.Timeout.Duration(),
22
- WriteTimeout: conf.Timeout.Duration(),
19
+ Address: conf.Address,
20
+ Timeout: conf.Timeout.Duration(),
21
})}
22
}
23
src/go/plugin/go.d/modules/gearman/client.go
+2
-4
@@ -19,10 +19,8 @@ type gearmanConn interface {
19
20
func newGearmanConn(conf Config) gearmanConn {
21
return &gearmanClient{conn: socket.New(socket.Config{
22
- Address: conf.Address,
23
- ConnectTimeout: conf.Timeout.Duration(),
24
- ReadTimeout: conf.Timeout.Duration(),
25
- WriteTimeout: conf.Timeout.Duration(),
22
+ Address: conf.Address,
23
+ Timeout: conf.Timeout.Duration(),
24
})}
25
}
26
src/go/plugin/go.d/modules/hddtemp/client.go
+2
-4
@@ -29,10 +29,8 @@ func (c *hddtempClient) queryHddTemp() (string, error) {
29
var s string
30
31
cfg := socket.Config{
32
- Address: c.address,
33
- ConnectTimeout: c.timeout,
34
- ReadTimeout: c.timeout,
35
- WriteTimeout: c.timeout,
32
+ Address: c.address,
33
+ Timeout: c.timeout,
34
}
35
36
err := socket.ConnectAndRead(cfg, func(bs []byte) bool {
src/go/plugin/go.d/modules/memcached/client.go
+2
-4
@@ -11,10 +11,8 @@ import (
11
12
func newMemcachedConn(conf Config) memcachedConn {
13
return &memcachedClient{conn: socket.New(socket.Config{
14
- Address: conf.Address,
15
- ConnectTimeout: conf.Timeout.Duration(),
16
- ReadTimeout: conf.Timeout.Duration(),
17
- WriteTimeout: conf.Timeout.Duration(),
14
+ Address: conf.Address,
15
+ Timeout: conf.Timeout.Duration(),
16
})}
17
}
18
src/go/plugin/go.d/modules/openvpn/init.go
+2
-4
@@ -21,10 +21,8 @@ func (o *OpenVPN) initPerUserMatcher() (matcher.Matcher, error) {
21
22
func (o *OpenVPN) initClient() *client.Client {
23
config := socket.Config{
24
- Address: o.Address,
25
- ConnectTimeout: o.Timeout.Duration(),
26
- ReadTimeout: o.Timeout.Duration(),
27
- WriteTimeout: o.Timeout.Duration(),
24
+ Address: o.Address,
25
+ Timeout: o.Timeout.Duration(),
26
}
27
return &client.Client{Client: socket.New(config)}
28
}
src/go/plugin/go.d/modules/tor/client.go
+2
-4
@@ -31,10 +31,8 @@ func newControlConn(conf Config) controlConn {
31
return &torControlClient{
32
password: conf.Password,
33
conn: socket.New(socket.Config{
34
- Address: conf.Address,
35
- ConnectTimeout: conf.Timeout.Duration(),
36
- ReadTimeout: conf.Timeout.Duration(),
37
- WriteTimeout: conf.Timeout.Duration(),
34
+ Address: conf.Address,
35
+ Timeout: conf.Timeout.Duration(),
36
})}
37
}
38
src/go/plugin/go.d/modules/unbound/init.go
+3
-5
@@ -86,11 +86,9 @@ func (u *Unbound) initClient() (err error) {
86
}
87
88
u.client = socket.New(socket.Config{
89
- Address: u.Address,
90
- ConnectTimeout: u.Timeout.Duration(),
91
- ReadTimeout: u.Timeout.Duration(),
92
- WriteTimeout: u.Timeout.Duration(),
93
- TLSConf: tlsCfg,
89
+ Address: u.Address,
90
+ Timeout: u.Timeout.Duration(),
91
+ TLSConf: tlsCfg,
92
})
93
return nil
94
}
src/go/plugin/go.d/modules/upsd/client.go
+2
-4
@@ -29,10 +29,8 @@ type upsUnit struct {
29
30
func newUpsdConn(conf Config) upsdConn {
31
return &upsdClient{conn: socket.New(socket.Config{
32
- ConnectTimeout: conf.Timeout.Duration(),
33
- ReadTimeout: conf.Timeout.Duration(),
34
- WriteTimeout: conf.Timeout.Duration(),
35
- Address: conf.Address,
32
+ Timeout: conf.Timeout.Duration(),
33
+ Address: conf.Address,
34
})}
35
}
36
src/go/plugin/go.d/modules/uwsgi/client.go
+2
-4
@@ -33,10 +33,8 @@ func (c *uwsgiClient) queryStats() ([]byte, error) {
33
const readLineLimit = 1000 * 10
34
35
cfg := socket.Config{
36
- Address: c.address,
37
- ConnectTimeout: c.timeout,
38
- ReadTimeout: c.timeout,
39
- WriteTimeout: c.timeout,
36
+ Address: c.address,
37
+ Timeout: c.timeout,
38
}
39
40
clientErr := socket.ConnectAndRead(cfg, func(bs []byte) bool {
src/go/plugin/go.d/modules/zookeeper/init.go
+3
-5
@@ -30,11 +30,9 @@ func (z *Zookeeper) initZookeeperFetcher() (fetcher, error) {
30
}
31
32
sock := socket.New(socket.Config{
33
- Address: z.Address,
34
- ConnectTimeout: z.Timeout.Duration(),
35
- ReadTimeout: z.Timeout.Duration(),
36
- WriteTimeout: z.Timeout.Duration(),
37
- TLSConf: tlsConf,
33
+ Address: z.Address,
34
+ Timeout: z.Timeout.Duration(),
35
+ TLSConf: tlsConf,
36
})
37
38
return &zookeeperFetcher{Client: sock}, nil
src/go/plugin/go.d/pkg/socket/client.go
+66
-28
@@ -10,24 +10,37 @@ import (
10
"time"
11
)
12
13
-// New returns a new pointer to a socket client given the socket
14
-// type (IP, TCP, UDP, UNIX), a network address (IP/domain:port),
15
-// a timeout and a TLS config. It supports both IPv4 and IPv6 address
16
-// and reuses connection where possible.
17
-func New(config Config) *Socket {
18
- return &Socket{
19
- Config: config,
20
- conn: nil,
21
- }
13
+// Processor function passed to the Socket.Command function.
14
+// It is passed by the caller to process a command's response line by line.
15
+type Processor func([]byte) bool
16
+
17
+// Client is the interface that wraps the basic socket client operations
18
+// and hides the implementation details from the users.
19
+// Implementations should return TCP, UDP or Unix ready sockets.
20
+type Client interface {
21
+ Connect() error
22
+ Disconnect() error
23
+ Command(command string, process Processor) error
24
}
25
24
-func ConnectAndRead(config Config, process Processor) error {
25
- s := New(config)
26
- if err := s.Connect(); err != nil {
26
+func ConnectAndRead(cfg Config, process Processor) error {
27
+ sock := New(cfg)
28
+
29
+ if err := sock.Connect(); err != nil {
30
return err
31
}
29
- defer func() { _ = s.Disconnect() }()
30
- return read(s.conn, process, s.ReadTimeout)
32
+
33
+ defer func() { _ = sock.Disconnect() }()
34
+
35
+ return sock.read(process)
36
+}
37
+
38
+// New returns a new pointer to a socket client given the socket
39
+// type (IP, TCP, UDP, UNIX), a network address (IP/domain:port),
40
+// a timeout and a TLS config. It supports both IPv4 and IPv6 address
41
+// and reuses connection where possible.
42
+func New(cfg Config) *Socket {
43
+ return &Socket{Config: cfg}
44
}
45
46
// Socket is the implementation of a socket client.
@@ -36,6 +49,14 @@ type Socket struct {
49
conn net.Conn
50
}
51
52
+// Config holds the network ip v4 or v6 address, port,
53
+// Socket type(ip, tcp, udp, unix), timeout and TLS configuration for a Socket
54
+type Config struct {
55
+ Address string
56
+ Timeout time.Duration
57
+ TLSConf *tls.Config
58
+}
59
+
60
// Connect connects to the Socket address on the named network.
61
// If the address is a domain name it will also perform the DNS resolution.
62
// Address like :80 will attempt to connect to the localhost.
@@ -46,10 +67,10 @@ func (s *Socket) Connect() error {
67
var err error
68
69
if s.TLSConf == nil {
49
- conn, err = net.DialTimeout(network, address, s.ConnectTimeout)
70
+ conn, err = net.DialTimeout(network, address, s.timeout())
71
} else {
72
var d net.Dialer
52
- d.Timeout = s.ConnectTimeout
73
+ d.Timeout = s.timeout()
74
conn, err = tls.DialWithDialer(&d, network, address, s.TLSConf)
75
}
76
if err != nil {
@@ -81,35 +102,52 @@ func (s *Socket) Command(command string, process Processor) error {
102
if s.conn == nil {
103
return errors.New("cannot send command on nil connection")
104
}
84
- if err := write(command, s.conn, s.WriteTimeout); err != nil {
105
+
106
+ if err := s.write(command); err != nil {
107
return err
108
}
87
- return read(s.conn, process, s.ReadTimeout)
109
+
110
+ return s.read(process)
111
}
112
90
-func write(command string, writer net.Conn, timeout time.Duration) error {
91
- if writer == nil {
113
+func (s *Socket) write(command string) error {
114
+ if s.conn == nil {
115
return errors.New("attempt to write on nil connection")
116
}
94
- if err := writer.SetWriteDeadline(time.Now().Add(timeout)); err != nil {
117
+
118
+ if err := s.conn.SetWriteDeadline(time.Now().Add(s.timeout())); err != nil {
119
return err
120
}
97
- _, err := writer.Write([]byte(command))
121
+
122
+ _, err := s.conn.Write([]byte(command))
123
+
124
return err
125
}
126
101
-func read(reader net.Conn, process Processor, timeout time.Duration) error {
127
+func (s *Socket) read(process Processor) error {
128
if process == nil {
129
return errors.New("process func is nil")
130
}
105
- if reader == nil {
131
+
132
+ if s.conn == nil {
133
return errors.New("attempt to read on nil connection")
134
}
108
- if err := reader.SetReadDeadline(time.Now().Add(timeout)); err != nil {
135
+
136
+ if err := s.conn.SetReadDeadline(time.Now().Add(s.timeout())); err != nil {
137
return err
138
}
111
- scanner := bufio.NewScanner(reader)
112
- for scanner.Scan() && process(scanner.Bytes()) {
139
+
140
+ sc := bufio.NewScanner(s.conn)
141
+
142
+ for sc.Scan() && process(sc.Bytes()) {
143
+ }
144
+
145
+ return sc.Err()
146
+}
147
+
148
+func (s *Socket) timeout() time.Duration {
149
+ if s.Timeout == 0 {
150
+ return time.Second
151
}
114
- return scanner.Err()
152
+ return s.Timeout
153
}
src/go/plugin/go.d/pkg/socket/client_test.go
+14
-23
@@ -19,35 +19,27 @@ const (
19
)
20
21
var tcpConfig = Config{
22
- Address: testServerAddress,
23
- ConnectTimeout: defaultTimeout,
24
- ReadTimeout: defaultTimeout,
25
- WriteTimeout: defaultTimeout,
26
- TLSConf: nil,
22
+ Address: testServerAddress,
23
+ Timeout: defaultTimeout,
24
+ TLSConf: nil,
25
}
26
27
var udpConfig = Config{
30
- Address: testUdpServerAddress,
31
- ConnectTimeout: defaultTimeout,
32
- ReadTimeout: defaultTimeout,
33
- WriteTimeout: defaultTimeout,
34
- TLSConf: nil,
28
+ Address: testUdpServerAddress,
29
+ Timeout: defaultTimeout,
30
+ TLSConf: nil,
31
}
32
33
var unixConfig = Config{
38
- Address: testUnixServerAddress,
39
- ConnectTimeout: defaultTimeout,
40
- ReadTimeout: defaultTimeout,
41
- WriteTimeout: defaultTimeout,
42
- TLSConf: nil,
34
+ Address: testUnixServerAddress,
35
+ Timeout: defaultTimeout,
36
+ TLSConf: nil,
37
}
38
39
var tcpTlsConfig = Config{
46
- Address: testServerAddress,
47
- ConnectTimeout: defaultTimeout,
48
- ReadTimeout: defaultTimeout,
49
- WriteTimeout: defaultTimeout,
50
- TLSConf: &tls.Config{},
40
+ Address: testServerAddress,
41
+ Timeout: defaultTimeout,
42
+ TLSConf: &tls.Config{},
43
}
44
45
func Test_clientCommand(t *testing.T) {
@@ -72,13 +64,12 @@ func Test_clientTimeout(t *testing.T) {
64
time.Sleep(time.Millisecond * 100)
65
sock := New(tcpConfig)
66
require.NoError(t, sock.Connect())
75
- sock.ReadTimeout = 0
76
- sock.ReadTimeout = 0
67
+ sock.Timeout = 0
68
err := sock.Command("ping\n", func(bytes []byte) bool {
69
assert.Equal(t, "pong", string(bytes))
70
return true
71
})
81
- require.Error(t, err)
72
+ require.NoError(t, err)
73
}
74
75
func Test_clientIncompleteSSL(t *testing.T) {
src/go/plugin/go.d/pkg/socket/types.go
deleted
-41
@@ -1,41 +0,0 @@
1
-// SPDX-License-Identifier: GPL-3.0-or-later
2
-
3
-package socket
4
-
5
-import (
6
- "crypto/tls"
7
- "time"
8
-)
9
-
10
-// Processor function passed to the Socket.Command function.
11
-// It is passed by the caller to process a command's response
12
-// line by line.
13
-type Processor func([]byte) bool
14
-
15
-// Client is the interface that wraps the basic socket client operations
16
-// and hides the implementation details from the users.
17
-//
18
-// Connect should prepare the connection.
19
-//
20
-// Disconnect should stop any in-flight connections.
21
-//
22
-// Command should send the actual data to the wire and pass
23
-// any results to the processor function.
24
-//
25
-// Implementations should return TCP, UDP or Unix ready sockets.
26
-type Client interface {
27
- Connect() error
28
- Disconnect() error
29
- Command(command string, process Processor) error
30
-}
31
-
32
-// Config holds the network ip v4 or v6 address, port,
33
-// Socket type(ip, tcp, udp, unix), timeout and TLS configuration
34
-// for a Socket
35
-type Config struct {
36
- Address string
37
- ConnectTimeout time.Duration
38
- ReadTimeout time.Duration
39
- WriteTimeout time.Duration
40
- TLSConf *tls.Config
41
-}