func TestWrite_concurrentTempError(t *testing.T) {
	p := newPostgresqlTest(t)
	p.dbConfig.MaxConns = 2
	require.NoError(t, p.Connect())

	// To avoid a race condition, we need to know when our goroutine has started listening to the log.
	wg := sync.WaitGroup{}
	wg.Add(1)

	go func() {
		// Wait for the CREATE TABLE, and then kill the connection.
		// The WaitFor callback holds a lock on the log. Meaning it will block logging of the next action. So we trigger
		// on CREATE TABLE so that there's a few statements to go before the COMMIT.
		p.Logger.WaitFor(func(log Log) bool {
			if strings.Contains(log.String(), "release wg") {
				wg.Done()
			}

			if !strings.Contains(log.String(), "CREATE TABLE") {
				return false
			}
			pid := log.args[1].(MSI)["pid"].(uint32)

			conf := p.db.Config().ConnConfig
			conf.Logger = nil
			c, err := pgx.ConnectConfig(context.Background(), conf)
			if !assert.NoError(t, err) {
				return true
			}
			_, err = c.Exec(context.Background(), "SELECT pg_terminate_backend($1)", pid)
			assert.NoError(t, err)
			return true
		}, false)
	}()
	p.Logger.Infof("release wg")
	wg.Wait()

	metrics := []telegraf.Metric{
		newMetric(t, "_a", MSS{}, MSI{"v": 1}),
	}
	require.NoError(t, p.Write(metrics))

	p.Logger.WaitForCopy(t.Name()+"_a", false)
	dumpA := dbTableDump(t, p.db, "_a")
	assert.Len(t, dumpA, 1)

	haveError := false
	for _, l := range p.Logger.Logs() {
		if strings.Contains(l.String(), "write error") {
			haveError = true
			break
		}
	}
	assert.True(t, haveError, "write error not found in log")
}
// Verify that when using concurrency, errors are not returned, but instead logged and automatically retried
Verify that when using concurrency, errors are not returned, but instead logged and automatically retried
// Verify that when using concurrency errors are not returned but instead logged and automatically retried
func TestWrite_concurrentTempError(t *testing.T) {
	p := newPostgresqlTest(t)
	p.dbConfig.MaxConns = 2
	require.NoError(t, p.Connect())
	wg := sync.WaitGroup{}
	wg.Add(1)
	go func() {
		p.Logger.WaitFor(func(log Log) bool {
			if strings.Contains(log.String(), "release wg") {
				wg.Done()
			}
			if !strings.Contains(log.String(), "CREATE TABLE") {
				return false
			}
			pid := log.args[1].(MSI)["pid"].(uint32)
			conf := p.db.Config().ConnConfig
			conf.Logger = nil
			c, err := pgx.ConnectConfig(context.Background(), conf)
			if !assert.NoError(t, err) {
				return true
			}
			_, err = c.Exec(context.Background(), "SELECT pg_terminate_backend($1)", pid)
			assert.NoError(t, err)
			return true
		}, false)
	}()
	p.Logger.Infof("release wg")
	wg.Wait()
	metrics := []telegraf.Metric{
		newMetric(t, "_a", MSS{}, MSI{"v": 1}),
	}
	require.NoError(t, p.Write(metrics))
	p.Logger.WaitForCopy(t.Name()+"_a", false)
	dumpA := dbTableDump(t, p.db, "_a")
	assert.Len(t, dumpA, 1)
	haveError := false
	for _, l := range p.Logger.Logs() {
		if strings.Contains(l.String(), "write error") {
			haveError = true
			break
		}
	}
	assert.True(t, haveError, "write error not found in log")
}
func TestWrite_concurrentTempError ( t * testing . T ) { p := newPostgresqlTest ( t ) p . dbConfig . MaxConns = 2 require . NoError ( t , p . Connect ( ) ) wg := sync . WaitGroup { } wg . Add ( 1 ) go func ( ) { p . Logger . WaitFor ( func ( log Log ) bool { if strings . Contains ( log . String ( ) , " " ) { wg . Done ( ) } if ! strings . Contains ( log . String ( ) , " " ) { return false } pid := log . args [ 1 ] . ( MSI ) [ " " ] . ( uint32 ) conf := p . db . Config ( ) . ConnConfig conf . Logger = nil c , err := pgx . ConnectConfig ( context . Background ( ) , conf ) if ! assert . NoError ( t , err ) { return true } _ , err = c . Exec ( context . Background ( ) , " " , pid ) assert . NoError ( t , err ) return true } , false ) } ( ) p . Logger . Infof ( " " ) wg . Wait ( ) metrics := [ ] telegraf . Metric { newMetric ( t , " " , MSS { } , MSI { " " : 1 } ) , } require . NoError ( t , p . Write ( metrics ) ) p . Logger . WaitForCopy ( t . Name ( ) + " " , false ) dumpA := dbTableDump ( t , p . db , " " ) assert . Len ( t , dumpA , 1 ) haveError := false for _ , l := range p . Logger . Logs ( ) { if strings . Contains ( l . String ( ) , " " ) { haveError = true break } } assert . True ( t , haveError , " " ) }
Verify that when using concurrency, errors are not returned, but instead logged and automatically retried
Verify that when using concurrency errors are not returned but instead logged and automatically retried
// To avoid a race condition, we need to know when our goroutine has started listening to the log. // Wait for the CREATE TABLE, and then kill the connection. // The WaitFor callback holds a lock on the log. Meaning it will block logging of the next action. So we trigger // on CREATE TABLE so that there's a few statements to go before the COMMIT.