Merge remote-tracking branch 'origin/master' into 444-async-commit

ozontech · Jul 31, 2023 · ed07843 · ed07843
2 parents 248c3cb + 634eeff
commit ed07843
Show file tree

Hide file tree

Showing 6 changed files with 26 additions and 22 deletions.
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
@@ -45,7 +45,7 @@ jobs:
           GOARCH: ${{ matrix.arch }}
           GOFLAGS: ${{ matrix.flags }}
           LOG_LEVEL: error
-        run: go test -coverprofile=profile.out -covermode=atomic -coverpkg=./... ./...
+        run: go test -coverprofile=profile.out -covermode=atomic -v -coverpkg=./... ./...
 
       - name: Upload artifact
         uses: actions/upload-artifact@v3

diff --git a/go.mod b/go.mod
@@ -10,7 +10,7 @@ require (
 	github.com/alecthomas/kingpin v2.2.6+incompatible
 	github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d
 	github.com/alicebob/miniredis/v2 v2.30.4
-	github.com/bitly/go-simplejson v0.5.0
+	github.com/bitly/go-simplejson v0.5.1
 	github.com/euank/go-kmsg-parser v2.0.0+incompatible
 	github.com/go-redis/redis v6.15.9+incompatible
 	github.com/golang/mock v1.6.0
@@ -44,7 +44,6 @@ require (
 	github.com/andybalholm/brotli v1.0.5 // indirect
 	github.com/benbjohnson/clock v1.1.0 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
-	github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869 // indirect
 	github.com/cenkalti/backoff/v3 v3.0.0 // indirect
 	github.com/cespare/xxhash/v2 v2.1.1 // indirect
 	github.com/cilium/ebpf v0.9.1 // indirect

diff --git a/go.sum b/go.sum
@@ -33,10 +33,8 @@ github.com/beorn7/perks v1.0.0/go.mod h1:KWe93zE9D1o94FZ5RNwFwVgaQK1VOXiVxmqh+Ce
 github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
 github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
 github.com/bgentry/speakeasy v0.1.0/go.mod h1:+zsyZBPWlz7T6j88CTgSN5bM796AkVf0kBD4zp0CCIs=
-github.com/bitly/go-simplejson v0.5.0 h1:6IH+V8/tVMab511d5bn4M7EwGXZf9Hj6i2xSwkNEM+Y=
-github.com/bitly/go-simplejson v0.5.0/go.mod h1:cXHtHw4XUPsvGaxgjIAn8PhEWG9NfngEKAMDJEczWVA=
-github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869 h1:DDGfHa7BWjL4YnC6+E63dPcxHo2sUxDIu8g3QgEJdRY=
-github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869/go.mod h1:Ekp36dRnpXw/yCqJaO+ZrUyxD+3VXMFFr56k5XYrpB4=
+github.com/bitly/go-simplejson v0.5.1 h1:xgwPbetQScXt1gh9BmoJ6j9JMr3TElvuIyjR8pgdoow=
+github.com/bitly/go-simplejson v0.5.1/go.mod h1:YOPVLzCfwK14b4Sff3oP1AmGhI9T9Vsg84etUnlyp+Q=
 github.com/cenkalti/backoff/v3 v3.0.0 h1:ske+9nBpD9qZsTBoF41nW5L+AIuFBKMeze18XQ3eG1c=
 github.com/cenkalti/backoff/v3 v3.0.0/go.mod h1:cIeZDE3IrqwwJl6VUwCN6trj1oXrTS4rc0ij+ULvLYs=
 github.com/cespare/xxhash/v2 v2.1.1 h1:6MnRN8NT7+YBpUIWxHtefFZOKTAPgGjpQSxqLNn0+qY=

diff --git a/plugin/input/file/file_test.go b/plugin/input/file/file_test.go
@@ -101,10 +101,6 @@ func renameFile(oldFile string, newFile string) {
 }
 
 func closeFile(f *os.File) {
-	if err := f.Sync(); err != nil {
-		panic(err)
-	}
-
 	if err := f.Close(); err != nil {
 		panic(err.Error())
 	}
@@ -805,6 +801,7 @@ func TestReadManyPreparedFilesRace(t *testing.T) {
 	if testing.Short() {
 		t.Skip("skip long tests in short mode")
 	}
+
 	lineCount := 2
 	blockCount := 128 * 128
 	fileCount := 32

diff --git a/plugin/input/k8s/k8s_test.go b/plugin/input/k8s/k8s_test.go
@@ -79,23 +79,33 @@ func TestEnrichment(t *testing.T) {
 	putMeta(podInfo)
 	selfNodeName = "node_1"
 
-	var event *pipeline.Event = nil
-	filename := getLogFilename("/k8s-logs", item)
+	var (
+		k8sPod           string
+		k8sNamespace     string
+		k8sContainer     string
+		k8sNode          string
+		k8sNodeLabelZone string
+	)
 	input.SetCommitFn(func(e *pipeline.Event) {
-		event = e
+		k8sPod = strings.Clone(e.Root.Dig("k8s_pod").AsString())
+		k8sNamespace = strings.Clone(e.Root.Dig("k8s_namespace").AsString())
+		k8sContainer = strings.Clone(e.Root.Dig("k8s_container").AsString())
+		k8sNode = strings.Clone(e.Root.Dig("k8s_node").AsString())
+		k8sNodeLabelZone = strings.Clone(e.Root.Dig("k8s_node_label_zone").AsString())
 		wg.Done()
 	})
 
+	filename := getLogFilename("/k8s-logs", item)
 	input.In(0, filename, 0, []byte(`{"time":"time","log":"log\n"}`))
 
 	wg.Wait()
 	p.Stop()
 
-	assert.Equal(t, "advanced-logs-checker-1566485760-trtrq", event.Root.Dig("k8s_pod").AsString(), "wrong event field")
-	assert.Equal(t, "sre", event.Root.Dig("k8s_namespace").AsString(), "wrong event field")
-	assert.Equal(t, "duty-bot", event.Root.Dig("k8s_container").AsString(), "wrong event field")
-	assert.Equal(t, "node_1", event.Root.Dig("k8s_node").AsString(), "wrong event field")
-	assert.Equal(t, "z34", event.Root.Dig("k8s_node_label_zone").AsString(), "wrong event field")
+	assert.Equal(t, "advanced-logs-checker-1566485760-trtrq", k8sPod, "wrong event field")
+	assert.Equal(t, "sre", k8sNamespace, "wrong event field")
+	assert.Equal(t, "duty-bot", k8sContainer, "wrong event field")
+	assert.Equal(t, "node_1", k8sNode, "wrong event field")
+	assert.Equal(t, "z34", k8sNodeLabelZone, "wrong event field")
 }
 
 func TestAllowedLabels(t *testing.T) {

diff --git a/test/test.go b/test/test.go
@@ -74,7 +74,7 @@ func startCasePipeline(act func(pipeline *pipeline.Pipeline), out func(event *pi
 		if x.Load() <= 0 {
 			break
 		}
-		if time.Since(t) > time.Second*20 {
+		if time.Since(t) > time.Minute*2 {
 			panic("too long act")
 		}
 	}
@@ -107,11 +107,11 @@ func NewPipeline(actions []*pipeline.ActionPluginStaticInfo, pipelineOpts ...str
 	if Opts(pipelineOpts).Has("short_event_timeout") {
 		eventTimeout = 10 * time.Millisecond
 	}
-	capacity := 256
 
+	capacity := 256
 	if perf {
 		parallel = true
-		capacity = 10000
+		capacity = 20000
 	}
 
 	settings := &pipeline.Settings{