Configuration for dropping non-sampled transactions (elastic#3702)

* beater/config: add Config.Sampling

* Handle "apm-server.sampling.keep_non_sampled"

Add a publisher.Reporter middleware, which drops
non-sampled transactions.

* tests/system: test sampling.keep_non_sampled

* Add changelog entry

* Non-sampled => unsampled

* Rename approvals

Author: axw

beater/beater.go

@@ -35,6 +35,7 @@ import (
 	"github.com/elastic/apm-server/ingest/pipeline"
 	logs "github.com/elastic/apm-server/log"
 	"github.com/elastic/apm-server/publish"
+	"github.com/elastic/apm-server/sampling"
 )
 
 var (
@@ -172,6 +173,14 @@ func (bt *beater) Run(b *beat.Beat) error {
 	}
 	defer publisher.Stop()
 
+	reporter := publisher.Send
+	if !bt.config.Sampling.KeepUnsampled {
+		// The server has been configured to discard unsampled
+		// transactions. Make sure this is done just before calling
+		// the publisher to avoid affecting aggregations.
+		reporter = sampling.NewDiscardUnsampledReporter(reporter)
+	}
+
 	stopped := make(chan struct{})
 	defer close(stopped)
 	ctx, cancelContext := context.WithCancel(context.Background())
@@ -196,7 +205,7 @@ func (bt *beater) Run(b *beat.Beat) error {
 		Config:   bt.config,
 		Logger:   bt.logger,
 		Tracer:   tracer,
-		Reporter: publisher.Send,
+		Reporter: reporter,
 	})
 }
 

beater/config/config.go

@@ -89,6 +89,7 @@ type Config struct {
 	APIKeyConfig        *APIKeyConfig           `config:"api_key"`
 	JaegerConfig        JaegerConfig            `config:"jaeger"`
 	Aggregation         AggregationConfig       `config:"aggregation"`
+	Sampling            SamplingConfig          `config:"sampling"`
 
 	Pipeline string
 }
@@ -141,6 +142,18 @@ func NewConfig(version string, ucfg *common.Config, outputESCfg *common.Config)
 		return nil, err
 	}
 
+	if !c.Sampling.KeepUnsampled && !c.Aggregation.Enabled {
+		// Unsampled transactions should only be dropped
+		// when transaction aggregation is enabled in the
+		// server. This means the aggregations performed
+		// by the APM UI will not have access to a complete
+		// representation of the latency distribution.
+		logger.Warn("" +
+			"apm-server.sampling.keep_unsampled and " +
+			"apm-server.aggregation.enabled are both false, " +
+			"which will lead to incorrect metrics being reported in the APM UI",
+		)
+	}
 	return c, nil
 }
 
@@ -174,5 +187,6 @@ func DefaultConfig(beatVersion string) *Config {
 		APIKeyConfig: defaultAPIKeyConfig(),
 		JaegerConfig: defaultJaeger(),
 		Aggregation:  defaultAggregationConfig(),
+		Sampling:     defaultSamplingConfig(),
 	}
 }

beater/config/config_test.go

@@ -210,6 +210,9 @@ func Test_UnpackConfig(t *testing.T) {
 					MaxTransactionGroups:           123,
 					HDRHistogramSignificantFigures: 1,
 				},
+				Sampling: SamplingConfig{
+					KeepUnsampled: true,
+				},
 			},
 		},
 		"merge config with default": {
@@ -239,9 +242,10 @@ func Test_UnpackConfig(t *testing.T) {
 						},
 					},
 				},
-				"jaeger.grpc.enabled": true,
-				"api_key.enabled":     true,
-				"aggregation.enabled": true,
+				"jaeger.grpc.enabled":     true,
+				"api_key.enabled":         true,
+				"aggregation.enabled":     true,
+				"sampling.keep_unsampled": false,
 			},
 			outCfg: &Config{
 				Host:            "localhost:3000",
@@ -316,6 +320,9 @@ func Test_UnpackConfig(t *testing.T) {
 					MaxTransactionGroups:           1000,
 					HDRHistogramSignificantFigures: 2,
 				},
+				Sampling: SamplingConfig{
+					KeepUnsampled: false,
+				},
 			},
 		},
 		"kibana trailing slash": {

beater/config/sampling.go

@@ -0,0 +1,33 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package config
+
+// SamplingConfig holds configuration related to sampling.
+type SamplingConfig struct {
+	// KeepUnsampled controls whether unsampled
+	// transactions should be recorded.
+	KeepUnsampled bool `config:"keep_unsampled"`
+}
+
+func defaultSamplingConfig() SamplingConfig {
+	return SamplingConfig{
+		// In a future major release we will set this to
+		// false, and then later remove the option.
+		KeepUnsampled: true,
+	}
+}