diff --git a/cfg/config.go b/cfg/config.go index fa0279c3a..d23991876 100644 --- a/cfg/config.go +++ b/cfg/config.go @@ -647,7 +647,7 @@ func ParseNestedFields(fields []string) ([][]string, error) { return result, nil } -func SetDefaultValues(data interface{}) error { +func SetDefaultValues(data any) error { t := reflect.TypeOf(data).Elem() v := reflect.ValueOf(data).Elem() @@ -686,9 +686,10 @@ func SetDefaultValues(data interface{}) error { case reflect.Bool: currentValue := vField.Bool() if !currentValue { - if defaultValue == "true" { + switch defaultValue { + case "true": vField.SetBool(true) - } else if defaultValue == "false" { + case "false": vField.SetBool(false) } } @@ -747,8 +748,8 @@ func mergeYAMLs(a, b map[interface{}]interface{}) map[interface{}]interface{} { } for k, v := range b { if existingValue, exists := merged[k]; exists { - if existingMap, ok := existingValue.(map[interface{}]interface{}); ok { - if newMap, ok := v.(map[interface{}]interface{}); ok { + if existingMap, ok := existingValue.(map[any]any); ok { + if newMap, ok := v.(map[any]any); ok { merged[k] = mergeYAMLs(existingMap, newMap) continue } diff --git a/cfg/config_test.go b/cfg/config_test.go index 37a725739..da11f61d4 100644 --- a/cfg/config_test.go +++ b/cfg/config_test.go @@ -500,7 +500,6 @@ func TestApplyEnvs(t *testing.T) { }, } for _, tt := range tests { - tt := tt t.Run(tt.name, func(t *testing.T) { object, err := simplejson.NewJson([]byte(tt.json)) require.NoError(t, err) @@ -736,21 +735,21 @@ func TestExpression_UnmarshalJSON(t *testing.T) { func TestMergeYAMLs(t *testing.T) { tests := []struct { name string - a map[interface{}]interface{} - b map[interface{}]interface{} - expected map[interface{}]interface{} + a map[any]any + b map[any]any + expected map[any]any }{ { name: "simple merge", - a: map[interface{}]interface{}{ + a: map[any]any{ "key1": "value1", "key2": "value2", }, - b: map[interface{}]interface{}{ + b: map[any]any{ "key2": "newValue2", "key3": "value3", }, - expected: map[interface{}]interface{}{ + expected: map[any]any{ "key1": "value1", "key2": "newValue2", "key3": "value3", @@ -758,19 +757,19 @@ func TestMergeYAMLs(t *testing.T) { }, { name: "nested maps", - a: map[interface{}]interface{}{ - "key1": map[interface{}]interface{}{ + a: map[any]any{ + "key1": map[any]any{ "subkey1": "subvalue1", }, }, - b: map[interface{}]interface{}{ - "key1": map[interface{}]interface{}{ + b: map[any]any{ + "key1": map[any]any{ "subkey2": "subvalue2", }, "key2": "value2", }, - expected: map[interface{}]interface{}{ - "key1": map[interface{}]interface{}{ + expected: map[any]any{ + "key1": map[any]any{ "subkey1": "subvalue1", "subkey2": "subvalue2", }, @@ -779,19 +778,19 @@ func TestMergeYAMLs(t *testing.T) { }, { name: "overwriting nested maps", - a: map[interface{}]interface{}{ - "key1": map[interface{}]interface{}{ + a: map[any]any{ + "key1": map[any]any{ "subkey1": "subvalue1", }, }, - b: map[interface{}]interface{}{ - "key1": map[interface{}]interface{}{ + b: map[any]any{ + "key1": map[any]any{ "subkey1": "newSubvalue1", "subkey2": "subvalue2", }, }, - expected: map[interface{}]interface{}{ - "key1": map[interface{}]interface{}{ + expected: map[any]any{ + "key1": map[any]any{ "subkey1": "newSubvalue1", "subkey2": "subvalue2", }, @@ -799,56 +798,56 @@ func TestMergeYAMLs(t *testing.T) { }, { name: "empty maps", - a: map[interface{}]interface{}{}, - b: map[interface{}]interface{}{}, - expected: map[interface{}]interface{}{ + a: map[any]any{}, + b: map[any]any{}, + expected: map[any]any{ // Expecting an empty map }, }, { name: "a is empty", - a: map[interface{}]interface{}{}, - b: map[interface{}]interface{}{ + a: map[any]any{}, + b: map[any]any{ "key1": "value1", }, - expected: map[interface{}]interface{}{ + expected: map[any]any{ "key1": "value1", }, }, { name: "b is empty", - a: map[interface{}]interface{}{ + a: map[any]any{ "key1": "value1", }, - b: map[interface{}]interface{}{}, - expected: map[interface{}]interface{}{ + b: map[any]any{}, + expected: map[any]any{ "key1": "value1", }, }, { name: "override slice", - a: map[interface{}]interface{}{ - "key1": []interface{}{"value1", "value2"}, + a: map[any]any{ + "key1": []any{"value1", "value2"}, }, - b: map[interface{}]interface{}{ - "key1": []interface{}{"newValue1", "newValue2"}, + b: map[any]any{ + "key1": []any{"newValue1", "newValue2"}, }, - expected: map[interface{}]interface{}{ - "key1": []interface{}{"newValue1", "newValue2"}, + expected: map[any]any{ + "key1": []any{"newValue1", "newValue2"}, }, }, { name: "merge slice with map", - a: map[interface{}]interface{}{ - "key1": []interface{}{"value1", "value2"}, + a: map[any]any{ + "key1": []any{"value1", "value2"}, }, - b: map[interface{}]interface{}{ - "key1": map[interface{}]interface{}{ + b: map[any]any{ + "key1": map[any]any{ "subkey1": "subvalue1", }, }, - expected: map[interface{}]interface{}{ - "key1": map[interface{}]interface{}{ + expected: map[any]any{ + "key1": map[any]any{ "subkey1": "subvalue1", }, }, @@ -909,7 +908,6 @@ func TestBuildFieldSelector(t *testing.T) { } for _, tt := range tests { - tt := tt t.Run(tt.name, func(t *testing.T) { t.Parallel() diff --git a/cfg/matchrule/matchrule_test.go b/cfg/matchrule/matchrule_test.go index 1665ce083..96892d1d1 100644 --- a/cfg/matchrule/matchrule_test.go +++ b/cfg/matchrule/matchrule_test.go @@ -77,7 +77,6 @@ func TestRule_Match(t *testing.T) { }, } for _, tt := range tests { - tt := tt t.Run(tt.name, func(t *testing.T) { t.Parallel() diff --git a/cfg/substitution/regex_filter.go b/cfg/substitution/regex_filter.go index 54470edf9..19a67ce32 100644 --- a/cfg/substitution/regex_filter.go +++ b/cfg/substitution/regex_filter.go @@ -84,7 +84,7 @@ func (r *RegexFilter) compareArgs(args []any) error { if len(wantGroups) != len(gotGroups) { return fmt.Errorf("wrong regex filter groups, want=%v got=%v", wantGroups, gotGroups) } - for i := 0; i < len(wantGroups); i++ { + for i := range wantGroups { if wantGroups[i] != gotGroups[i] { return fmt.Errorf("wrong regex filter groups, want=%v got=%v", wantGroups, gotGroups) } diff --git a/cfg/substitution/substitution_test.go b/cfg/substitution/substitution_test.go index 1fa29926e..4b2d13282 100644 --- a/cfg/substitution/substitution_test.go +++ b/cfg/substitution/substitution_test.go @@ -406,7 +406,6 @@ func TestParseSubstitution(t *testing.T) { } for _, tt := range tests { - tt := tt t.Run(tt.name, func(t *testing.T) { t.Parallel() result, err := ParseSubstitution(tt.substitution, nil, lg) @@ -526,7 +525,6 @@ func TestFilterApply(t *testing.T) { }, } for _, tt := range tests { - tt := tt t.Run(tt.name, func(t *testing.T) { t.Parallel() subOps, err := ParseSubstitution(tt.substitution, nil, lg) diff --git a/cmd/file.d/file.d_test.go b/cmd/file.d/file.d_test.go index 749c6e82f..a20223982 100644 --- a/cmd/file.d/file.d_test.go +++ b/cmd/file.d/file.d_test.go @@ -76,7 +76,7 @@ func TestEndToEnd(t *testing.T) { fileCount := 8 // we are very deterministic :) - rand.Seed(0) + rnd := rand.New(rand.NewSource(0)) // disable k8s environment meta.DisableMetaUpdates = true @@ -96,8 +96,8 @@ func TestEndToEnd(t *testing.T) { tm := time.Now() for { - for i := 0; i < writerCount; i++ { - go runWriter(filesDir, fileCount) + for range writerCount { + go runWriter(filesDir, fileCount, rnd) } time.Sleep(iterationInterval) @@ -107,7 +107,7 @@ func TestEndToEnd(t *testing.T) { } } -func runWriter(tempDir string, files int) { +func runWriter(tempDir string, files int, rnd *rand.Rand) { format := `{"log":"%s\n","stream":"stderr"}` panicLines := make([]string, 0) for _, line := range strings.Split(panicContent, "\n") { @@ -117,7 +117,7 @@ func runWriter(tempDir string, files int) { panicLines = append(panicLines, fmt.Sprintf(format, line)) } - for i := 0; i < files; i++ { + for range files { u1 := strings.ReplaceAll(uuid.NewV4().String(), "-", "") u2 := strings.ReplaceAll(uuid.NewV4().String(), "-", "") name := path.Join(tempDir, "pod_ns_container-"+u1+u2+".log") @@ -131,10 +131,10 @@ func runWriter(tempDir string, files int) { } stream := "stderr" - if rand.Int()%3 == 0 { + if rnd.Int()%3 == 0 { stream = "stderr" } - if rand.Int()%100 == 0 { + if rnd.Int()%100 == 0 { for k := 0; k < 8; k++ { _, _ = fmt.Fprintf(logFile, multilineJSON, stream) _, _ = logFile.Write([]byte{'\n'}) @@ -251,7 +251,6 @@ func TestConfigParseValid(t *testing.T) { }, } for _, tl := range testList { - tl := tl t.Run(tl.name, func(t *testing.T) { t.Parallel() pluginInfo, err := fd.DefaultPluginRegistry.Get(tl.kind, tl.name) @@ -291,7 +290,6 @@ func TestConfigParseInvalid(t *testing.T) { }, } for _, tl := range testList { - tl := tl t.Run(tl.name, func(t *testing.T) { t.Parallel() pluginInfo, err := fd.DefaultPluginRegistry.Get(tl.kind, tl.name) diff --git a/decoder/csv_test.go b/decoder/csv_test.go index e79d50817..8632f1486 100644 --- a/decoder/csv_test.go +++ b/decoder/csv_test.go @@ -77,7 +77,6 @@ func TestDecodeCSV(t *testing.T) { }, } for _, tt := range tests { - tt := tt t.Run(tt.name, func(t *testing.T) { t.Parallel() @@ -161,7 +160,6 @@ func TestDecodeToJsonCSV(t *testing.T) { }, } for _, tt := range tests { - tt := tt t.Run(tt.name, func(t *testing.T) { t.Parallel() diff --git a/decoder/json_test.go b/decoder/json_test.go index 999f4e7bf..24d09bee0 100644 --- a/decoder/json_test.go +++ b/decoder/json_test.go @@ -107,7 +107,6 @@ func TestJson(t *testing.T) { }, } for _, tt := range tests { - tt := tt t.Run(tt.name, func(t *testing.T) { t.Parallel() @@ -143,15 +142,15 @@ func TestJson(t *testing.T) { func genBenchFields(count int) string { var sb strings.Builder - for i := 0; i < count; i++ { - sb.WriteString(fmt.Sprintf(`"field_%d":"vaaaaaaaaaaaaaal_%d",`, i, i)) + for i := range count { + fmt.Fprintf(&sb, `"field_%d":"vaaaaaaaaaaaaaal_%d",`, i, i) } return sb.String() } func genBenchParams(count, maxLen int) map[string]int { m := map[string]int{} - for i := 0; i < count; i++ { + for i := range count { m[fmt.Sprintf("field_%d", i)] = maxLen } return m @@ -165,25 +164,25 @@ func BenchmarkCutFieldsBySize(b *testing.B) { params map[string]int }{ { - json: []byte(fmt.Sprintf(benchJsonFormat, genBenchFields(0))), + json: fmt.Appendf(nil, benchJsonFormat, genBenchFields(0)), params: map[string]int{ "message": 7, }, }, { - json: []byte(fmt.Sprintf(benchJsonFormat, genBenchFields(10))), + json: fmt.Appendf(nil, benchJsonFormat, genBenchFields(10)), params: genBenchParams(9, 3), }, { - json: []byte(fmt.Sprintf(benchJsonFormat, genBenchFields(100))), + json: fmt.Appendf(nil, benchJsonFormat, genBenchFields(100)), params: genBenchParams(98, 5), }, { - json: []byte(fmt.Sprintf(benchJsonFormat, genBenchFields(1000))), + json: fmt.Appendf(nil, benchJsonFormat, genBenchFields(1000)), params: genBenchParams(997, 7), }, { - json: []byte(fmt.Sprintf(benchJsonFormat, genBenchFields(10000))), + json: fmt.Appendf(nil, benchJsonFormat, genBenchFields(10000)), params: genBenchParams(9996, 9), }, } @@ -199,7 +198,7 @@ func BenchmarkCutFieldsBySize(b *testing.B) { cutPositions: make([]jsonCutPos, 0, len(benchCase.params)), mu: &sync.Mutex{}, } - for i := 0; i < b.N; i++ { + for range b.N { _ = d.cutFieldsBySize(benchCase.json) } }) diff --git a/decoder/nginx_test.go b/decoder/nginx_test.go index 82d8301be..44c3c16c5 100644 --- a/decoder/nginx_test.go +++ b/decoder/nginx_test.go @@ -133,7 +133,6 @@ func TestNginxError(t *testing.T) { }, } for _, tt := range tests { - tt := tt t.Run(tt.name, func(t *testing.T) { t.Parallel() diff --git a/decoder/protobuf_test.go b/decoder/protobuf_test.go index 45c3e2259..e01e5c5c5 100644 --- a/decoder/protobuf_test.go +++ b/decoder/protobuf_test.go @@ -162,7 +162,6 @@ func TestProtobuf(t *testing.T) { }, } for _, tt := range tests { - tt := tt t.Run(tt.name, func(t *testing.T) { t.Parallel() diff --git a/decoder/syslog_rfc3164_test.go b/decoder/syslog_rfc3164_test.go index cd7d8fa8b..4a90bdade 100644 --- a/decoder/syslog_rfc3164_test.go +++ b/decoder/syslog_rfc3164_test.go @@ -144,7 +144,6 @@ func TestSyslogRFC3164(t *testing.T) { }, } for _, tt := range tests { - tt := tt t.Run(tt.name, func(t *testing.T) { t.Parallel() diff --git a/decoder/syslog_rfc5424_test.go b/decoder/syslog_rfc5424_test.go index 4c4668d96..f8bb5610b 100644 --- a/decoder/syslog_rfc5424_test.go +++ b/decoder/syslog_rfc5424_test.go @@ -462,7 +462,6 @@ func TestSyslogRFC5424(t *testing.T) { }, } for _, tt := range tests { - tt := tt t.Run(tt.name, func(t *testing.T) { t.Parallel() diff --git a/decoder/syslog_test.go b/decoder/syslog_test.go index 439ad83be..7741a2a2a 100644 --- a/decoder/syslog_test.go +++ b/decoder/syslog_test.go @@ -66,7 +66,6 @@ func TestSyslogParsePriority(t *testing.T) { }, } for _, tt := range tests { - tt := tt t.Run(tt.name, func(t *testing.T) { t.Parallel() diff --git a/e2e/file_clickhouse/sample.go b/e2e/file_clickhouse/sample.go index 9f3c170de..ef24c17a5 100644 --- a/e2e/file_clickhouse/sample.go +++ b/e2e/file_clickhouse/sample.go @@ -73,8 +73,8 @@ func (s *Sample) MarshalJSON() ([]byte, error) { StrArr *[]string `json:"str_arr"` MapStrStr map[string]string `json:"map_str_str"` - UUID string `json:"uuid,omitempty"` - UUIDNullable uuid.NullUUID `json:"uuid_nullable,omitempty"` + UUID string `json:"uuid,omitempty"` + UUIDNullable *uuid.NullUUID `json:"uuid_nullable,omitempty"` }{ C1: s.C1, C2: s.C2, @@ -93,6 +93,6 @@ func (s *Sample) MarshalJSON() ([]byte, error) { StrArr: s.StrArr, MapStrStr: s.MapStrStr, UUID: s.UUID.String(), - UUIDNullable: s.UUIDNullable, + UUIDNullable: &s.UUIDNullable, }) } diff --git a/e2e/file_es_split/file_es_split.go b/e2e/file_es_split/file_es_split.go index ca0ef03bf..450bf20a4 100644 --- a/e2e/file_es_split/file_es_split.go +++ b/e2e/file_es_split/file_es_split.go @@ -51,7 +51,7 @@ func (c *Config) Send(t *testing.T) { _ = file.Close() }() - for i := 0; i < n; i++ { + for range n { err = addEvent(file, successEvent) require.NoError(t, err) } @@ -59,7 +59,7 @@ func (c *Config) Send(t *testing.T) { err = addEvent(file, failEvent) require.NoError(t, err) - for i := 0; i < 2*n; i++ { + for range 2 * n { err = addEvent(file, successEvent) require.NoError(t, err) } diff --git a/e2e/file_loki/file_loki.go b/e2e/file_loki/file_loki.go index 773e70701..0d0dd8c04 100644 --- a/e2e/file_loki/file_loki.go +++ b/e2e/file_loki/file_loki.go @@ -45,7 +45,7 @@ func (c *Config) Configure(t *testing.T, conf *cfg.Config, pipelineName string) labels, err := output.Get("labels").Array() r.NoError(err) - c.label = labels[0].(map[string]interface{})["label"].(string) + c.label = labels[0].(map[string]any)["label"].(string) c.samples = samples url := fmt.Sprintf("%s%s", c.lokiAddr, "/ready") diff --git a/e2e/file_socket/helpers.go b/e2e/file_socket/helpers.go index 80468ae4f..6f4b5ece9 100644 --- a/e2e/file_socket/helpers.go +++ b/e2e/file_socket/helpers.go @@ -132,10 +132,10 @@ func (s *testServer) collected() []string { return out } -func (s *testServer) waitForMessages(minCount, retries int, delay time.Duration) ([]map[string]interface{}, error) { +func (s *testServer) waitForMessages(minCount, retries int, delay time.Duration) ([]map[string]any, error) { time.Sleep(2 * time.Second) - for i := 0; i < retries; i++ { + for range retries { msgs := s.collected() if len(msgs) >= minCount { return decodeMessages(msgs) @@ -150,10 +150,10 @@ func (s *testServer) waitForMessages(minCount, retries int, delay time.Duration) ) } -func decodeMessages(raw []string) ([]map[string]interface{}, error) { - result := make([]map[string]interface{}, 0, len(raw)) +func decodeMessages(raw []string) ([]map[string]any, error) { + result := make([]map[string]any, 0, len(raw)) for _, s := range raw { - var m map[string]interface{} + var m map[string]any if err := json.Unmarshal([]byte(s), &m); err != nil { return nil, fmt.Errorf("failed to decode message %q: %w", s, err) } diff --git a/e2e/kafka_file/kafka_file.go b/e2e/kafka_file/kafka_file.go index c37f43848..d33155e9a 100644 --- a/e2e/kafka_file/kafka_file.go +++ b/e2e/kafka_file/kafka_file.go @@ -70,7 +70,7 @@ func (c *Config) Send(t *testing.T) { msgs[i].Partition = int32(i) } - for i := 0; i < c.Count; i++ { + for range c.Count { result := client.ProduceSync(context.TODO(), msgs...) err := result.FirstErr() if err != nil { diff --git a/e2e/split_join/split_join.go b/e2e/split_join/split_join.go index b4510797e..8fd4fa470 100644 --- a/e2e/split_join/split_join.go +++ b/e2e/split_join/split_join.go @@ -87,7 +87,7 @@ func (c *Config) Send(t *testing.T) { _ = file.Close() }(file) - for i := 0; i < messages; i++ { + for range messages { _, err = file.WriteString(sample + "\n") require.NoError(t, err) } diff --git a/fd/file.d.go b/fd/file.d.go index e071c1acf..5606eb4c7 100644 --- a/fd/file.d.go +++ b/fd/file.d.go @@ -18,6 +18,7 @@ import ( "github.com/ozontech/file.d/pipeline" insaneJSON "github.com/ozontech/insane-json" "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/collectors" "github.com/prometheus/client_golang/prometheus/promhttp" "go.uber.org/atomic" ) @@ -70,8 +71,8 @@ func (f *FileD) initMetrics() { func (f *FileD) createRegistry() { f.registry = prometheus.NewRegistry() - f.registry.MustRegister(prometheus.NewProcessCollector(prometheus.ProcessCollectorOpts{})) - f.registry.MustRegister(prometheus.NewGoCollector()) + f.registry.MustRegister(collectors.NewProcessCollector(collectors.ProcessCollectorOpts{})) + f.registry.MustRegister(collectors.NewGoCollector()) f.registry.MustRegister(newFdsCollector()) } diff --git a/go.mod b/go.mod index 67162cced..249043a0e 100644 --- a/go.mod +++ b/go.mod @@ -51,7 +51,6 @@ require ( go.uber.org/atomic v1.11.0 go.uber.org/automaxprocs v1.5.3 go.uber.org/zap v1.27.0 - golang.org/x/net v0.49.0 google.golang.org/protobuf v1.36.5 gopkg.in/yaml.v2 v2.4.0 gopkg.in/yaml.v3 v3.0.1 @@ -144,6 +143,7 @@ require ( go.yaml.in/yaml/v3 v3.0.4 // indirect golang.org/x/crypto v0.48.0 // indirect golang.org/x/mod v0.32.0 // indirect + golang.org/x/net v0.49.0 // indirect golang.org/x/oauth2 v0.27.0 // indirect golang.org/x/sync v0.19.0 // indirect golang.org/x/sys v0.41.0 // indirect diff --git a/pipeline/batch.go b/pipeline/batch.go index ac975f9fd..c2350777b 100644 --- a/pipeline/batch.go +++ b/pipeline/batch.go @@ -158,7 +158,7 @@ func NewBatcher(opts BatcherOptions) *Batcher { // nolint: gocritic // hugeParam freeBatches := make(chan *Batch, opts.Workers) fullBatches := make(chan *Batch, opts.Workers) - for i := 0; i < opts.Workers; i++ { + for range opts.Workers { freeBatches <- newBatch(opts.BatchSizeCount, opts.BatchSizeBytes, opts.FlushTimeout) } @@ -185,7 +185,7 @@ func (b *Batcher) Start(ctx context.Context) { }() b.workersWg.Add(b.opts.Workers) - for i := 0; i < b.opts.Workers; i++ { + for range b.opts.Workers { go b.work() } diff --git a/pipeline/batch_test.go b/pipeline/batch_test.go index 32a5f1e70..2284a0537 100644 --- a/pipeline/batch_test.go +++ b/pipeline/batch_test.go @@ -75,13 +75,13 @@ func TestBatcher(t *testing.T) { eventsCh := make(chan *Event, 1024) go func() { - for i := 0; i < eventCount; i++ { + for i := range eventCount { eventsCh <- &Event{SeqID: uint64(i)} } close(eventsCh) }() - for p := 0; p < processors; p++ { + for p := range processors { go func(x int) { for event := range eventsCh { event.SourceID = SourceID(x) @@ -145,13 +145,13 @@ func TestBatcherMaxSize(t *testing.T) { eventsCh := make(chan *Event, 1024) go func() { - for i := 0; i < eventCount; i++ { + for i := range eventCount { eventsCh <- &Event{SeqID: uint64(i), Size: eventSize} } close(eventsCh) }() - for p := 0; p < processors; p++ { + for p := range processors { go func(x int) { for event := range eventsCh { event.SourceID = SourceID(x) diff --git a/pipeline/delta_wrapper_test.go b/pipeline/delta_wrapper_test.go index 49328c673..88f553fee 100644 --- a/pipeline/delta_wrapper_test.go +++ b/pipeline/delta_wrapper_test.go @@ -13,7 +13,7 @@ func TestDeltaWrapper(t *testing.T) { initial := int64(0) - for i := 0; i <= 1000000; i++ { + for range 1000000 { delta := rand.Int63n(5000) initial += delta assert.Equal(t, delta, int64(dw.updateValue(initial))) diff --git a/pipeline/doif/check_type_op.go b/pipeline/doif/check_type_op.go index b4801159f..493b6bf74 100644 --- a/pipeline/doif/check_type_op.go +++ b/pipeline/doif/check_type_op.go @@ -196,7 +196,7 @@ func (n *checkTypeOpNode) isEqualTo(n2 Node, _ int) error { "null": root.Dig("null"), "nil": root.Dig("nil"), } - for i := 0; i < len(n.checkTypeFns); i++ { + for i := range n.checkTypeFns { for key, node := range nodes { res1 := n.checkTypeFns[i](node) res2 := n2f.checkTypeFns[i](node) diff --git a/pipeline/doif/check_type_test.go b/pipeline/doif/check_type_test.go index d0694f055..6ca032e95 100644 --- a/pipeline/doif/check_type_test.go +++ b/pipeline/doif/check_type_test.go @@ -200,7 +200,6 @@ func TestCheckType(t *testing.T) { } for _, tt := range tests { - tt := tt t.Run(tt.name, func(t *testing.T) { t.Parallel() var eventRoot *insaneJSON.Root @@ -354,7 +353,6 @@ func TestCheckTypeDuplicateValues(t *testing.T) { } for _, tt := range tests { - tt := tt t.Run(tt.name, func(t *testing.T) { t.Parallel() node, err := NewCheckTypeOpNode(tt.node.field, tt.node.values) diff --git a/pipeline/doif/ctor_test.go b/pipeline/doif/ctor_test.go index 7a52453ab..4dc70119b 100644 --- a/pipeline/doif/ctor_test.go +++ b/pipeline/doif/ctor_test.go @@ -622,7 +622,6 @@ func Test_extractDoIfChecker(t *testing.T) { }, } for _, tt := range tests { - tt := tt t.Run(tt.name, func(t *testing.T) { t.Parallel() reader := bytes.NewBufferString(tt.args.cfgStr) diff --git a/pipeline/doif/do_if_test.go b/pipeline/doif/do_if_test.go index a4baa0734..f4aa16843 100644 --- a/pipeline/doif/do_if_test.go +++ b/pipeline/doif/do_if_test.go @@ -88,7 +88,7 @@ func checkNode(t *testing.T, want, got Node) { assert.Equal(t, wantNode.values, gotNode.values) } else { require.Equal(t, len(wantNode.values), len(gotNode.values)) - for i := 0; i < len(wantNode.values); i++ { + for i := range wantNode.values { wantValues := wantNode.values[i] gotValues := gotNode.values[i] assert.Equal(t, 0, slices.Compare(wantValues, gotValues)) @@ -103,7 +103,7 @@ func checkNode(t *testing.T, want, got Node) { assert.True(t, ok, "values by key %d not present in got node", k) if ok { require.Equal(t, len(wantVals), len(gotVals)) - for i := 0; i < len(wantVals); i++ { + for i := range wantVals { assert.Equal(t, 0, slices.Compare(wantVals[i], gotVals[i])) } } @@ -529,7 +529,6 @@ func TestBuildNodes(t *testing.T) { } for _, tt := range tests { - tt := tt t.Run(tt.name, func(t *testing.T) { t.Parallel() got, err := buildTree(tt.tree) @@ -1133,7 +1132,6 @@ func TestCheck(t *testing.T) { }, } for _, tt := range tests { - tt := tt t.Run(tt.name, func(t *testing.T) { var root Node var eventRoot *insaneJSON.Root @@ -1937,7 +1935,6 @@ func TestNodeIsEqual(t *testing.T) { }, } for _, tt := range tests { - tt := tt t.Run(tt.name, func(t *testing.T) { t.Parallel() root1, err := buildTree(tt.t1) diff --git a/pipeline/doif/field_op.go b/pipeline/doif/field_op.go index 68876808b..4771b834d 100644 --- a/pipeline/doif/field_op.go +++ b/pipeline/doif/field_op.go @@ -282,12 +282,8 @@ func NewFieldOpNode(op string, field string, caseSensitive bool, values [][]byte if !caseSensitive && curVal != nil { curVal = bytes.ToLower(curVal) } - if len(values[i]) < minValLen { - minValLen = len(values[i]) - } - if len(values[i]) > maxValLen { - maxValLen = len(values[i]) - } + minValLen = min(minValLen, len(values[i])) + maxValLen = max(maxValLen, len(values[i])) if fop == fieldEqualOp { valsBySize[len(curVal)] = append(valsBySize[len(curVal)], curVal) } else { @@ -422,7 +418,7 @@ func (n *fieldOpNode) isEqualTo(n2 Node, _ int) error { } else if len(v) != len(v2) { return fmt.Errorf("nodes have different valuesBySize values len under key %d expected: %d", k, len(v)) } else { - for i := 0; i < len(v); i++ { + for i := range len(v) { if !bytes.Equal(v[i], v2[i]) { return fmt.Errorf("nodes have different valuesBySize data under key %d: %v", k, v) } diff --git a/pipeline/doif/logical_op.go b/pipeline/doif/logical_op.go index 964ee36a5..29d6de35a 100644 --- a/pipeline/doif/logical_op.go +++ b/pipeline/doif/logical_op.go @@ -213,7 +213,7 @@ func (n *logicalNode) isEqualTo(n2 Node, level int) error { for i := 0; i < len(n.operands); i++ { if err := n.operands[i].isEqualTo(n2l.operands[i], level+1); err != nil { tabs := make([]byte, 0, level) - for j := 0; j < level; j++ { + for range level { tabs = append(tabs, '\t') } return fmt.Errorf("nodes with op %q have different operand nodes on position %d:\n%s%w", n.op, i, tabs, err) diff --git a/pipeline/event.go b/pipeline/event.go index b6569c531..31e23217d 100644 --- a/pipeline/event.go +++ b/pipeline/event.go @@ -4,6 +4,7 @@ import ( "fmt" "math/bits" "runtime" + "strings" "sync" "time" @@ -234,7 +235,7 @@ func newEventPool(capacity, avgEventSize int) *eventPool { eventPool.getCond = sync.NewCond(eventPool.getMu) - for i := 0; i < capacity; i++ { + for range capacity { eventPool.free1 = append(eventPool.free1, *atomic.NewBool(true)) eventPool.free2 = append(eventPool.free2, *atomic.NewBool(true)) eventPool.events = append(eventPool.events, newEvent()) @@ -251,13 +252,13 @@ func (p *eventPool) get(size int) *Event { for { if x < p.backCounter.Load() { // fast path - if p.free1[x].CAS(true, false) { + if p.free1[x].CompareAndSwap(true, false) { break } - if p.free1[x].CAS(true, false) { + if p.free1[x].CompareAndSwap(true, false) { break } - if p.free1[x].CAS(true, false) { + if p.free1[x].CompareAndSwap(true, false) { break } } @@ -295,13 +296,13 @@ func (p *eventPool) back(event *Event) { var tries int for { // fast path - if p.free2[x].CAS(false, true) { + if p.free2[x].CompareAndSwap(false, true) { break } - if p.free2[x].CAS(false, true) { + if p.free2[x].CompareAndSwap(false, true) { break } - if p.free2[x].CAS(false, true) { + if p.free2[x].CompareAndSwap(false, true) { break } tries++ @@ -339,17 +340,21 @@ func (p *eventPool) wakeupWaiters() { func (p *eventPool) dump() string { out := logger.Cond(len(p.events) == 0, logger.Header("no events"), func() string { - o := logger.Header("events") - for i := 0; i < p.capacity; i++ { + var sb strings.Builder + + sb.WriteString(logger.Header("events")) + for i := range p.capacity { event := p.events[i] eventStr := event.String() if eventStr == "" { eventStr = "nil" } - o += eventStr + "\n" + + sb.WriteString(eventStr) + sb.WriteByte('\n') } - return o + return sb.String() }) return out @@ -405,7 +410,7 @@ type lowMemoryEventPool struct { func newLowMemoryEventPool(capacity int) *lowMemoryEventPool { pools := [syncPools]*sync.Pool{} - for i := 0; i < syncPools; i++ { + for i := range syncPools { pools[i] = &sync.Pool{ New: func() any { return newEvent() diff --git a/pipeline/event_test.go b/pipeline/event_test.go index 323a1775b..cc69251c8 100644 --- a/pipeline/event_test.go +++ b/pipeline/event_test.go @@ -21,7 +21,7 @@ func TestEventPoolDump(t *testing.T) { func BenchmarkEventPoolOneGoroutine(b *testing.B) { bench := func(b *testing.B, p pool) { - for i := 0; i < b.N; i++ { + for range b.N { p.back(p.get(1)) } } @@ -39,14 +39,14 @@ func BenchmarkEventPoolOneGoroutine(b *testing.B) { func BenchmarkEventPoolManyGoroutines(b *testing.B) { bench := func(b *testing.B, p pool) { workers := runtime.GOMAXPROCS(0) - for i := 0; i < b.N; i++ { + for range b.N { wg := &sync.WaitGroup{} wg.Add(workers) - for j := 0; j < workers; j++ { + for range workers { go func() { defer wg.Done() - for k := 0; k < 1000; k++ { + for range 1000 { p.back(p.get(1)) } }() @@ -68,10 +68,10 @@ func BenchmarkEventPoolManyGoroutines(b *testing.B) { func BenchmarkEventPoolSlowestPath(b *testing.B) { bench := func(b *testing.B, p pool) { wg := &sync.WaitGroup{} - for i := 0; i < b.N; i++ { + for range b.N { const concurrency = 1_000 wg.Add(concurrency) - for j := 0; j < concurrency; j++ { + for range concurrency { go func() { defer wg.Done() e := p.get(1) @@ -98,15 +98,15 @@ func TestLowMemPool(t *testing.T) { r := require.New(t) test := func(capacity, batchSize int) { p := newTestLowMemoryEventPool(capacity) - for i := 0; i < batchSize; i++ { + for range batchSize { batch := make([]*Event, batchSize) - for j := 0; j < batchSize; j++ { + for j := range batchSize { batch[j] = p.get(1) } r.Equal(int64(batchSize), p.inUse()) r.Equal(int64(0), p.waiters()) - for j := 0; j < batchSize; j++ { + for j := range batchSize { p.back(batch[j]) } } @@ -133,7 +133,7 @@ func TestEventPoolSlowWait(t *testing.T) { const waiters = 16 // Create 16 goroutines to wait on new events. wg := new(sync.WaitGroup) - for i := 0; i < waiters; i++ { + for range waiters { wg.Add(1) go func() { defer wg.Done() @@ -146,7 +146,7 @@ func TestEventPoolSlowWait(t *testing.T) { } // Wait for all goroutines to be waiting. - for i := 0; i < 50; i++ { + for range 50 { if p.waiters() == waiters { break } @@ -183,10 +183,10 @@ func TestWakeupWaiters(t *testing.T) { concurrency = 5_000 ) pool := newTestEventPool(poolCapacity) - for i := 0; i < 1_000; i++ { + for range 1000 { wg := new(sync.WaitGroup) wg.Add(concurrency) - for i := 0; i < concurrency; i++ { + for range concurrency { go func() { defer wg.Done() e := pool.get(1) diff --git a/pipeline/metadata/templater.go b/pipeline/metadata/templater.go index 8973e92ac..988bcb8b1 100644 --- a/pipeline/metadata/templater.go +++ b/pipeline/metadata/templater.go @@ -37,7 +37,7 @@ type MetaTemplate struct { // NewMetaTemplate creates a new TemplateWrapper with default function func NewMetaTemplate(source string) *MetaTemplate { tmpl := template.Must(template.New("").Funcs(template.FuncMap{ - "default": func(defaultValue string, value interface{}) interface{} { + "default": func(defaultValue string, value any) any { if value == nil || value == "" { return defaultValue } @@ -74,8 +74,8 @@ func NewMetaTemplater(templates cfg.MetaTemplates, logger *zap.Logger, cacheSize continue } expression := strings.TrimSpace(match[1]) - components := strings.Fields(expression) - for _, component := range components { + + for component := range strings.FieldsSeq(expression) { // catch all variables if !strings.HasPrefix(component, ".") { continue @@ -140,7 +140,7 @@ func NewMetaTemplater(templates cfg.MetaTemplates, logger *zap.Logger, cacheSize valueTypes: valueTypes, logger: logger, poolBuffer: sync.Pool{ - New: func() interface{} { return new(bytes.Buffer) }, + New: func() any { return new(bytes.Buffer) }, }, cache: cache, } diff --git a/pipeline/metadata/templater_test.go b/pipeline/metadata/templater_test.go index cd1214106..0c3007834 100644 --- a/pipeline/metadata/templater_test.go +++ b/pipeline/metadata/templater_test.go @@ -146,6 +146,8 @@ func TestTemplaterRender(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { + t.Parallel() + templater := NewMetaTemplater( tt.templates, zap.NewExample(), @@ -187,7 +189,7 @@ func BenchmarkMetaTemplater_Render(b *testing.B) { "auth": nil, } - for i := 0; i < b.N; i++ { + for b.Loop() { _, err := templater.Render(testMetadata{data: mockData}) if err != nil { b.Fatalf("Render failed: %v", err) diff --git a/pipeline/pipeline_test.go b/pipeline/pipeline_test.go index fa31121c8..fb6b5f51c 100644 --- a/pipeline/pipeline_test.go +++ b/pipeline/pipeline_test.go @@ -68,6 +68,8 @@ func TestInInvalidMessages(t *testing.T) { for _, tCase := range cases { t.Run(tCase.name, func(t *testing.T) { + t.Parallel() + pipe := pipeline.New("test_pipeline", tCase.pipelineSettings, prometheus.NewRegistry(), zap.NewNop()) pipe.SetInput(getFakeInputInfo()) @@ -101,7 +103,7 @@ func BenchmarkMetaTemplater(b *testing.B) { }, }) - for i := 0; i < b.N; i++ { + for i := range b.N { rest := i % 100 pipe.In( pipeline.SourceID(1<<16+rest), diff --git a/pipeline/pipeline_whitebox_test.go b/pipeline/pipeline_whitebox_test.go index 0f42d74ac..d56ffefb1 100644 --- a/pipeline/pipeline_whitebox_test.go +++ b/pipeline/pipeline_whitebox_test.go @@ -157,6 +157,8 @@ func TestCheckInputBytes(t *testing.T) { for _, tCase := range cases { t.Run(tCase.name, func(t *testing.T) { + t.Parallel() + pipe := New("test_pipeline", tCase.pipelineSettings, prometheus.NewRegistry(), zap.NewNop()) data, cutoff, ok := pipe.checkInputBytes(tCase.input, "test", nil) @@ -245,6 +247,8 @@ func TestCheckInputBytesMetric(t *testing.T) { for _, tCase := range cases { t.Run(tCase.name, func(t *testing.T) { + t.Parallel() + pipe := New("test_pipeline", tCase.pipelineSettings, prometheus.NewRegistry(), zap.NewNop()) pipe.checkInputBytes([]byte("some log"), tCase.sourceName, tCase.meta) @@ -291,6 +295,8 @@ func TestSuggestDecoder(t *testing.T) { for _, tCase := range tCases { t.Run(tCase.name, func(t *testing.T) { + t.Parallel() + p := New("file_d", tCase.settings, prometheus.NewPedanticRegistry(), zap.NewNop()) p.SuggestDecoder(tCase.suggestType) require.Equal(t, tCase.expectedType, p.decoderType) diff --git a/pipeline/processor_test.go b/pipeline/processor_test.go index 0784f4e1a..b67038b69 100644 --- a/pipeline/processor_test.go +++ b/pipeline/processor_test.go @@ -91,6 +91,8 @@ func Test_processor_isMatch(t *testing.T) { for i, tc := range tcs { t.Run(strconv.Itoa(i), func(t *testing.T) { + t.Parallel() + proc := processor{ busyActions: []bool{false}, actionInfos: []*ActionPluginStaticInfo{ diff --git a/pipeline/util.go b/pipeline/util.go index 8e5363201..9c155553e 100644 --- a/pipeline/util.go +++ b/pipeline/util.go @@ -1,7 +1,6 @@ package pipeline import ( - "reflect" "strings" "unsafe" @@ -27,9 +26,7 @@ func ByteToStringUnsafe(b []byte) string { // StringToByteUnsafe converts string to byte slice without memory copy // This creates mutable string, thus unsafe method, should be used with caution (never modify resulting byte slice) func StringToByteUnsafe(s string) []byte { - var buf = *(*[]byte)(unsafe.Pointer(&s)) - (*reflect.SliceHeader)(unsafe.Pointer(&buf)).Cap = len(s) - return buf + return unsafe.Slice(unsafe.StringData(s), len(s)) } /* There are actually a lot of interesting ways to do this. Saving them here, for the purpose of possible debugging. diff --git a/pipeline/util_test.go b/pipeline/util_test.go index 394e0710e..6ea081f05 100644 --- a/pipeline/util_test.go +++ b/pipeline/util_test.go @@ -85,6 +85,8 @@ func TestCreateNestedFieldPositive(t *testing.T) { for _, tt := range tests { t.Run(tt.Name, func(t *testing.T) { + t.Parallel() + root := insaneJSON.Spawn() defer insaneJSON.Release(root) require.NoError(t, root.DecodeString(tt.Args.Root)) diff --git a/playground/playground.go b/playground/playground.go index 61b65a259..69c2aacbf 100644 --- a/playground/playground.go +++ b/playground/playground.go @@ -166,7 +166,7 @@ func formatZapFields(fields []zapcore.Field) strings.Builder { if fatalPayload.Len() > 0 { fatalPayload.WriteString("; ") } - fatalPayload.WriteString(fmt.Sprintf("%q=%q", key, value)) + fmt.Fprintf(&fatalPayload, "%q=%q", key, value) } return fatalPayload } diff --git a/plugin/action/cardinality/cache_test.go b/plugin/action/cardinality/cache_test.go index a2a742413..bc82f1102 100644 --- a/plugin/action/cardinality/cache_test.go +++ b/plugin/action/cardinality/cache_test.go @@ -122,7 +122,7 @@ func TestConcurrentOperations(t *testing.T) { for _, key := range keys { go func(k string) { defer wg.Done() - for i := 0; i < 100; i++ { + for range 100 { cache.Set(prefix, k) } }(key) @@ -140,7 +140,7 @@ func TestConcurrentOperations(t *testing.T) { for _, key := range keys { go func(k string) { defer wg.Done() - for i := 0; i < 100; i++ { + for range 100 { cacheKeyIsExists(cache, prefix, k) cache.Set(prefix, k+"-new") } @@ -153,7 +153,7 @@ func TestConcurrentOperations(t *testing.T) { for _, key := range keys { go func(k string) { defer wg.Done() - for i := 0; i < 100; i++ { + for range 100 { cache.delete(prefix, k) } }(key) @@ -164,13 +164,13 @@ func TestConcurrentOperations(t *testing.T) { wg.Add(2) go func() { defer wg.Done() - for i := 0; i < 100; i++ { + for range 100 { cache.CountPrefix(prefix) } }() go func() { defer wg.Done() - for i := 0; i < 100; i++ { + for range 100 { cache.Set(prefix, "prefix-key-x") cache.Set(prefix, "prefix-key-y") cache.delete(prefix, "prefix-key-x") @@ -183,7 +183,7 @@ func TestCountPrefixWith10kElements(t *testing.T) { cache := NewCache(time.Minute) n := 10000 prefix := randString(64) - for i := 0; i < n; i++ { + for i := range n { key := fmt.Sprintf("%s-%s", prefix, randString(48)) cache.Set(prefix, key) cache.Set(prefix, key) diff --git a/plugin/action/cardinality/cardinality_test.go b/plugin/action/cardinality/cardinality_test.go index 0687ae31d..66114c5dc 100644 --- a/plugin/action/cardinality/cardinality_test.go +++ b/plugin/action/cardinality/cardinality_test.go @@ -35,10 +35,10 @@ func TestParseFields(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { + t.Parallel() + parsedFields := parseFields(tt.m) - for i := range tt.v { - parsedFields.valsBuf[i] = tt.v[i] - } + copy(parsedFields.valsBuf, tt.v) var buf []byte result := parsedFields.appendTo(buf) @@ -80,7 +80,7 @@ func TestCardinalityLimitDiscard(t *testing.T) { outEventsCnt++ }) - for i := 0; i < genEventsCnt; i++ { + for i := range genEventsCnt { json := fmt.Sprintf(`{"info": {"host":"localhost"},"value":{"i":"%d"}}`, i) input.In(10, "test", test.NewOffset(0), []byte(json)) } @@ -134,7 +134,7 @@ func TestCardinalityLimitRemoveFields(t *testing.T) { } }) - for i := 0; i < genEventsCnt; i++ { + for i := range genEventsCnt { json := fmt.Sprintf(`{"host":"localhost","i":"%d"}`, i) input.In(10, "test", test.NewOffset(0), []byte(json)) } @@ -174,7 +174,7 @@ func TestCardinalityLimitDiscardIfNoSetKeyFields(t *testing.T) { outEventsCnt++ }) - for i := 0; i < genEventsCnt; i++ { + for i := range genEventsCnt { json := fmt.Sprintf(`{"host":"localhost%d","i":"%d"}`, i, i) input.In(10, "test", test.NewOffset(0), []byte(json)) } diff --git a/plugin/action/convert_log_level/convert_log_level_test.go b/plugin/action/convert_log_level/convert_log_level_test.go index f6de05494..78a5f99a6 100644 --- a/plugin/action/convert_log_level/convert_log_level_test.go +++ b/plugin/action/convert_log_level/convert_log_level_test.go @@ -223,6 +223,8 @@ func TestDo(t *testing.T) { for _, tc := range tcs { t.Run(tc.Name, func(t *testing.T) { + t.Parallel() + config := test.NewConfig(&tc.Config, nil) p, input, output := test.NewPipelineMock(test.NewActionPluginStaticInfo(factory, config, pipeline.MatchModeAnd, nil, false)) diff --git a/plugin/action/convert_utf8_bytes/convert_utf8_bytes_test.go b/plugin/action/convert_utf8_bytes/convert_utf8_bytes_test.go index 8c3fbbc30..1e6c22bd3 100644 --- a/plugin/action/convert_utf8_bytes/convert_utf8_bytes_test.go +++ b/plugin/action/convert_utf8_bytes/convert_utf8_bytes_test.go @@ -130,7 +130,6 @@ func TestConvertUTF8Bytes(t *testing.T) { }, } for _, tt := range cases { - tt := tt t.Run(tt.name, func(t *testing.T) { t.Parallel() diff --git a/plugin/action/decode/decode_test.go b/plugin/action/decode/decode_test.go index 6440b23d5..53b1271e5 100644 --- a/plugin/action/decode/decode_test.go +++ b/plugin/action/decode/decode_test.go @@ -388,7 +388,6 @@ func TestDecode(t *testing.T) { }, } for _, tt := range tests { - tt := tt t.Run(tt.name, func(t *testing.T) { t.Parallel() diff --git a/plugin/action/discard/discard_test.go b/plugin/action/discard/discard_test.go index aa9ad181d..ffe290b6f 100644 --- a/plugin/action/discard/discard_test.go +++ b/plugin/action/discard/discard_test.go @@ -117,7 +117,6 @@ func TestDiscard(t *testing.T) { }, } for _, tt := range cases { - tt := tt t.Run(tt.name, func(t *testing.T) { t.Parallel() diff --git a/plugin/action/hash/hash_test.go b/plugin/action/hash/hash_test.go index a4250798f..9d36fca7e 100644 --- a/plugin/action/hash/hash_test.go +++ b/plugin/action/hash/hash_test.go @@ -174,7 +174,6 @@ func TestHash(t *testing.T) { }, } for _, tt := range tests { - tt := tt t.Run(tt.name, func(t *testing.T) { config := test.NewConfig(tt.config, nil) p, input, output := test.NewPipelineMock(test.NewActionPluginStaticInfo(factory, config, pipeline.MatchModeAnd, nil, false), tt.pipeOpts...) diff --git a/plugin/action/join/join_test.go b/plugin/action/join/join_test.go index faf1e3089..c4c60b618 100644 --- a/plugin/action/join/join_test.go +++ b/plugin/action/join/join_test.go @@ -125,12 +125,13 @@ func TestSimpleJoin(t *testing.T) { } for _, tt := range cases { - tt := tt t.Run(tt.name, func(t *testing.T) { + t.Parallel() + format := `{"log":"%s\n"}` content := strings.ReplaceAll(tt.content, "# ===next===\n", "") lines := make([]string, 0) - for _, line := range strings.Split(content, "\n") { + for line := range strings.SplitSeq(content, "\n") { if line == "" { continue } @@ -225,11 +226,13 @@ func TestJoinAfterNilNode(t *testing.T) { } for _, tt := range cases { t.Run(tt.name, func(t *testing.T) { + t.Parallel() + formatNode := `{"log":"%s\n"}` formatNilNode := `{"notlog":"%s\n"}` content := strings.ReplaceAll(tt.content, "# ===next===\n", "") lines := make([]string, 0) - for _, line := range strings.Split(content, "\n") { + for line := range strings.SplitSeq(content, "\n") { if strings.HasPrefix(line, "NilNode:") { lines = append(lines, fmt.Sprintf(formatNilNode, line)) continue diff --git a/plugin/action/join_template/join_template_test.go b/plugin/action/join_template/join_template_test.go index a57819734..916e457c4 100644 --- a/plugin/action/join_template/join_template_test.go +++ b/plugin/action/join_template/join_template_test.go @@ -54,12 +54,11 @@ func TestSimpleJoin(t *testing.T) { } for _, tt := range cases { - tt := tt t.Run(tt.name, func(t *testing.T) { format := `{"log":"%s\n"}` content := strings.ReplaceAll(tt.content, "# ===next===\n", "") lines := make([]string, 0) - for _, line := range strings.Split(content, "\n") { + for line := range strings.SplitSeq(content, "\n") { if line == "" { continue } @@ -144,7 +143,7 @@ func TestJoinAfterNilNode(t *testing.T) { formatNilNode := `{"notlog":"%s\n"}` content := strings.ReplaceAll(tt.content, "# ===next===\n", "") lines := make([]string, 0) - for _, line := range strings.Split(content, "\n") { + for line := range strings.SplitSeq(content, "\n") { if line == "" { continue } diff --git a/plugin/action/join_template/template/cs_exception_test.go b/plugin/action/join_template/template/cs_exception_test.go index 0ed5defb3..1fc7430dd 100644 --- a/plugin/action/join_template/template/cs_exception_test.go +++ b/plugin/action/join_template/template/cs_exception_test.go @@ -10,8 +10,7 @@ import ( func BenchmarkSharpStartMixedRes(b *testing.B) { lines := getLines(sample.SharpException) - b.ResetTimer() - for i := 0; i < b.N; i++ { + for b.Loop() { for _, line := range lines { sharpStartCheck(line) } @@ -21,8 +20,7 @@ func BenchmarkSharpStartMixedRes(b *testing.B) { func BenchmarkSharpContinueMixedRes(b *testing.B) { lines := getLines(sample.SharpException) - b.ResetTimer() - for i := 0; i < b.N; i++ { + for b.Loop() { for _, line := range lines { sharpContinueCheck(line) } @@ -32,8 +30,7 @@ func BenchmarkSharpContinueMixedRes(b *testing.B) { func BenchmarkSharpStartNegativeRes(b *testing.B) { lines := getRandLines() - b.ResetTimer() - for i := 0; i < b.N; i++ { + for b.Loop() { for _, line := range lines { sharpStartCheck(line) } @@ -43,8 +40,7 @@ func BenchmarkSharpStartNegativeRes(b *testing.B) { func BenchmarkSharpContinueNegativeRes(b *testing.B) { lines := getRandLines() - b.ResetTimer() - for i := 0; i < b.N; i++ { + for b.Loop() { for _, line := range lines { sharpContinueCheck(line) } diff --git a/plugin/action/join_template/template/go_data_race_test.go b/plugin/action/join_template/template/go_data_race_test.go index 2b7621910..30c4fd953 100644 --- a/plugin/action/join_template/template/go_data_race_test.go +++ b/plugin/action/join_template/template/go_data_race_test.go @@ -10,8 +10,7 @@ import ( func BenchmarkGoDataRaceStartMixedRes(b *testing.B) { lines := getLines(sample.GoDataRace) - b.ResetTimer() - for i := 0; i < b.N; i++ { + for b.Loop() { for _, line := range lines { goDataRaceStartCheck(line) } @@ -21,8 +20,7 @@ func BenchmarkGoDataRaceStartMixedRes(b *testing.B) { func BenchmarkGoDataRaceFinishMixedRes(b *testing.B) { lines := getLines(sample.GoDataRace) - b.ResetTimer() - for i := 0; i < b.N; i++ { + for b.Loop() { for _, line := range lines { goDataRaceFinishCheck(line) } @@ -32,8 +30,7 @@ func BenchmarkGoDataRaceFinishMixedRes(b *testing.B) { func BenchmarkGoDataRaceStartNegativeRes(b *testing.B) { lines := getRandLines() - b.ResetTimer() - for i := 0; i < b.N; i++ { + for b.Loop() { for _, line := range lines { goDataRaceStartCheck(line) } @@ -43,8 +40,7 @@ func BenchmarkGoDataRaceStartNegativeRes(b *testing.B) { func BenchmarkGoDataRaceFinishNegativeRes(b *testing.B) { lines := getRandLines() - b.ResetTimer() - for i := 0; i < b.N; i++ { + for b.Loop() { for _, line := range lines { goDataRaceFinishCheck(line) } diff --git a/plugin/action/join_template/template/go_panic_test.go b/plugin/action/join_template/template/go_panic_test.go index e261cfadd..30b0ea46f 100644 --- a/plugin/action/join_template/template/go_panic_test.go +++ b/plugin/action/join_template/template/go_panic_test.go @@ -10,8 +10,7 @@ import ( func BenchmarkPanicStartMixedRes(b *testing.B) { lines := getLines(sample.Panics) - b.ResetTimer() - for i := 0; i < b.N; i++ { + for b.Loop() { for _, line := range lines { goPanicStartCheck(line) } @@ -21,8 +20,7 @@ func BenchmarkPanicStartMixedRes(b *testing.B) { func BenchmarkPanicContinueMixedRes(b *testing.B) { lines := getLines(sample.Panics) - b.ResetTimer() - for i := 0; i < b.N; i++ { + for b.Loop() { for _, line := range lines { goPanicContinueCheck(line) } @@ -32,8 +30,7 @@ func BenchmarkPanicContinueMixedRes(b *testing.B) { func BenchmarkPanicStartNegativeRes(b *testing.B) { lines := getRandLines() - b.ResetTimer() - for i := 0; i < b.N; i++ { + for b.Loop() { for _, line := range lines { goPanicStartCheck(line) } @@ -43,8 +40,7 @@ func BenchmarkPanicStartNegativeRes(b *testing.B) { func BenchmarkPanicContinueNegativeRes(b *testing.B) { lines := getRandLines() - b.ResetTimer() - for i := 0; i < b.N; i++ { + for b.Loop() { for _, line := range lines { goPanicContinueCheck(line) } diff --git a/plugin/action/join_template/template/test_utils.go b/plugin/action/join_template/template/test_utils.go index ef525cfbc..b0446a312 100644 --- a/plugin/action/join_template/template/test_utils.go +++ b/plugin/action/join_template/template/test_utils.go @@ -8,7 +8,7 @@ import ( func getLines(joinedContent string) []string { content := strings.ReplaceAll(joinedContent, "# ===next===\n", "") lines := make([]string, 0) - for _, line := range strings.Split(content, "\n") { + for line := range strings.SplitSeq(content, "\n") { if line == "" { continue } @@ -22,7 +22,7 @@ func getRandLines() []string { const count = 100 lines := make([]string, 0, count) - for i := 0; i < count; i++ { + for range count { lines = append(lines, getRandLine()) } @@ -42,7 +42,7 @@ func getRandLine() string { var b strings.Builder b.Grow(sz) - for i := 0; i < sz; i++ { + for range sz { b.WriteByte(from + byte(rand.Intn(to-from))) } diff --git a/plugin/action/json_extract/json_extract_test.go b/plugin/action/json_extract/json_extract_test.go index 8ffafa158..5ff52b8f5 100644 --- a/plugin/action/json_extract/json_extract_test.go +++ b/plugin/action/json_extract/json_extract_test.go @@ -175,7 +175,6 @@ func TestJsonExtract(t *testing.T) { }, } for _, tt := range cases { - tt := tt t.Run(tt.name, func(t *testing.T) { t.Parallel() @@ -209,8 +208,8 @@ func TestJsonExtract(t *testing.T) { func genBenchFields(count int) string { var sb strings.Builder - for i := 0; i < count; i++ { - sb.WriteString(fmt.Sprintf(`"field_%d":"val_%d",`, i, i)) + for i := range count { + fmt.Fprintf(&sb, `"field_%d":"val_%d",`, i, i) } return sb.String() } diff --git a/plugin/action/mask/mask_test.go b/plugin/action/mask/mask_test.go index 45d87b478..16f0f7e85 100644 --- a/plugin/action/mask/mask_test.go +++ b/plugin/action/mask/mask_test.go @@ -230,6 +230,8 @@ func TestMaskFunctions(t *testing.T) { for _, tCase := range suits { t.Run(tCase.name, func(t *testing.T) { + t.Parallel() + buf := make([]byte, 0, 2048) tCase.masks.Re_ = regexp.MustCompile(tCase.masks.Re) buf, masked := tCase.masks.maskValue(tCase.input, buf) @@ -396,6 +398,8 @@ func TestGroupNumbers(t *testing.T) { for _, s := range suits { t.Run(s.name, func(t *testing.T) { + t.Parallel() + if s.isFatal { assert.PanicsWithValue(t, s.fatalMsg, @@ -527,6 +531,8 @@ func TestPlugin(t *testing.T) { for _, s := range suits { t.Run(s.name, func(t *testing.T) { + t.Parallel() + sut, input, output := test.NewPipelineMock( test.NewActionPluginStaticInfo(factory, config, pipeline.MatchModeAnd, @@ -784,6 +790,8 @@ func TestPluginWithComplexMasks(t *testing.T) { for _, s := range suits { t.Run(s.name, func(t *testing.T) { + t.Parallel() + config := test.NewConfig(&Config{ Masks: s.masks, AppliedMetricName: s.metricName, diff --git a/plugin/action/modify/modify_test.go b/plugin/action/modify/modify_test.go index 27c9ae9f8..8989294a6 100644 --- a/plugin/action/modify/modify_test.go +++ b/plugin/action/modify/modify_test.go @@ -89,7 +89,7 @@ func TestModifyRegex(t *testing.T) { root := insaneJSON.Spawn() defer insaneJSON.Release(root) - for i := 0; i < len(testEvents); i++ { + for i := range len(testEvents) { fvs := testEvents[i].fieldsValues _ = root.DecodeString(outEvents[i]) for field := range fvs { diff --git a/plugin/action/move/move_test.go b/plugin/action/move/move_test.go index ccd48c139..8754c5ef2 100644 --- a/plugin/action/move/move_test.go +++ b/plugin/action/move/move_test.go @@ -49,7 +49,6 @@ func TestConfigValidate(t *testing.T) { }, } for _, tt := range cases { - tt := tt t.Run(tt.name, func(t *testing.T) { t.Parallel() @@ -242,7 +241,6 @@ func TestMove(t *testing.T) { }, } for _, tt := range cases { - tt := tt t.Run(tt.name, func(t *testing.T) { t.Parallel() diff --git a/plugin/action/parse_es/parse_es_test.go b/plugin/action/parse_es/parse_es_test.go index 64b3afc97..1979a77dc 100644 --- a/plugin/action/parse_es/parse_es_test.go +++ b/plugin/action/parse_es/parse_es_test.go @@ -66,8 +66,9 @@ func TestDoPassAndDiscard(t *testing.T) { } for _, tCase := range cases { - tCase := tCase t.Run(tCase.name, func(t *testing.T) { + t.Parallel() + p := &Plugin{} p.passNext = tCase.passNextStartState p.discardNext = tCase.discardNextStartState @@ -123,8 +124,9 @@ func TestDo(t *testing.T) { } for _, tCase := range cases { - tCase := tCase t.Run(tCase.name, func(t *testing.T) { + t.Parallel() + p := &Plugin{} root := insaneJSON.Spawn() defer insaneJSON.Release(root) diff --git a/plugin/action/parse_es/pipeline_test.go b/plugin/action/parse_es/pipeline_test.go index d699a6f98..c5a8af3f4 100644 --- a/plugin/action/parse_es/pipeline_test.go +++ b/plugin/action/parse_es/pipeline_test.go @@ -88,8 +88,9 @@ func TestPipeline(t *testing.T) { } for _, tCase := range cases { - tCase := tCase t.Run(tCase.name, func(t *testing.T) { + t.Parallel() + config := test.NewConfig(&Config{}, nil) p, input, output := test.NewPipelineMock(test.NewActionPluginStaticInfo(factory, config, pipeline.MatchModeOr, nil, false)) diff --git a/plugin/action/throttle/buckets.go b/plugin/action/throttle/buckets.go index 9fdb9cf17..a0dc55143 100644 --- a/plugin/action/throttle/buckets.go +++ b/plugin/action/throttle/buckets.go @@ -113,7 +113,7 @@ func (b *simpleBuckets) isEmpty(index int) bool { func (b *simpleBuckets) rebuild(currentTs, ts time.Time) int { resetFn := func(count int) { b.b = append(b.b[count:], b.b[:count]...) - for i := 0; i < count; i++ { + for i := range count { b.reset(b.getCount() - 1 - i) } } @@ -139,7 +139,7 @@ func newDistributedBuckets(count, distributionSize int, interval time.Duration) bucketsMeta: newBucketsMeta(count, interval), b: make([]distributedBucket, count), } - for i := 0; i < count; i++ { + for i := range count { db.b[i] = newDistributedBucket(distributionSize) } return db @@ -179,7 +179,7 @@ func (b *distributedBuckets) isEmpty(index int) bool { func (b *distributedBuckets) rebuild(currentTs, ts time.Time) int { resetFn := func(count int) { b.b = append(b.b[count:], b.b[:count]...) - for i := 0; i < count; i++ { + for i := range count { b.reset(b.getCount() - 1 - i) } } diff --git a/plugin/action/throttle/buckets_test.go b/plugin/action/throttle/buckets_test.go index ce7fd3632..7dcc44449 100644 --- a/plugin/action/throttle/buckets_test.go +++ b/plugin/action/throttle/buckets_test.go @@ -54,7 +54,6 @@ func TestMetaActualizeIndex(t *testing.T) { }, } for _, tt := range tests { - tt := tt t.Run(tt.name, func(t *testing.T) { t.Parallel() @@ -184,13 +183,12 @@ func TestRebuildBuckets(t *testing.T) { }, } for _, tt := range tests { - tt := tt t.Run(tt.name, func(t *testing.T) { t.Parallel() resetFn := func(count int) { tt.tc.b = append(tt.tc.b[count:], tt.tc.b[:count]...) - for i := 0; i < count; i++ { + for i := range count { tt.tc.b[len(tt.tc.b)-1-i] = 0 } } diff --git a/plugin/action/throttle/limiters_map.go b/plugin/action/throttle/limiters_map.go index 817365088..e7f5bdc97 100644 --- a/plugin/action/throttle/limiters_map.go +++ b/plugin/action/throttle/limiters_map.go @@ -187,7 +187,7 @@ func (l *limitersMap) runSync(ctx context.Context, workerCount int, syncInterval wg := sync.WaitGroup{} jobs := make(chan limiter, workerCount) - for i := 0; i < workerCount; i++ { + for range workerCount { go l.syncWorker(jobs, &wg) } diff --git a/plugin/action/throttle/redis_limiter.go b/plugin/action/throttle/redis_limiter.go index 013c04c7f..1ebdb453f 100644 --- a/plugin/action/throttle/redis_limiter.go +++ b/plugin/action/throttle/redis_limiter.go @@ -118,7 +118,7 @@ func (l *redisLimiter) sync() { l.keyIdxsForSync = l.keyIdxsForSync[:0] l.bucketIdsForSync = l.bucketIdsForSync[:0] - for i := 0; i < count; i++ { + for i := range count { l.bucketValuesForSync[i] = l.bucketValuesForSync[i][:0] // no new events passed diff --git a/plugin/action/throttle/redis_limiter_test.go b/plugin/action/throttle/redis_limiter_test.go index 9fdb9ec5d..36f5713a0 100644 --- a/plugin/action/throttle/redis_limiter_test.go +++ b/plugin/action/throttle/redis_limiter_test.go @@ -314,7 +314,6 @@ func Test_updateKeyLimit(t *testing.T) { }, } for _, tt := range tests { - tt := tt t.Run(tt.name, func(t *testing.T) { t.Parallel() ctl := metric.NewCtl("test", prometheus.NewRegistry(), time.Minute, 0) @@ -462,7 +461,6 @@ func Test_decodeKeyLimitValue(t *testing.T) { }, } for _, tt := range tests { - tt := tt t.Run(tt.name, func(t *testing.T) { t.Parallel() gotLimit, gotDistr, err := decodeKeyLimitValue(tt.args.data, tt.args.valField, tt.args.distrField) diff --git a/plugin/action/throttle/throttle_test.go b/plugin/action/throttle/throttle_test.go index 965323e39..b97a0c1f8 100644 --- a/plugin/action/throttle/throttle_test.go +++ b/plugin/action/throttle/throttle_test.go @@ -377,7 +377,7 @@ func TestRedisThrottleMultiPipes(t *testing.T) { `{"time":"%s","k8s_ns":"ns_3","k8s_pod":"pod_1"}`, `{"time":"%s","k8s_ns":"ns_4","k8s_pod":"pod_1"}`, } - for i := 0; i < len(firstPipeEvents); i++ { + for i := range len(firstPipeEvents) { json := fmt.Sprintf(firstPipeEvents[i], time.Now().Format(time.RFC3339Nano)) input.In(10, sourceNames[rand.Int()%len(sourceNames)], test.NewOffset(0), []byte(json)) // timeout required due shifting time call to redis @@ -386,7 +386,7 @@ func TestRedisThrottleMultiPipes(t *testing.T) { // limit is 1 while events count is 3 assert.Greater(t, len(firstPipeEvents), outEvents, "wrong in events count") - for i := 0; i < len(secondPipeEvents); i++ { + for i := range len(secondPipeEvents) { json := fmt.Sprintf(secondPipeEvents[i], time.Now().Format(time.RFC3339Nano)) inputSec.In(10, sourceNames[rand.Int()%len(sourceNames)], test.NewOffset(0), []byte(json)) // timeout required due shifting time call to redis @@ -778,7 +778,7 @@ func TestThrottleLimiterExpiration(t *testing.T) { } nowTs := time.Now().Format(time.RFC3339Nano) - for i := 0; i < eventsTotal; i++ { + for i := range eventsTotal { json := fmt.Sprintf(events[i], nowTs) input.In(10, sourceNames[rand.Int()%len(sourceNames)], test.NewOffset(0), []byte(json)) @@ -909,7 +909,7 @@ func TestThrottleWithDistribution(t *testing.T) { } nowTs := time.Now().Format(time.RFC3339Nano) - for i := 0; i < len(events); i++ { + for i := range len(events) { json := fmt.Sprintf(events[i], nowTs) input.In(0, "test", test.NewOffset(0), []byte(json)) } diff --git a/plugin/input/file/file_test.go b/plugin/input/file/file_test.go index 2bf3babe4..9e583a92e 100644 --- a/plugin/input/file/file_test.go +++ b/plugin/input/file/file_test.go @@ -411,7 +411,7 @@ func TestWatch(t *testing.T) { addString(file, content, true, true) }, Act: func(p *pipeline.Pipeline) { - for x := 0; x < iterations; x++ { + for x := range iterations { dir := fmt.Sprintf("dir_%d", x) go func(dir string) { dir = filepath.Join(filepath.Dir(file), dir) @@ -441,7 +441,7 @@ func TestReadSimple(t *testing.T) { run(&test.Case{ Prepare: func() { - for i := 0; i < eventCount; i++ { + for i := range eventCount { events = append(events, fmt.Sprintf(`{"field":"value_%d"}`, i)) } }, @@ -475,7 +475,7 @@ func TestReadContinue(t *testing.T) { }, Act: func(p *pipeline.Pipeline) { file = createTempFile() - for x := 0; x < blockSize; x++ { + for x := range blockSize { line := fmt.Sprintf(`{"data_1":"line_%d"}`, x) size += len(line) + newLine inputEvents[line] = true @@ -484,7 +484,7 @@ func TestReadContinue(t *testing.T) { }, Assert: func(p *pipeline.Pipeline) { processed = p.GetEventsTotal() - for i := 0; i < processed; i++ { + for i := range processed { outputEvents[p.GetEventLogItem(i)] = true } }, @@ -497,7 +497,7 @@ func TestReadContinue(t *testing.T) { Prepare: func() { }, Act: func(p *pipeline.Pipeline) { - for x := 0; x < blockSize; x++ { + for x := range blockSize { line := fmt.Sprintf(`{"data_2":"line_%d"}`, x) size += len(line) + newLine inputEvents[line] = true @@ -528,7 +528,7 @@ func TestReadCompressed(t *testing.T) { run(&test.Case{ Prepare: func() { - for i := 0; i < eventCount; i++ { + for i := range eventCount { events = append(events, fmt.Sprintf(`{"field":"value_%d"}`, i)) } }, @@ -571,7 +571,7 @@ func TestOffsetsSaveSimple(t *testing.T) { run(&test.Case{ Prepare: func() { s := "1" - for i := 0; i < eventCount; i++ { + for range eventCount { s += "1" event := fmt.Sprintf(`{"field":"value_%s"}`, s) events = append(events, event) @@ -686,7 +686,7 @@ func TestReadBufferOverflow(t *testing.T) { run(&test.Case{ Prepare: func() { file = createTempFile() - for i := 0; i < iterations; i++ { + for range iterations { addString(file, firstLine+"\n"+secondLine+"\n", false, false) } }, @@ -709,7 +709,7 @@ func TestReadManyCharsRace(t *testing.T) { file = createTempFile() addString(file, `"`, false, false) - for i := 0; i < charCount; i++ { + for range charCount { addString(file, "a", false, false) } addString(file, `"`, true, false) @@ -750,7 +750,7 @@ func TestReadManyFilesRace(t *testing.T) { files := make([]string, 0, fileCount) run(&test.Case{ Prepare: func() { - for i := 0; i < fileCount; i++ { + for range fileCount { file := createTempFile() oneFileSize = addLines(file, eventCountPerFile, eventCountPerFile*2) @@ -776,7 +776,7 @@ func TestReadLongJSON(t *testing.T) { }, Act: func(p *pipeline.Pipeline) { file = createTempFile() - for i := 0; i < eventCount; i++ { + for range eventCount { addBytes(file, json, false, true) } }, @@ -802,7 +802,7 @@ func TestReadManyFilesParallelRace(t *testing.T) { run(&test.Case{ Prepare: func() { - for f := 0; f < fileCount; f++ { + for range fileCount { file := createTempFile() f, err := os.OpenFile(file, os.O_APPEND|os.O_WRONLY, perm) if err != nil { @@ -815,7 +815,7 @@ func TestReadManyFilesParallelRace(t *testing.T) { Act: func(p *pipeline.Pipeline) { for i := range fds { go func(index int) { - for i := 0; i < blockCount; i++ { + for range blockCount { addDataFile(fds[index], json) } }(i) @@ -855,7 +855,7 @@ func TestReadManyCharsParallelRace(t *testing.T) { eventCount := lineCount * blockCount * fileCount run(&test.Case{ Prepare: func() { - for f := 0; f < fileCount; f++ { + for range fileCount { file := createTempFile() f, err := os.OpenFile(file, os.O_APPEND|os.O_WRONLY, perm) if err != nil { @@ -868,7 +868,7 @@ func TestReadManyCharsParallelRace(t *testing.T) { Act: func(p *pipeline.Pipeline) { for i := range fds { go func(index int) { - for i := 0; i < blockCount; i++ { + for range blockCount { addDataFile(fds[index], json1) addDataFile(fds[index], json2) } @@ -929,11 +929,11 @@ func TestReadStreamRace(t *testing.T) { run(&test.Case{ Prepare: func() { content := make([]byte, 0, len(json)*blocksCount) - for i := 0; i < blocksCount; i++ { + for range blocksCount { content = append(content, json...) } - for f := 0; f < filesCount; f++ { + for range filesCount { file := createTempFile() fileNames = append(fileNames, file) addBytes(file, content, false, false) @@ -1098,7 +1098,7 @@ func TestTruncationSeq(t *testing.T) { wg := &sync.WaitGroup{} - for k := 0; k < 4; k++ { + for range 4 { wg.Add(1) go func() { lwg := &sync.WaitGroup{} @@ -1203,11 +1203,11 @@ func BenchmarkLightJsonReadPar(b *testing.B) { json := getContent("../../../testdata/json/light.json") content := make([]byte, 0, len(json)*lines) - for i := 0; i < lines; i++ { + for range lines { content = append(content, json...) } - for f := 0; f < files; f++ { + for range files { file := createTempFile() addBytes(file, content, false, false) } diff --git a/plugin/input/file/provider_test.go b/plugin/input/file/provider_test.go index 3622e9a5c..4f645d0db 100644 --- a/plugin/input/file/provider_test.go +++ b/plugin/input/file/provider_test.go @@ -221,7 +221,7 @@ func TestEOFInfo(t *testing.T) { e := &eofInfo{} var wg sync.WaitGroup - for i := 0; i < 100; i++ { + for i := range 100 { wg.Add(1) go func(idx int) { defer wg.Done() @@ -239,7 +239,7 @@ func TestEOFInfo(t *testing.T) { var wg sync.WaitGroup // Start multiple goroutines writing different offsets - for i := 0; i < 100; i++ { + for i := range 100 { wg.Add(1) go func(offset int64) { defer wg.Done() diff --git a/plugin/input/file/resetter_test.go b/plugin/input/file/resetter_test.go index 957a8371b..e64813f89 100644 --- a/plugin/input/file/resetter_test.go +++ b/plugin/input/file/resetter_test.go @@ -64,7 +64,6 @@ func Test_deleteOneOffsetByField(t *testing.T) { }, } for _, tt := range tests { - tt := tt t.Run(tt.name, func(t *testing.T) { fname := "./one_offset_test.yaml" err := os.WriteFile(fname, []byte(tt.fileData), os.ModePerm) diff --git a/plugin/input/file/watcher_test.go b/plugin/input/file/watcher_test.go index 1dc000d9f..0e1f8e032 100644 --- a/plugin/input/file/watcher_test.go +++ b/plugin/input/file/watcher_test.go @@ -28,7 +28,6 @@ func TestWatcher(t *testing.T) { }, } for _, tt := range tests { - tt := tt t.Run(tt.name, func(t *testing.T) { dir := t.TempDir() shouldCreate := atomic.Int64{} @@ -178,7 +177,6 @@ func TestWatcherPaths(t *testing.T) { }, } for _, tt := range tests { - tt := tt relFilename, _ := filepath.Rel(dir, tt.filename) t.Run(relFilename, func(t *testing.T) { filename := tt.filename diff --git a/plugin/input/file/worker.go b/plugin/input/file/worker.go index 18bb6738d..daeaeb0c3 100644 --- a/plugin/input/file/worker.go +++ b/plugin/input/file/worker.go @@ -244,8 +244,7 @@ func isNotFileBeingWritten(filePath string) bool { } // Check the output for write access - lines := strings.Split(string(output), "\n") - for _, line := range lines { + for line := range strings.SplitSeq(string(output), "\n") { // Check if the line contains 'w' indicating write access if strings.Contains(line, "w") { return true // File is being written to diff --git a/plugin/input/http/elasticsearch_test.go b/plugin/input/http/elasticsearch_test.go index 73db5ac37..7000ef377 100644 --- a/plugin/input/http/elasticsearch_test.go +++ b/plugin/input/http/elasticsearch_test.go @@ -53,7 +53,6 @@ func TestElasticsearchResponse(t *testing.T) { } for _, tc := range tcs { - tc := tc t.Run(tc.Name, func(t *testing.T) { t.Parallel() diff --git a/plugin/input/http/http_test.go b/plugin/input/http/http_test.go index d9de29a31..8a46e7899 100644 --- a/plugin/input/http/http_test.go +++ b/plugin/input/http/http_test.go @@ -554,7 +554,7 @@ func BenchmarkHttpInputJson(b *testing.B) { time.Sleep(100 * time.Millisecond) // http listen start delay go func() { - for j := 0; j < DocumentCount; j++ { + for range DocumentCount { jobs <- struct{}{} } close(jobs) @@ -625,7 +625,6 @@ func TestGzip(t *testing.T) { } for _, tc := range tests { - tc := tc t.Run(tc.Name, func(t *testing.T) { t.Parallel() @@ -798,7 +797,6 @@ func TestCORSPrepareAllowedOrigins(t *testing.T) { } for _, tc := range tests { - tc := tc t.Run(tc.Name, func(t *testing.T) { t.Parallel() corsCfg := CORSConfig{ @@ -908,7 +906,6 @@ func TestCORSGetAllowedByOrigin(t *testing.T) { } for _, tc := range tests { - tc := tc t.Run(tc.Name, func(t *testing.T) { t.Parallel() corsCfg := CORSConfig{ diff --git a/plugin/input/socket/socket_test.go b/plugin/input/socket/socket_test.go index c414fa2d6..95df3a575 100644 --- a/plugin/input/socket/socket_test.go +++ b/plugin/input/socket/socket_test.go @@ -66,7 +66,7 @@ func doTest(t *testing.T, config *Config, clients int, parallel bool) { if parallel { conns := make([]net.Conn, 0, clients) - for i := 0; i < clients; i++ { + for range clients { conn, err := net.Dial(config.Network, config.Address) require.NoError(t, err) @@ -84,7 +84,7 @@ func doTest(t *testing.T, config *Config, clients int, parallel bool) { conn.Close() } } else { - for i := 0; i < clients; i++ { + for range clients { conn, err := net.Dial(config.Network, config.Address) require.NoError(t, err) diff --git a/plugin/output/clickhouse/README.md b/plugin/output/clickhouse/README.md index 4d12e9860..221e5618d 100644 --- a/plugin/output/clickhouse/README.md +++ b/plugin/output/clickhouse/README.md @@ -215,7 +215,7 @@ After this timeout batch will be sent even if batch isn't completed.
-**`ban_period`** *`cfg.Duration`* *`default=10s`* +**`ban_period`** *`cfg.Duration`* *`default=0s`* Period for which addresses will be banned in case of unavailability. diff --git a/plugin/output/clickhouse/clickhouse.go b/plugin/output/clickhouse/clickhouse.go index 1675d9d1a..e0d63183f 100644 --- a/plugin/output/clickhouse/clickhouse.go +++ b/plugin/output/clickhouse/clickhouse.go @@ -10,6 +10,7 @@ import ( "net" "strings" "sync" + "sync/atomic" "time" "github.com/ClickHouse/ch-go" @@ -21,7 +22,6 @@ import ( "github.com/ozontech/file.d/pipeline" "github.com/ozontech/file.d/xtime" "github.com/ozontech/file.d/xtls" - "go.uber.org/atomic" "go.uber.org/zap" "go.uber.org/zap/zapcore" ) @@ -64,17 +64,19 @@ type Plugin struct { // TODO: support shards instances []instance - requestID atomic.Int64 + requestID atomic.Uint64 // plugin metrics - insertErrorsMetric *metric.Counter - queriesCountMetric *metric.Counter + insertErrorsMetric *metric.Counter + queriesCountMetric *metric.Counter + bannedEndpointsMetric *metric.Gauge router *pipeline.Router compression ch.Compression tlsConfig *tls.Config + cbEnabled bool poolsByAddr map[Address]Clickhouse bannedHosts map[Address]time.Time pendingHosts map[Address]struct{} @@ -356,7 +358,7 @@ type Config struct { // > @3@4@5@6 // > // > Period for which addresses will be banned in case of unavailability. - BanPeriod cfg.Duration `json:"ban_period" default:"10s" parse:"duration"` // * + BanPeriod cfg.Duration `json:"ban_period" default:"0s" parse:"duration"` // * BanPeriod_ time.Duration // > @3@4@5@6 @@ -380,6 +382,8 @@ func Factory() (pipeline.AnyPlugin, pipeline.AnyConfig) { func (p *Plugin) registerMetrics(ctl *metric.Ctl) { p.insertErrorsMetric = ctl.RegisterCounter("output_clickhouse_errors_total", "Total clickhouse insert errors") p.queriesCountMetric = ctl.RegisterCounter("output_clickhouse_queries_count_total", "How many queries sent by clickhouse output plugin") + p.bannedEndpointsMetric = ctl.RegisterGauge("output_clickhouse_banned_endpoints_count", "Current number of endpoints banned by circuit breaker") + p.bannedEndpointsMetric.Set(0) } func (p *Plugin) Start(config pipeline.AnyConfig, params *pipeline.OutputPluginParams) { @@ -390,7 +394,6 @@ func (p *Plugin) Start(config pipeline.AnyConfig, params *pipeline.OutputPluginP p.logger.Fatal("config.addresses can't be empty") } - p.bannedHosts = make(map[Address]time.Time, len(p.config.Addresses)) p.pendingHosts = make(map[Address]struct{}, len(p.config.Addresses)) p.poolsByAddr = make(map[Address]Clickhouse, len(p.config.Addresses)) @@ -406,8 +409,18 @@ func (p *Plugin) Start(config pipeline.AnyConfig, params *pipeline.OutputPluginP if p.config.ReconnectInterval_ < 1 { p.logger.Fatal("'reconnect_interval' can't be <1") } - if p.config.BanPeriod_ < 1 { - p.logger.Fatal("'ban_period' cant't be <1") + if p.config.BanPeriod_ < 0 { + p.logger.Fatal("'ban_period' cant't be <0") + } + + p.cbEnabled = p.config.BanPeriod_ > 0 && len(p.config.Addresses) > 1 + if p.cbEnabled { + p.logger.Info( + "circuit breaker enabled", + zap.Duration("ban_period", p.config.BanPeriod_), + zap.Duration("reconnect_interval", p.config.ReconnectInterval_), + zap.Int("addresses_count", len(p.config.Addresses)), + ) } schema, err := inferInsaneColInputs(p.config.Columns) @@ -468,7 +481,10 @@ func (p *Plugin) Start(config pipeline.AnyConfig, params *pipeline.OutputPluginP p.logger.Fatal("cannot start: no available clickhouse addresses in config") } - go p.checkBannedHosts() + if p.cbEnabled { + p.bannedHosts = make(map[Address]time.Time, len(p.config.Addresses)) + go p.checkBannedHosts() + } if len(p.pendingHosts) > 0 { go p.checkPendingHosts() @@ -566,7 +582,7 @@ func (p *Plugin) checkPendingHosts() { p.mu.Lock() p.poolsByAddr[addr] = pool - for j := 0; j < *addr.Weight; j++ { + for range *addr.Weight { p.instances = append(p.instances, instance{ addr: addr, pool: pool, @@ -588,6 +604,7 @@ func (p *Plugin) checkBannedHosts() { case <-p.ctx.Done(): return case <-ticker.C: + restoredAddresses := make([]string, 0, len(p.config.Addresses)) p.mu.Lock() for addr, banUntil := range p.bannedHosts { if !xtime.GetInaccurateTime().After(banUntil) { @@ -601,8 +618,19 @@ func (p *Plugin) checkBannedHosts() { }) } delete(p.bannedHosts, addr) + restoredAddresses = append(restoredAddresses, addr.Addr) } + activeCount, bannedCount := len(p.instances), len(p.bannedHosts) p.mu.Unlock() + + if len(restoredAddresses) > 0 { + p.logger.Info("clickhouse hosts restored", + zap.Strings("addrs", restoredAddresses), + zap.Int("active_instances", activeCount), + zap.Int("banned_hosts", bannedCount), + ) + p.bannedEndpointsMetric.Set(float64(bannedCount)) + } } } } @@ -681,6 +709,11 @@ func (p *Plugin) out(workerData *pipeline.WorkerData, batch *pipeline.Batch) err var err error if attempts == 0 { + p.logger.Warn("all clickhouse hosts banned, failing back to all hosts") + if err := p.fallbackInsert(data.input); err == nil { + return nil + } + if p.config.FatalOnFailedInsert { p.logger.Fatal("cannot start: no available clickhouse addresses in config") } @@ -692,8 +725,8 @@ func (p *Plugin) out(workerData *pipeline.WorkerData, batch *pipeline.Batch) err return err } - for i := 0; i < attempts; i++ { - requestID := p.requestID.Inc() + for i := range attempts { + requestID := p.requestID.Add(1) p.mu.RLock() if len(p.instances) == 0 { @@ -710,8 +743,8 @@ func (p *Plugin) out(workerData *pipeline.WorkerData, batch *pipeline.Batch) err } var netError net.Error - if errors.As(err, &netError) { - p.banInstance(instance.addr) + if errors.As(err, &netError) || errors.Is(err, context.DeadlineExceeded) { + p.banInstance(instance.addr, err) } } if err != nil { @@ -738,7 +771,36 @@ func (p *Plugin) do(clickhouse Clickhouse, queryInput proto.Input) error { }) } -func (p *Plugin) banInstance(addr Address) { +func (p *Plugin) fallbackInsert(input proto.Input) error { + lastErr := errNoAvailableClickhouseAddresses + + for range len(p.config.Addresses) { + idx := int(p.requestID.Add(1) % uint64(len(p.config.Addresses))) + addr := p.config.Addresses[idx] + + p.mu.RLock() + pool := p.poolsByAddr[addr] + p.mu.RUnlock() + + if pool == nil { + continue + } + + if err := p.do(pool, input); err == nil { + return nil + } else { + lastErr = err + } + } + + return lastErr +} + +func (p *Plugin) banInstance(addr Address, err error) { + if !p.cbEnabled { + return + } + p.mu.Lock() defer p.mu.Unlock() @@ -748,17 +810,29 @@ func (p *Plugin) banInstance(addr Address) { filtered = append(filtered, it) } } + p.instances = filtered p.bannedHosts[addr] = xtime.GetInaccurateTime().Add(p.config.BanPeriod_) + activeCount, bannedCount := len(p.instances), len(p.bannedHosts) + + p.logger.Warn("clickhouse host banned", + zap.Error(err), + zap.String("addr", addr.Addr), + zap.Duration("ban_period", p.config.BanPeriod_), + zap.Int("active_instances", activeCount), + zap.Int("banned_hosts", bannedCount), + ) + + p.bannedEndpointsMetric.Set(float64(bannedCount)) } -func (p *Plugin) getInstance(requestID int64, retry int) instance { +func (p *Plugin) getInstance(requestID uint64, retry int) instance { var instanceIdx int switch p.config.InsertStrategy_ { case StrategyInOrder: instanceIdx = retry % len(p.instances) case StrategyRoundRobin: - instanceIdx = int(requestID) % len(p.instances) + instanceIdx = int(requestID % uint64(len(p.instances))) } return p.instances[instanceIdx] } diff --git a/plugin/output/clickhouse/clickhouse_test.go b/plugin/output/clickhouse/clickhouse_test.go index 14829360a..ce9973a83 100644 --- a/plugin/output/clickhouse/clickhouse_test.go +++ b/plugin/output/clickhouse/clickhouse_test.go @@ -39,7 +39,7 @@ func TestPlugin_getInstance(t *testing.T) { } type args struct { - id int64 + id uint64 retry int } tests := []struct { @@ -54,21 +54,21 @@ func TestPlugin_getInstance(t *testing.T) { name: "one instance and first retry", instances: instances[:1], stategy: StrategyInOrder, - args: args{id: rand.Int63(), retry: 0}, + args: args{id: rand.Uint64(), retry: 0}, want: instances[0], }, { name: "one instance and some retry", instances: instances[:1], stategy: StrategyInOrder, - args: args{id: rand.Int63(), retry: 123}, + args: args{id: rand.Uint64(), retry: 123}, want: instances[0], }, { name: "many instances and some retry", instances: instances, stategy: StrategyInOrder, - args: args{id: rand.Int63(), retry: 123}, + args: args{id: rand.Uint64(), retry: 123}, want: instances[3], // 123%3 }, // round-robin @@ -90,13 +90,12 @@ func TestPlugin_getInstance(t *testing.T) { name: "one instances and rand retry", instances: instances[:1], stategy: StrategyRoundRobin, - args: args{id: rand.Int63(), retry: rand.Int()}, + args: args{id: rand.Uint64(), retry: rand.Int()}, want: instances[0], }, } for _, tt := range tests { - tt := tt t.Run(tt.name, func(t *testing.T) { t.Parallel() @@ -218,7 +217,6 @@ func TestAddress_UnmarshalJSON(t *testing.T) { }, } for _, tt := range tests { - tt := tt t.Run(tt.name, func(t *testing.T) { t.Parallel() a := &Address{} diff --git a/plugin/output/file/file.go b/plugin/output/file/file.go index 8311e08da..a4e40c4fc 100644 --- a/plugin/output/file/file.go +++ b/plugin/output/file/file.go @@ -1,6 +1,7 @@ package file import ( + "context" "fmt" "os" "path" @@ -14,7 +15,6 @@ import ( "github.com/ozontech/file.d/logger" "github.com/ozontech/file.d/pipeline" "go.uber.org/zap" - "golang.org/x/net/context" ) /*{ introduction diff --git a/plugin/output/file/files_test.go b/plugin/output/file/files_test.go index 52aba7c59..428e4478f 100644 --- a/plugin/output/file/files_test.go +++ b/plugin/output/file/files_test.go @@ -35,6 +35,8 @@ func TestPluginsExists(t *testing.T) { for _, tCase := range cases { t.Run(tCase.name, func(t *testing.T) { + t.Parallel() + plugins := tCase.prepareFunc() exists := plugins.Exists(tCase.search) require.Equal(t, tCase.expected, exists) @@ -71,6 +73,8 @@ func TestPluginsIsStatic(t *testing.T) { for _, tCase := range cases { t.Run(tCase.name, func(t *testing.T) { + t.Parallel() + plugins := tCase.prepareFunc() exists := plugins.IsStatic(tCase.search) require.Equal(t, tCase.expected, exists) @@ -108,6 +112,8 @@ func TestPluginsIsDynamic(t *testing.T) { for _, tCase := range cases { t.Run(tCase.name, func(t *testing.T) { + t.Parallel() + plugins := tCase.prepareFunc() exists := plugins.IsDynamic(tCase.search) require.Equal(t, tCase.expected, exists) diff --git a/plugin/output/loki/loki_test.go b/plugin/output/loki/loki_test.go index 84e7ada5e..2235d862b 100644 --- a/plugin/output/loki/loki_test.go +++ b/plugin/output/loki/loki_test.go @@ -59,6 +59,8 @@ func TestPluginParseLabels(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { + t.Parallel() + pl := &Plugin{ config: &Config{ Labels: tt.lables, @@ -102,6 +104,8 @@ func TestPluginGetAuthHeaders(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { + t.Parallel() + pl := &Plugin{ config: &Config{ Auth: AuthConfig{ diff --git a/plugin/output/postgres/postgres_test.go b/plugin/output/postgres/postgres_test.go index 805c936f2..f6521fbc0 100644 --- a/plugin/output/postgres/postgres_test.go +++ b/plugin/output/postgres/postgres_test.go @@ -71,7 +71,7 @@ func TestPrivateOut(t *testing.T) { pool := mockpool ctx := context.Background() - var ctxMock = reflect.TypeOf((*context.Context)(nil)).Elem() + var ctxMock = reflect.TypeFor[context.Context]() mockpool.EXPECT().Query( gomock.AssignableToTypeOf(ctxMock), @@ -141,7 +141,7 @@ func TestPrivateOutWithRetry(t *testing.T) { pool := mockpool ctx := context.Background() - var ctxMock = reflect.TypeOf((*context.Context)(nil)).Elem() + var ctxMock = reflect.TypeFor[context.Context]() mockpool.EXPECT().Query( gomock.AssignableToTypeOf(ctxMock), @@ -274,7 +274,7 @@ func TestPrivateOutDeduplicatedEvents(t *testing.T) { pool := mockpool ctx := context.Background() - var ctxMock = reflect.TypeOf((*context.Context)(nil)).Elem() + var ctxMock = reflect.TypeFor[context.Context]() mockpool.EXPECT().Query( gomock.AssignableToTypeOf(ctxMock), @@ -444,7 +444,7 @@ func TestPrivateOutFewUniqueEventsYetWithDeduplicationEventsAnpooladEvents(t *te pool := mockpool ctx := context.Background() - var ctxMock = reflect.TypeOf((*context.Context)(nil)).Elem() + var ctxMock = reflect.TypeFor[context.Context]() mockpool.EXPECT().Query( gomock.AssignableToTypeOf(ctxMock), diff --git a/plugin/output/s3/s3_test.go b/plugin/output/s3/s3_test.go index 9dc503ea3..dc455f16e 100644 --- a/plugin/output/s3/s3_test.go +++ b/plugin/output/s3/s3_test.go @@ -2,6 +2,7 @@ package s3 import ( "bufio" + "context" "fmt" "math/rand" "os" @@ -22,7 +23,6 @@ import ( "github.com/stretchr/testify/assert" "go.uber.org/atomic" "go.uber.org/zap" - "golang.org/x/net/context" ) const targetFile = "filetests/log.log" diff --git a/plugin/output/splunk/splunk_test.go b/plugin/output/splunk/splunk_test.go index fec6de177..192cdd0e4 100644 --- a/plugin/output/splunk/splunk_test.go +++ b/plugin/output/splunk/splunk_test.go @@ -28,7 +28,6 @@ func TestSplunk(t *testing.T) { } for _, tt := range suites { - tt := tt t.Run(tt.name, func(t *testing.T) { t.Parallel() @@ -97,7 +96,6 @@ func TestParseSplunkError(t *testing.T) { }, } for _, tt := range cases { - tt := tt t.Run(tt.name, func(t *testing.T) { t.Parallel() @@ -158,7 +156,6 @@ func TestCopyFields(t *testing.T) { } for _, tt := range suites { - tt := tt t.Run(tt.name, func(t *testing.T) { t.Parallel() diff --git a/xtime/parse_time_test.go b/xtime/parse_time_test.go index ba0905a87..97931f85f 100644 --- a/xtime/parse_time_test.go +++ b/xtime/parse_time_test.go @@ -44,6 +44,8 @@ func TestParseTime(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { + t.Parallel() + got, err := ParseTime(tt.format, tt.value) require.NoError(t, err, "must be no error") require.Equal(t, tt.want, got.UnixNano())