sampling/pubsub: fix subscriber

[axw]

Motivation/summary

The tail-based sampling subscriber can enter an infinite loop when one of the sampled traces indices contains more than 1000 documents. In this case, we never update the _seq_no filter and continuously issue the same search in an infinite loop.

We fix this by by setting the minimum bound on subsequent searches to the _seq_no returned by the previous search.

Checklist

• Update CHANGELOG.asciidoc
  - [ ] Update package changelog.yml (only if changes to apmpackage have been made)
  - [ ] Documentation has been updated

How to test these changes

See #6642 (comment)

Related issues

Closes #6639
Closes #6642

[apmmachine]

💚 Build Succeeded

the below badges are clickable and redirect to their specific view in the CI or DOCS

Expand to view the summary

Build stats

• Start Time: 2022-02-08T01:42:34.804+0000

• Duration: 86 min 42 sec

Test stats 🧪

Test	Results
Failed	0
Passed	5374
Skipped	18
Total	5392

🤖 GitHub comments

To re-run your PR in the CI, just comment with:

• /test : Re-trigger the build.

• /hey-apm : Run the hey-apm benchmark.

• /package : Generate and publish the docker images.

• run elasticsearch-ci/docs : Re-trigger the docs validation. (use unformatted text in the comment!)

[axw]

/test

[axw]

/test

[stuartnelson3]

Confirmed with bc2.

Steps:

1. Started stack (docker-compose up -d), enabled internal stack monitoring
2. Added default tail-sampling policy to apm-server.yml:

   apm-server:
     sampling:
       tail:
         enabled: true
         interval: 10s
         policies:
         - default:
           sample_rate: 0.5

3. Ran two APM Servers:
   1. ./apm-server -c apm-server.dev.yml -e
   2. ./apm-server -c apm-server.dev.yml -e -E apm-server.host=localhost:8201 -E path.data=/tmp
4. Ran the below program 5 times; waited for docs to be indexed (watching Elasticsearch stack monitoring); repeat once more
5. Search rate and CPU do not increase; graph indicates documents are being indexed

package main

import (
        "os"

        "go.elastic.co/apm"
        "go.elastic.co/apm/transport"
)

func main() {
	os.Setenv("ELASTIC_APM_SERVER_URL", "http://localhost:8200")
	transport1, _ := transport.NewHTTPTransport()
	tracer1, _ := apm.NewTracerOptions(apm.TracerOptions{
		ServiceName: "svc1",
		Transport:   transport1,
	})
	defer tracer1.Flush(nil)

	os.Setenv("ELASTIC_APM_SERVER_URL", "http://localhost:8201")
	transport2, _ := transport.NewHTTPTransport()
	tracer2, _ := apm.NewTracerOptions(apm.TracerOptions{
		ServiceName: "svc2",
		Transport:   transport2,
	})
	defer tracer2.Flush(nil)

	for i := 0; i < 500; i++ {
		tx1 := tracer1.StartTransaction("tx1", "type")
		span := tx1.StartSpan("span", "type", nil)
		tx2 := tracer2.StartTransactionOptions("tx2", "type", apm.TransactionOptions{
			TraceContext: span.TraceContext(),
		})
		tx2.End()
		span.End()
		tx1.End()
	}
}