sampling: fix pubsub implementation

The initial implementation was written as a ~quick hack, with the
expectation that it would be replaced by the Changes API. It was
broken due to its ignorance of data streams, and multi-shard indices.
Sequence numbers are only comparable within a single shard.

Given that there is no known delivery date for the Changes API,
we propose to instead revise the pubsub implementation to address
the problems by:

 - enforcing single-shard indices for sampled trace data streams
 - searching (now single-shard) backing indices individually

In addition, we now use global checkpoints to bound searches, and
use PIT (point in time) for paging through results. Querying underlying
indices and global checkpoints requires an additional "monitor" index
privilege.

Author: axw

apmpackage/apm/0.2.0/data_stream/sampled_traces/manifest.yml

@@ -2,3 +2,10 @@ title: APM tail-sampled traces
 type: traces
 dataset: sampled
 ilm_policy: traces-apm.sampled-default_policy
+elasticsearch:
+  index_template:
+    settings:
+      # Create a single shard per index, so we can use
+      # global checkpoints as a way of limiting search
+      # results.
+      number_of_shards: 1

x-pack/apm-server/sampling/pubsub/checkpoints.go

@@ -0,0 +1,83 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package pubsub
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+
+	"github.com/pkg/errors"
+
+	"github.com/elastic/go-elasticsearch/v7/esapi"
+
+	"github.com/elastic/apm-server/elasticsearch"
+)
+
+// getGlobalCheckpoints returns the current global checkpoint for each index
+// underlying dataStream. Each index is required to have a single (primary) shard.
+func getGlobalCheckpoints(
+	ctx context.Context,
+	client elasticsearch.Client,
+	dataStream string,
+) (map[string]int64, error) {
+	indexGlobalCheckpoints := make(map[string]int64)
+	resp, err := esapi.IndicesStatsRequest{
+		Index: []string{dataStream},
+		Level: "shards",
+	}.Do(ctx, client)
+	if err != nil {
+		return nil, errors.New("index stats request failed")
+	}
+	defer resp.Body.Close()
+	if resp.IsError() {
+		switch resp.StatusCode {
+		case http.StatusNotFound:
+			// Data stream does not yet exist.
+			return indexGlobalCheckpoints, nil
+		}
+		message, _ := ioutil.ReadAll(resp.Body)
+		return nil, fmt.Errorf("index stats request failed: %s", message)
+	}
+
+	var stats dataStreamStats
+	if err := json.NewDecoder(resp.Body).Decode(&stats); err != nil {
+		return nil, err
+	}
+
+	for index, indexStats := range stats.Indices {
+		if n := len(indexStats.Shards); n > 1 {
+			return nil, fmt.Errorf("expected 1 shard, got %d for index %q", n, index)
+		}
+		for _, shardStats := range indexStats.Shards {
+			for _, shardStats := range shardStats {
+				if shardStats.Routing.Primary {
+					indexGlobalCheckpoints[index] = shardStats.SeqNo.GlobalCheckpoint
+					break
+				}
+			}
+		}
+	}
+	return indexGlobalCheckpoints, nil
+}
+
+type dataStreamStats struct {
+	Indices map[string]indexStats `json:"indices"`
+}
+
+type indexStats struct {
+	Shards map[string][]shardStats `json:"shards"`
+}
+
+type shardStats struct {
+	Routing struct {
+		Primary bool `json:"primary"`
+	} `json:"routing"`
+	SeqNo struct {
+		GlobalCheckpoint int64 `json:"global_checkpoint"`
+	} `json:"seq_no"`
+}

x-pack/apm-server/sampling/pubsub/datastream.go

@@ -87,7 +87,8 @@ const dataStreamIndexTemplate = `{
   "data_stream": {},
   "template": {
     "settings": {
-      "index.lifecycle.name": %q
+      "index.lifecycle.name": %q,
+      "index.number_of_shards": 1
     },
     "mappings": {
       "properties": {