WIP in intake v2.

Author: Ron cohen

beater/handlers.go

@@ -18,11 +18,12 @@ import (
 
 	conf "github.com/elastic/apm-server/config"
 	"github.com/elastic/apm-server/decoder"
+	"github.com/elastic/apm-server/model"
 	"github.com/elastic/apm-server/processor"
 	perr "github.com/elastic/apm-server/processor/error"
-	"github.com/elastic/apm-server/processor/healthcheck"
 	"github.com/elastic/apm-server/processor/sourcemap"
 	"github.com/elastic/apm-server/processor/transaction"
+
 	"github.com/elastic/apm-server/utility"
 	"github.com/elastic/beats/libbeat/logp"
 	"github.com/elastic/beats/libbeat/monitoring"
@@ -35,10 +36,13 @@ const (
 	FrontendErrorsURL       = "/v1/client-side/errors"
 	HealthCheckURL          = "/healthcheck"
 	SourcemapsURL           = "/v1/client-side/sourcemaps"
+	StreamBackendURL        = "/v2/intake"
 
 	rateLimitCacheSize       = 1000
 	rateLimitBurstMultiplier = 2
 
+	v2TransformBatchSize = 100
+
 	supportedHeaders = "Content-Type, Content-Encoding, Accept"
 	supportedMethods = "POST, OPTIONS"
 )
@@ -58,6 +62,10 @@ type serverResponse struct {
 	counter *monitoring.Int
 }
 
+func (s *serverResponse) IsError() bool {
+	return s.err != nil
+}
+
 var (
 	serverMetrics = monitoring.Default.NewRegistry("apm-server.server", monitoring.PublishExpvar)
 	counter       = func(s string) *monitoring.Int {
@@ -117,13 +125,18 @@ var (
 		}
 	}
 
+	errHeaderMissing = errors.New("header must be first object in stream")
+
 	Routes = map[string]routeMapping{
 		BackendTransactionsURL:  {backendHandler, transaction.NewProcessor},
 		FrontendTransactionsURL: {frontendHandler, transaction.NewProcessor},
-		BackendErrorsURL:        {backendHandler, perr.NewProcessor},
-		FrontendErrorsURL:       {frontendHandler, perr.NewProcessor},
-		HealthCheckURL:          {healthCheckHandler, healthcheck.NewProcessor},
-		SourcemapsURL:           {sourcemapHandler, sourcemap.NewProcessor},
+
+		StreamBackendURL: {streamBackendHandler, nil},
+
+		BackendErrorsURL:  {backendHandler, perr.NewProcessor},
+		FrontendErrorsURL: {frontendHandler, perr.NewProcessor},
+		HealthCheckURL:    {healthCheckHandler, nil},
+		SourcemapsURL:     {sourcemapHandler, sourcemap.NewProcessor},
 	}
 )
 
@@ -163,19 +176,29 @@ func concurrencyLimitHandler(beaterConfig *Config, h http.Handler) http.Handler
 }
 
 func backendHandler(pf ProcessorFactory, beaterConfig *Config, report reporter) http.Handler {
+	extractors := []decoder.Extractor{}
+	if beaterConfig.AugmentEnabled {
+		extractors = append(extractors, decoder.SystemExtractor)
+	}
+
 	return logHandler(
 		concurrencyLimitHandler(beaterConfig,
 			authHandler(beaterConfig.SecretToken,
-				processRequestHandler(pf, conf.Config{}, report,
-					decoder.DecodeSystemData(decoder.DecodeLimitJSONData(beaterConfig.MaxUnzippedSize), beaterConfig.AugmentEnabled)))))
+				processRequestHandler(pf, conf.TransformConfig{}, report, extractors,
+					decoder.DecodeLimitJSONData(beaterConfig.MaxUnzippedSize)))))
 }
 
 func frontendHandler(pf ProcessorFactory, beaterConfig *Config, report reporter) http.Handler {
+	extractors := []decoder.Extractor{}
+	if beaterConfig.AugmentEnabled {
+		extractors = append(extractors, decoder.UserExtractor)
+	}
+
 	smapper, err := beaterConfig.Frontend.memoizedSmapMapper()
 	if err != nil {
 		logp.NewLogger("handler").Error(err.Error())
 	}
-	config := conf.Config{
+	config := conf.TransformConfig{
 		SmapMapper:          smapper,
 		LibraryPattern:      regexp.MustCompile(beaterConfig.Frontend.LibraryPattern),
 		ExcludeFromGrouping: regexp.MustCompile(beaterConfig.Frontend.ExcludeFromGrouping),
@@ -185,8 +208,21 @@ func frontendHandler(pf ProcessorFactory, beaterConfig *Config, report reporter)
 			concurrencyLimitHandler(beaterConfig,
 				ipRateLimitHandler(beaterConfig.Frontend.RateLimit,
 					corsHandler(beaterConfig.Frontend.AllowOrigins,
-						processRequestHandler(pf, config, report,
-							decoder.DecodeUserData(decoder.DecodeLimitJSONData(beaterConfig.MaxUnzippedSize), beaterConfig.AugmentEnabled)))))))
+						processRequestHandler(pf, config, report, extractors,
+							decoder.DecodeLimitJSONData(beaterConfig.MaxUnzippedSize)))))))
+}
+
+func streamBackendHandler(_ ProcessorFactory, beaterConfig *Config, report reporter) http.Handler {
+	extractors := []decoder.Extractor{
+		decoder.SystemExtractor,
+	}
+
+	requestDecodeer := decoder.StreamDecodeLimitJSONData(beaterConfig.MaxUnzippedSize)
+
+	return logHandler(
+		concurrencyLimitHandler(beaterConfig,
+			authHandler(beaterConfig.SecretToken,
+				processStreamRequest(v2TransformBatchSize, conf.TransformConfig{}, report, extractors, requestDecodeer))))
 }
 
 func sourcemapHandler(pf ProcessorFactory, beaterConfig *Config, report reporter) http.Handler {
@@ -197,7 +233,7 @@ func sourcemapHandler(pf ProcessorFactory, beaterConfig *Config, report reporter
 	return logHandler(
 		killSwitchHandler(beaterConfig.Frontend.isEnabled(),
 			authHandler(beaterConfig.SecretToken,
-				processRequestHandler(pf, conf.Config{SmapMapper: smapper}, report, decoder.DecodeSourcemapFormData))))
+				processRequestHandler(pf, conf.TransformConfig{SmapMapper: smapper}, report, []decoder.Extractor{}, decoder.DecodeSourcemapFormData))))
 }
 
 func healthCheckHandler(_ ProcessorFactory, _ *Config, _ reporter) http.Handler {
@@ -346,14 +382,14 @@ func corsHandler(allowedOrigins []string, h http.Handler) http.Handler {
 	})
 }
 
-func processRequestHandler(pf ProcessorFactory, config conf.Config, report reporter, decode decoder.Decoder) http.Handler {
+func processRequestHandler(pf ProcessorFactory, config conf.TransformConfig, report reporter, extractors []decoder.Extractor, decode func(req *http.Request) (map[string]interface{}, error)) http.Handler {
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		res := processRequest(r, pf, config, report, decode)
+		res := processRequest(r, pf, config, report, extractors, decode)
 		sendStatus(w, r, res)
 	})
 }
 
-func processRequest(r *http.Request, pf ProcessorFactory, config conf.Config, report reporter, decode decoder.Decoder) serverResponse {
+func processRequest(r *http.Request, pf ProcessorFactory, config conf.TransformConfig, report reporter, extractors []decoder.Extractor, decode func(req *http.Request) (map[string]interface{}, error)) serverResponse {
 	processor := pf()
 
 	if r.Method != "POST" {
@@ -369,16 +405,18 @@ func processRequest(r *http.Request, pf ProcessorFactory, config conf.Config, re
 
 	}
 
-	if err = processor.Validate(data); err != nil {
-		return cannotValidateResponse(err)
+	transformables, tctx, response := ProcessPayload(data, processor)
+	if response.err != nil {
+		return response
 	}
 
-	payload, err := processor.Decode(data)
-	if err != nil {
-		return cannotDecodeResponse(err)
+	req := pendingReq{
+		transformable: transformables,
+		config:        config,
+		context:       tctx,
 	}
 
-	if err = report(pendingReq{payload: payload, config: config}); err != nil {
+	if err = report(req); err != nil {
 		if strings.Contains(err.Error(), "publisher is being stopped") {
 			return serverShuttingDownResponse(err)
 		}
@@ -388,6 +426,24 @@ func processRequest(r *http.Request, pf ProcessorFactory, config conf.Config, re
 	return acceptedResponse
 }
 
+func ProcessPayload(data map[string]interface{}, p processor.Processor) (model.TransformableBatch, *model.TransformContext, serverResponse) {
+	var err error
+	if err = p.Validate(data); err != nil {
+		return nil, nil, cannotValidateResponse(err)
+	}
+
+	transformationContext, err := model.DecodeContext(data, err)
+	if err != nil {
+		return nil, nil, cannotDecodeResponse(err)
+	}
+
+	transformables, err := p.Decode(data)
+	if err != nil {
+		return nil, nil, cannotDecodeResponse(err)
+	}
+	return transformables, transformationContext, okResponse
+}
+
 func sendStatus(w http.ResponseWriter, r *http.Request, res serverResponse) {
 	contentType := "text/plain; charset=utf-8"
 	if acceptsJSON(r) {
@@ -397,7 +453,10 @@ func sendStatus(w http.ResponseWriter, r *http.Request, res serverResponse) {
 	w.WriteHeader(res.code)
 
 	responseCounter.Inc()
-	res.counter.Inc()
+	if res.counter != nil {
+		res.counter.Inc()
+	}
+
 	if res.err == nil {
 		responseSuccesses.Inc()
 		return

beater/metrics.go

@@ -0,0 +1,35 @@
+package beater
+
+// import "github.com/elastic/beats/libbeat/monitoring"
+
+// type EntityMetrics struct {
+// 	registry        *monitoring.Registry
+// 	validationCount *monitoring.Int
+// 	validationError *monitoring.Int
+// 	transformations *monitoring.Int
+// }
+
+// var errorMetricsReg = monitoring.Default.NewRegistry("apm-server.entity.error")
+
+// var ErrorMetrics = EntityHandlingMetrics{
+// 	registry:        errorMetricsReg,
+// 	transformations: monitoring.NewInt(errorMetricsReg, "transformations"),
+// 	validationTotal: monitoring.NewInt(errorMetricsReg, "validation.total"),
+// 	validationError: monitoring.NewInt(errorMetricsReg, "validation.errors"),
+// }
+
+// var transactionMetricsReg = monitoring.Default.NewRegistry("apm-server.entity.transaction")
+
+// var TransactionMetrics = EntityHandlingMetrics{
+// 	registry:        transactionMetricsReg,
+// 	transformations: monitoring.NewInt(transactionMetricsReg, "transformations"),
+// 	validationTotal: monitoring.NewInt(transactionMetricsReg, "validation.total"),
+// 	validationError: monitoring.NewInt(transactionMetricsReg, "validation.errors"),
+// }
+
+// // var errorMetrics =
+
+// // var validationCount =
+
+// // validationError = monitoring.NewInt(errorMetrics, "validation.errors")
+// // transformations = monitoring.NewInt(errorMetrics, "transformations")

beater/pub.go

@@ -8,7 +8,7 @@ import (
 	"github.com/pkg/errors"
 
 	"github.com/elastic/apm-server/config"
-	pr "github.com/elastic/apm-server/processor"
+	"github.com/elastic/apm-server/model"
 	"github.com/elastic/beats/libbeat/beat"
 )
 
@@ -27,8 +27,9 @@ type publisher struct {
 }
 
 type pendingReq struct {
-	payload pr.Payload
-	config  config.Config
+	transformable []model.Transformable
+	config        config.TransformConfig
+	context       *model.TransformContext
 }
 
 var (
@@ -102,6 +103,9 @@ func (p *publisher) Send(req pendingReq) error {
 
 func (p *publisher) run() {
 	for req := range p.pendingRequests {
-		p.client.PublishAll(req.payload.Transform(req.config))
+		for _, transformable := range req.transformable {
+			// todo: is this threadsafe?
+			p.client.Publish(transformable.Transform(req.config, req.context))
+		}
 	}
 }