go.d drop using cancelreader (#18219)
drop using cancelreader
Ilya Mashchenko committed
Jul 23, 2024 at 14:45 UTC
77189b364e4c1911b456243231bb97090a25d5e2
6 files changed
+134
-82
src/go/go.mod
-1
@@ -37,7 +37,6 @@ require (
37
github.com/mattn/go-xmlrpc v0.0.3
38
github.com/miekg/dns v1.1.61
39
github.com/mitchellh/go-homedir v1.1.0
40
- github.com/muesli/cancelreader v0.2.2
40
github.com/prometheus-community/pro-bing v0.4.0
41
github.com/prometheus/common v0.55.0
42
github.com/prometheus/prometheus v2.5.0+incompatible
src/go/go.sum
-2
@@ -270,8 +270,6 @@ github.com/montanaflynn/stats v0.7.1 h1:etflOAAHORrCC44V+aR6Ftzort912ZU+YLiSTuV8
270
github.com/montanaflynn/stats v0.7.1/go.mod h1:etXPPgVO6n31NxCd9KQUMvCM+ve0ruNzt6R8Bnaayow=
271
github.com/morikuni/aec v1.0.0 h1:nP9CBfwrvYnBRgY6qfDQkygYDmYwOilePFkwzv4dU8A=
272
github.com/morikuni/aec v1.0.0/go.mod h1:BbKIizmSmc5MMPqRYbxO4ZU0S0+P200+tUnFx7PXmsc=
273
-github.com/muesli/cancelreader v0.2.2 h1:3I4Kt4BQjOR54NavqnDogx/MIoWBFa0StPA8ELUXHmA=
274
-github.com/muesli/cancelreader v0.2.2/go.mod h1:3XuTXfFS2VjM+HTLZY9Ak0l6eUKfijIfMUZ4EgX0QYo=
273
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA=
274
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
275
github.com/nxadm/tail v1.4.8 h1:nPr65rt6Y5JFSKQO7qToXr7pePgD6Gwiw05lkbyAQTE=
src/go/plugin/go.d/agent/functions/function.go
+19
-11
@@ -3,8 +3,8 @@
3
package functions
4
5
import (
6
- "bufio"
6
"bytes"
7
+ "context"
8
"encoding/csv"
9
"fmt"
10
"strconv"
@@ -67,22 +67,30 @@ func parseFunction(s string) (*Function, error) {
67
return fn, nil
68
}
69
70
-func parseFunctionWithPayload(s string, sc *bufio.Scanner) (*Function, error) {
70
+func parseFunctionWithPayload(ctx context.Context, s string, in input) (*Function, error) {
71
fn, err := parseFunction(s)
72
if err != nil {
73
return nil, err
74
}
75
76
- var n int
76
var buf bytes.Buffer
78
- for sc.Scan() && sc.Text() != "FUNCTION_PAYLOAD_END" {
79
- if n++; n > 1 {
80
- buf.WriteString("\n")
77
+
78
+ for {
79
+ select {
80
+ case <-ctx.Done():
81
+ return nil, nil
82
+ case line, ok := <-in.lines():
83
+ if !ok {
84
+ return nil, nil
85
+ }
86
+ if line == "FUNCTION_PAYLOAD_END" {
87
+ fn.Payload = append(fn.Payload, buf.Bytes()...)
88
+ return fn, nil
89
+ }
90
+ if buf.Len() > 0 {
91
+ buf.WriteString("\n")
92
+ }
93
+ buf.WriteString(line)
94
}
82
- buf.WriteString(sc.Text())
95
}
84
-
85
- fn.Payload = append(fn.Payload, buf.Bytes()...)
86
-
87
- return fn, nil
96
}
src/go/plugin/go.d/agent/functions/input.go
new
+35
@@ -0,0 +1,35 @@
1
+// SPDX-License-Identifier: GPL-3.0-or-later
2
+
3
+package functions
4
+
5
+import (
6
+ "bufio"
7
+ "os"
8
+)
9
+
10
+type input interface {
11
+ lines() chan string
12
+}
13
+
14
+var stdinInput = func() input {
15
+ r := &stdinReader{chLines: make(chan string)}
16
+ go r.run()
17
+ return r
18
+}()
19
+
20
+type stdinReader struct {
21
+ chLines chan string
22
+}
23
+
24
+func (in *stdinReader) run() {
25
+ sc := bufio.NewScanner(bufio.NewReader(os.Stdin))
26
+
27
+ for sc.Scan() {
28
+ text := sc.Text()
29
+ in.chLines <- text
30
+ }
31
+}
32
+
33
+func (in *stdinReader) lines() chan string {
34
+ return in.chLines
35
+}
src/go/plugin/go.d/agent/functions/manager.go
+57
-66
@@ -3,13 +3,10 @@
3
package functions
4
5
import (
6
- "bufio"
6
"context"
7
"encoding/json"
8
"fmt"
10
- "io"
9
"log/slog"
12
- "os"
10
"strconv"
11
"strings"
12
"sync"
@@ -18,20 +15,15 @@ import (
15
"github.com/netdata/netdata/go/plugins/logger"
16
"github.com/netdata/netdata/go/plugins/plugin/go.d/agent/netdataapi"
17
"github.com/netdata/netdata/go/plugins/plugin/go.d/agent/safewriter"
21
-
22
- "github.com/mattn/go-isatty"
23
- "github.com/muesli/cancelreader"
18
)
19
26
-var isTerminal = isatty.IsTerminal(os.Stdout.Fd()) || isatty.IsTerminal(os.Stdin.Fd())
27
-
20
func NewManager() *Manager {
21
return &Manager{
22
Logger: logger.New().With(
23
slog.String("component", "functions manager"),
24
),
33
- Input: os.Stdin,
25
api: netdataapi.New(safewriter.Stdout),
26
+ input: stdinInput,
27
mux: &sync.Mutex{},
28
FunctionRegistry: make(map[string]func(Function)),
29
}
@@ -40,8 +32,10 @@ func NewManager() *Manager {
32
type Manager struct {
33
*logger.Logger
34
43
- Input io.Reader
44
- api *netdataapi.API
35
+ api *netdataapi.API
36
+
37
+ input input
38
+
39
mux *sync.Mutex
40
FunctionRegistry map[string]func(Function)
41
}
@@ -50,68 +44,65 @@ func (m *Manager) Run(ctx context.Context) {
44
m.Info("instance is started")
45
defer func() { m.Info("instance is stopped") }()
46
53
- if !isTerminal {
54
- r, err := cancelreader.NewReader(m.Input)
55
- if err != nil {
56
- m.Errorf("fail to create cancel reader: %v", err)
57
- return
58
- }
59
-
60
- go func() { <-ctx.Done(); r.Cancel() }()
61
-
62
- var wg sync.WaitGroup
47
+ var wg sync.WaitGroup
48
64
- wg.Add(1)
65
- go func() { defer wg.Done(); m.run(r) }()
49
+ wg.Add(1)
50
+ go func() { defer wg.Done(); m.run(ctx) }()
51
67
- wg.Wait()
68
- _ = r.Close()
69
- }
52
+ wg.Wait()
53
54
<-ctx.Done()
55
}
56
74
-func (m *Manager) run(r io.Reader) {
75
- sc := bufio.NewScanner(r)
76
-
77
- for sc.Scan() {
78
- text := sc.Text()
79
-
80
- var fn *Function
81
- var err error
82
-
83
- // FIXME: if we are waiting for FUNCTION_PAYLOAD_END and a new FUNCTION* appears,
84
- // we need to discard the current one and switch to the new one
85
- switch {
86
- case strings.HasPrefix(text, "FUNCTION "):
87
- fn, err = parseFunction(text)
88
- case strings.HasPrefix(text, "FUNCTION_PAYLOAD "):
89
- fn, err = parseFunctionWithPayload(text, sc)
90
- case text == "":
91
- continue
92
- default:
93
- m.Warningf("unexpected line: '%s'", text)
94
- continue
95
- }
96
-
97
- if err != nil {
98
- m.Warningf("parse function: %v ('%s')", err, text)
99
- continue
100
- }
101
-
102
- function, ok := m.lookupFunction(fn.Name)
103
- if !ok {
104
- m.Infof("skipping execution of '%s': unregistered function", fn.Name)
105
- m.respf(fn, 501, "unregistered function: %s", fn.Name)
106
- continue
107
- }
108
- if function == nil {
109
- m.Warningf("skipping execution of '%s': nil function registered", fn.Name)
110
- m.respf(fn, 501, "nil function: %s", fn.Name)
111
- continue
57
+func (m *Manager) run(ctx context.Context) {
58
+ for {
59
+ select {
60
+ case <-ctx.Done():
61
+ return
62
+ case line, ok := <-m.input.lines():
63
+ if !ok {
64
+ return
65
+ }
66
+
67
+ var fn *Function
68
+ var err error
69
+
70
+ // FIXME: if we are waiting for FUNCTION_PAYLOAD_END and a new FUNCTION* appears,
71
+ // we need to discard the current one and switch to the new one
72
+ switch {
73
+ case strings.HasPrefix(line, "FUNCTION "):
74
+ fn, err = parseFunction(line)
75
+ case strings.HasPrefix(line, "FUNCTION_PAYLOAD "):
76
+ fn, err = parseFunctionWithPayload(ctx, line, m.input)
77
+ case line == "":
78
+ continue
79
+ default:
80
+ m.Warningf("unexpected line: '%s'", line)
81
+ continue
82
+ }
83
+
84
+ if err != nil {
85
+ m.Warningf("parse function: %v ('%s')", err, line)
86
+ continue
87
+ }
88
+ if fn == nil {
89
+ continue
90
+ }
91
+
92
+ function, ok := m.lookupFunction(fn.Name)
93
+ if !ok {
94
+ m.Infof("skipping execution of '%s': unregistered function", fn.Name)
95
+ m.respf(fn, 501, "unregistered function: %s", fn.Name)
96
+ continue
97
+ }
98
+ if function == nil {
99
+ m.Warningf("skipping execution of '%s': nil function registered", fn.Name)
100
+ m.respf(fn, 501, "nil function: %s", fn.Name)
101
+ continue
102
+ }
103
+
104
+ function(*fn)
105
}
113
-
114
- function(*fn)
106
}
107
}
108
src/go/plugin/go.d/agent/functions/manager_test.go
+23
-2
@@ -3,6 +3,7 @@
3
package functions
4
5
import (
6
+ "bufio"
7
"context"
8
"sort"
9
"strings"
@@ -15,7 +16,7 @@ import (
16
func TestNewManager(t *testing.T) {
17
mgr := NewManager()
18
18
- assert.NotNilf(t, mgr.Input, "Input")
19
+ assert.NotNilf(t, mgr.input, "Input")
20
assert.NotNilf(t, mgr.FunctionRegistry, "FunctionRegistry")
21
}
22
@@ -261,7 +262,7 @@ FUNCTION_PAYLOAD_END
262
t.Run(name, func(t *testing.T) {
263
mgr := NewManager()
264
264
- mgr.Input = strings.NewReader(test.input)
265
+ mgr.input = newMockInput(test.input)
266
267
mock := &mockFunctionExecutor{}
268
for _, v := range test.register {
@@ -297,3 +298,23 @@ type mockFunctionExecutor struct {
298
func (m *mockFunctionExecutor) execute(fn Function) {
299
m.executed = append(m.executed, fn)
300
}
301
+
302
+func newMockInput(data string) *mockInput {
303
+ m := &mockInput{chLines: make(chan string)}
304
+ sc := bufio.NewScanner(strings.NewReader(data))
305
+ go func() {
306
+ for sc.Scan() {
307
+ m.chLines <- sc.Text()
308
+ }
309
+ close(m.chLines)
310
+ }()
311
+ return m
312
+}
313
+
314
+type mockInput struct {
315
+ chLines chan string
316
+}
317
+
318
+func (m *mockInput) lines() chan string {
319
+ return m.chLines
320
+}