Update go-concert to 0.2.0

CHANGELOG-developer.next.asciidoc

@@ -116,4 +116,5 @@ The list below covers the major changes between 7.0.0-rc2 and master only.
 - Update Go version to 1.15.12. {pull}25629[25629]
 - Update Go version to 1.16.4. {issue}25346[25346] {pull}25671[25671]
 - Add sorting to array fields for generated data files (*-generated.json) {pull}25320[25320]
+- Update to go-concert 0.2.0 {pull}27162[27162]
 - Update Go version to 1.16.5. {issue}26182[26182] {pull}26186[26186]

NOTICE.txt

@@ -6578,11 +6578,11 @@ SOFTWARE
 
 --------------------------------------------------------------------------------
 Dependency : github.com/elastic/go-concert
-Version: v0.1.0
+Version: v0.2.0
 Licence type (autodetected): Apache-2.0
 --------------------------------------------------------------------------------
 
-Contents of probable licence file $GOMODCACHE/github.com/elastic/go-concert@v0.1.0/LICENSE:
+Contents of probable licence file $GOMODCACHE/github.com/elastic/go-concert@v0.2.0/LICENSE:
 
                                  Apache License
                            Version 2.0, January 2004
@@ -15148,11 +15148,11 @@ SOFTWARE.
 
 --------------------------------------------------------------------------------
 Dependency : github.com/urso/sderr
-Version: v0.0.0-20200210124243-c2a16f3d43ec
+Version: v0.0.0-20210525210834-52b04e8f5c71
 Licence type (autodetected): Apache-2.0
 --------------------------------------------------------------------------------
 
-Contents of probable licence file $GOMODCACHE/github.com/urso/sderr@v0.0.0-20200210124243-c2a16f3d43ec/LICENSE:
+Contents of probable licence file $GOMODCACHE/github.com/urso/sderr@v0.0.0-20210525210834-52b04e8f5c71/LICENSE:
 
                                  Apache License
                            Version 2.0, January 2004

filebeat/input/filestream/filestream.go

@@ -18,6 +18,7 @@
 package filestream
 
 import (
+	"context"
 	"errors"
 	"io"
 	"os"
@@ -136,14 +137,14 @@ func (f *logFile) Read(buf []byte) (int, error) {
 
 func (f *logFile) startFileMonitoringIfNeeded() {
 	if f.closeInactive > 0 || f.closeRemoved || f.closeRenamed {
-		f.tg.Go(func(ctx unison.Canceler) error {
+		f.tg.Go(func(ctx context.Context) error {
 			f.periodicStateCheck(ctx)
 			return nil
 		})
 	}
 
 	if f.closeAfterInterval > 0 {
-		f.tg.Go(func(ctx unison.Canceler) error {
+		f.tg.Go(func(ctx context.Context) error {
 			f.closeIfTimeout(ctx)
 			return nil
 		})

filebeat/input/filestream/internal/input-logfile/harvester.go

@@ -158,10 +158,10 @@ func (hg *defaultHarvesterGroup) Restart(ctx input.Context, s Source) {
 	hg.tg.Go(startHarvester(ctx, hg, s, true))
 }
 
-func startHarvester(ctx input.Context, hg *defaultHarvesterGroup, s Source, restart bool) func(canceler unison.Canceler) error {
+func startHarvester(ctx input.Context, hg *defaultHarvesterGroup, s Source, restart bool) func(context.Context) error {
 	srcID := hg.identifier.ID(s)
 
-	return func(canceler unison.Canceler) error {
+	return func(canceler context.Context) error {
 		defer func() {
 			if v := recover(); v != nil {
 				err := fmt.Errorf("harvester panic with: %+v\n%s", v, debug.Stack())
@@ -226,7 +226,7 @@ func (hg *defaultHarvesterGroup) Continue(ctx input.Context, previous, next Sour
 	prevID := hg.identifier.ID(previous)
 	nextID := hg.identifier.ID(next)
 
-	hg.tg.Go(func(canceler unison.Canceler) error {
+	hg.tg.Go(func(canceler context.Context) error {
 		previousResource, err := lock(ctx, hg.store, prevID)
 		if err != nil {
 			return fmt.Errorf("error while locking previous resource: %v", err)
@@ -252,7 +252,7 @@ func (hg *defaultHarvesterGroup) Continue(ctx input.Context, previous, next Sour
 
 // Stop stops the running Harvester for a given Source.
 func (hg *defaultHarvesterGroup) Stop(s Source) {
-	hg.tg.Go(func(_ unison.Canceler) error {
+	hg.tg.Go(func(_ context.Context) error {
 		hg.readers.remove(hg.identifier.ID(s))
 		return nil
 	})

filebeat/input/filestream/internal/input-logfile/input.go

@@ -70,7 +70,9 @@ func (inp *managedInput) Run(
 		store:        groupStore,
 		ackCH:        inp.ackCH,
 		identifier:   inp.sourceIdentifier,
-		tg:           unison.TaskGroup{},
+		tg: unison.TaskGroup{
+			OnQuit: unison.ContinueOnErrors, // harvester should keep running if a single harvester errored
+		},
 	}
 
 	prospectorStore := inp.manager.getRetainedStore()

filebeat/input/filestream/internal/input-logfile/manager.go

@@ -18,6 +18,7 @@
 package input_logfile
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"strings"
@@ -128,7 +129,7 @@ func (cim *InputManager) Init(group unison.Group, mode v2.Mode) error {
 
 	store := cim.getRetainedStore()
 	cleaner := &cleaner{log: log}
-	err := group.Go(func(canceler unison.Canceler) error {
+	err := group.Go(func(canceler context.Context) error {
 		defer cim.shutdown()
 		defer store.Release()
 		interval := cim.StateStore.CleanupInterval()

filebeat/input/filestream/internal/input-logfile/update_writer.go

@@ -21,7 +21,6 @@ import (
 	"context"
 	"sync"
 
-	"github.com/elastic/go-concert/ctxtool"
 	"github.com/elastic/go-concert/unison"
 )
 
@@ -62,8 +61,8 @@ func newUpdateWriter(store *store, ch *updateChan) *updateWriter {
 		store: store,
 		ch:    ch,
 	}
-	w.tg.Go(func(ctx unison.Canceler) error {
-		w.run(ctxtool.FromCanceller(ctx))
+	w.tg.Go(func(ctx context.Context) error {
+		w.run(ctx)
 		return nil
 	})
 

filebeat/input/v2/input-cursor/manager.go

@@ -18,6 +18,7 @@
 package cursor
 
 import (
+	"context"
 	"errors"
 	"sync"
 	"time"
@@ -119,7 +120,7 @@ func (cim *InputManager) Init(group unison.Group, mode v2.Mode) error {
 	store := cim.store
 	cleaner := &cleaner{log: log}
 	store.Retain()
-	err := group.Go(func(canceler unison.Canceler) error {
+	err := group.Go(func(canceler context.Context) error {
 		defer cim.shutdown()
 		defer store.Release()
 		interval := cim.StateStore.CleanupInterval()

filebeat/inputsource/common/dgram/server.go

@@ -103,7 +103,7 @@ func (l *Listener) Start() error {
 		return err
 	}
 
-	l.tg.Go(func(ctx unison.Canceler) error {
+	l.tg.Go(func(ctx context.Context) error {
 		connCtx, connCancel := ctxtool.WithFunc(ctxtool.FromCanceller(ctx), func() {
 			conn.Close()
 		})

go.mod

@@ -63,7 +63,7 @@ require (
 	github.com/eclipse/paho.mqtt.golang v1.2.1-0.20200121105743-0d940dd29fd2
 	github.com/elastic/ecs v1.10.0
 	github.com/elastic/elastic-agent-client/v7 v7.0.0-20210727140539-f0905d9377f6
-	github.com/elastic/go-concert v0.1.0
+	github.com/elastic/go-concert v0.2.0
 	github.com/elastic/go-libaudit/v2 v2.2.0
 	github.com/elastic/go-licenser v0.3.1
 	github.com/elastic/go-lookslike v0.3.0
@@ -152,7 +152,7 @@ require (
 	github.com/tsg/go-daemon v0.0.0-20200207173439-e704b93fd89b
 	github.com/tsg/gopacket v0.0.0-20200626092518-2ab8e397a786
 	github.com/ugorji/go/codec v1.1.8
-	github.com/urso/sderr v0.0.0-20200210124243-c2a16f3d43ec
+	github.com/urso/sderr v0.0.0-20210525210834-52b04e8f5c71
 	github.com/vmware/govmomi v0.0.0-20170802214208-2cad15190b41
 	github.com/xdg/scram v0.0.0-20180814205039-7eeb5667e42c
 	github.com/yuin/gopher-lua v0.0.0-20170403160031-b402f3114ec7 // indirect

go.sum

@@ -253,8 +253,8 @@ github.com/elastic/elastic-agent-client/v7 v7.0.0-20210727140539-f0905d9377f6 h1
 github.com/elastic/elastic-agent-client/v7 v7.0.0-20210727140539-f0905d9377f6/go.mod h1:uh/Gj9a0XEbYoM4NYz4LvaBVARz3QXLmlNjsrKY9fTc=
 github.com/elastic/fsevents v0.0.0-20181029231046-e1d381a4d270 h1:cWPqxlPtir4RoQVCpGSRXmLqjEHpJKbR60rxh1nQZY4=
 github.com/elastic/fsevents v0.0.0-20181029231046-e1d381a4d270/go.mod h1:Msl1pdboCbArMF/nSCDUXgQuWTeoMmE/z8607X+k7ng=
-github.com/elastic/go-concert v0.1.0 h1:gz/yvA3bseuHzoF/lNMltkL30XdPqMo+bg5o2mBx2EE=
-github.com/elastic/go-concert v0.1.0/go.mod h1:9MtFarjXroUgmm0m6HY3NSe1XiKhdktiNRRj9hWvIaM=
+github.com/elastic/go-concert v0.2.0 h1:GAQrhRVXprnNjtvTP9pWJ1d4ToEA4cU5ci7TwTa20xg=
+github.com/elastic/go-concert v0.2.0/go.mod h1:HWjpO3IAEJUxOeaJOWXWEp7imKd27foxz9V5vegC/38=
 github.com/elastic/go-libaudit/v2 v2.2.0 h1:TY3FDpG4Zr9Qnv6KYW6olYr/U+nfu0rD2QAbv75VxMQ=
 github.com/elastic/go-libaudit/v2 v2.2.0/go.mod h1:MM/l/4xV7ilcl+cIblL8Zn448J7RZaDwgNLE4gNKYPg=
 github.com/elastic/go-licenser v0.3.1 h1:RmRukU/JUmts+rpexAw0Fvt2ly7VVu6mw8z4HrEzObU=
@@ -719,8 +719,8 @@ github.com/urso/magetools v0.0.0-20190919040553-290c89e0c230 h1:Ft1EJ6JL0F/RV6o2
 github.com/urso/magetools v0.0.0-20190919040553-290c89e0c230/go.mod h1:DFxTNgS/ExCGmmjVjSOgS2WjtfjKXgCyDzAFgbtovSA=
 github.com/urso/qcgen v0.0.0-20180131103024-0b059e7db4f4 h1:hhA8EBThzz9PztawVTycKvfETVuBqxAQ5keFlAVtbAw=
 github.com/urso/qcgen v0.0.0-20180131103024-0b059e7db4f4/go.mod h1:RspW+E2Yb7Fs7HclB2tiDaiu6Rp41BiIG4Wo1YaoXGc=
-github.com/urso/sderr v0.0.0-20200210124243-c2a16f3d43ec h1:HkZIDJrMKZHPsYhmH2XjTTSk1pbMCFfpxSnyzZUFm+k=
-github.com/urso/sderr v0.0.0-20200210124243-c2a16f3d43ec/go.mod h1:Wp40HwmjM59FkDIVFfcCb9LzBbnc0XAMp8++hJuWvSU=
+github.com/urso/sderr v0.0.0-20210525210834-52b04e8f5c71 h1:CehQeKbysHV8J2V7AD0w8NL2x1h04kmmo/Ft5su4lU0=
+github.com/urso/sderr v0.0.0-20210525210834-52b04e8f5c71/go.mod h1:Wp40HwmjM59FkDIVFfcCb9LzBbnc0XAMp8++hJuWvSU=
 github.com/vbatts/tar-split v0.11.1/go.mod h1:LEuURwDEiWjRjwu46yU3KVGuUdVv/dcnpcEPSzR8z6g=
 github.com/vmware/govmomi v0.0.0-20170802214208-2cad15190b41 h1:NeNpIvfvaFOh0BH7nMEljE5Rk/VJlxhm58M41SeOD20=
 github.com/vmware/govmomi v0.0.0-20170802214208-2cad15190b41/go.mod h1:URlwyTFZX72RmxtxuaFL2Uj3fD1JTvZdx59bHWk6aFU=