From 963e3615f62a971eb0f6c20d96577e6ef5c84fda Mon Sep 17 00:00:00 2001 From: Vishal Raj Date: Thu, 17 Nov 2022 07:27:01 +0100 Subject: [PATCH 1/3] Register partial transaction in aws extn --- modelwriter.go | 69 +++++++++++++++++---------------- module/apmlambda/go.mod | 1 + module/apmlambda/lambda.go | 43 ++++++++++++++++++++ module/apmlambda/lambda_test.go | 36 +++++++++++++++++ 4 files changed, 115 insertions(+), 34 deletions(-) create mode 100644 module/apmlambda/lambda_test.go diff --git a/modelwriter.go b/modelwriter.go index 2d7d2825b..cef59c0eb 100644 --- a/modelwriter.go +++ b/modelwriter.go @@ -48,7 +48,7 @@ type modelWriter struct { // writeTransaction encodes tx as JSON to the buffer, and then resets tx. func (w *modelWriter) writeTransaction(tx *Transaction, td *TransactionData) { var modelTx model.Transaction - w.buildModelTransaction(&modelTx, tx, td) + BuildModelTransaction(&modelTx, tx, td) w.json.RawString(`{"transaction":`) modelTx.MarshalFastJSON(&w.json) w.json.RawByte('}') @@ -103,39 +103,6 @@ func (w *modelWriter) writeMetrics(m *Metrics) { m.reset() } -func (w *modelWriter) buildModelTransaction(out *model.Transaction, tx *Transaction, td *TransactionData) { - out.ID = model.SpanID(tx.traceContext.Span) - out.TraceID = model.TraceID(tx.traceContext.Trace) - sampled := tx.traceContext.Options.Recorded() - if !sampled { - out.Sampled = ¬Sampled - } - if tx.traceContext.State.haveSampleRate { - out.SampleRate = &tx.traceContext.State.sampleRate - } - - out.ParentID = model.SpanID(tx.parentID) - out.Name = truncateString(td.Name) - out.Type = truncateString(td.Type) - out.Result = truncateString(td.Result) - out.Outcome = normalizeOutcome(td.Outcome) - out.Timestamp = model.Time(td.timestamp.UTC()) - out.Duration = td.Duration.Seconds() * 1000 - out.SpanCount.Started = td.spansCreated - out.SpanCount.Dropped = td.spansDropped - out.OTel = td.Context.otel - for _, sl := range td.links { - out.Links = append(out.Links, model.SpanLink{TraceID: model.TraceID(sl.Trace), SpanID: model.SpanID(sl.Span)}) - } - if dss := buildDroppedSpansStats(td.droppedSpansStats); len(dss) > 0 { - out.DroppedSpansStats = dss - } - - if sampled { - out.Context = td.Context.build() - } -} - func (w *modelWriter) buildModelSpan(out *model.Span, span *Span, sd *SpanData) { w.modelStacktrace = w.modelStacktrace[:0] out.ID = model.SpanID(span.traceContext.Span) @@ -258,6 +225,40 @@ func (w *modelWriter) buildModelError(out *model.Error, e *ErrorData) { out.Culprit = truncateString(out.Culprit) } +// BuildModelTransaction converts apm transaction to model transaction +func BuildModelTransaction(out *model.Transaction, tx *Transaction, td *TransactionData) { + out.ID = model.SpanID(tx.traceContext.Span) + out.TraceID = model.TraceID(tx.traceContext.Trace) + sampled := tx.traceContext.Options.Recorded() + if !sampled { + out.Sampled = ¬Sampled + } + if tx.traceContext.State.haveSampleRate { + out.SampleRate = &tx.traceContext.State.sampleRate + } + + out.ParentID = model.SpanID(tx.parentID) + out.Name = truncateString(td.Name) + out.Type = truncateString(td.Type) + out.Result = truncateString(td.Result) + out.Outcome = normalizeOutcome(td.Outcome) + out.Timestamp = model.Time(td.timestamp.UTC()) + out.Duration = td.Duration.Seconds() * 1000 + out.SpanCount.Started = td.spansCreated + out.SpanCount.Dropped = td.spansDropped + out.OTel = td.Context.otel + for _, sl := range td.links { + out.Links = append(out.Links, model.SpanLink{TraceID: model.TraceID(sl.Trace), SpanID: model.SpanID(sl.Span)}) + } + if dss := buildDroppedSpansStats(td.droppedSpansStats); len(dss) > 0 { + out.DroppedSpansStats = dss + } + + if sampled { + out.Context = td.Context.build() + } +} + func stacktraceCulprit(frames []model.StacktraceFrame) string { for _, frame := range frames { if !frame.LibraryFrame { diff --git a/module/apmlambda/go.mod b/module/apmlambda/go.mod index d149c0b7a..1cb8ca81a 100644 --- a/module/apmlambda/go.mod +++ b/module/apmlambda/go.mod @@ -3,6 +3,7 @@ module go.elastic.co/apm/module/apmlambda/v2 require ( github.com/aws/aws-lambda-go v1.8.0 go.elastic.co/apm/v2 v2.2.0 + go.elastic.co/fastjson v1.1.0 ) replace go.elastic.co/apm/v2 => ../.. diff --git a/module/apmlambda/lambda.go b/module/apmlambda/lambda.go index 449a84640..7863d772c 100644 --- a/module/apmlambda/lambda.go +++ b/module/apmlambda/lambda.go @@ -18,8 +18,10 @@ package apmlambda // import "go.elastic.co/apm/module/apmlambda/v2" import ( + "bytes" "log" "net" + "net/http" "net/rpc" "os" "unicode/utf8" @@ -28,7 +30,9 @@ import ( "github.com/aws/aws-lambda-go/lambdacontext" "go.elastic.co/apm/v2" + "go.elastic.co/apm/v2/model" "go.elastic.co/apm/v2/stacktrace" + "go.elastic.co/fastjson" ) const ( @@ -51,6 +55,10 @@ var ( Request string `json:"request,omitempty"` Response string `json:"response,omitempty"` } + + jsonw fastjson.Writer + + ignoreTxnRegistration bool ) func init() { @@ -72,6 +80,17 @@ func (f *Function) Ping(req *messages.PingRequest, response *messages.PingRespon return f.client.Call("Function.Ping", req, response) } +func createPartialTransactionJSON(apmTx *apm.Transaction, w *fastjson.Writer) error { + var tx model.Transaction + apm.BuildModelTransaction(&tx, apmTx, apmTx.TransactionData) + w.RawString(`{"transaction":`) + if err := tx.MarshalFastJSON(w); err != nil { + return err + } + w.RawByte('}') + return nil +} + // Invoke invokes the Lambda function. This is our main trace point. func (f *Function) Invoke(req *messages.InvokeRequest, response *messages.InvokeResponse) error { tx := f.tracer.StartTransaction(lambdacontext.FunctionName, "function") @@ -92,6 +111,30 @@ func (f *Function) Invoke(req *messages.InvokeRequest, response *messages.Invoke lambdaContext.Request = formatPayload(req.Payload) lambdaContext.Response = "" + if !ignoreTxnRegistration { + defer jsonw.Reset() + if err := createPartialTransactionJSON(tx, &jsonw); err != nil { + log.Printf("failed to create partial transaction for registration: %v", err) + } else { + resp, err := http.Post( + // TODO: @lahsivjar better way to get base URI + "http://localhost:8200/register/transaction", + "application/vnd.elastic.apm.transaction+json", + bytes.NewReader(jsonw.Bytes()), + ) + // Don't attempt registration for next invocations if network + // error or the registration endpoint is not found. + if err != nil || resp.StatusCode == 404 { + ignoreTxnRegistration = true + } + if err != nil { + log.Printf("failed to register transaction, req failed with error: %v", err) + } + if resp.StatusCode/100 != 2 { + log.Printf("failed to register transaction, req failed with status code: %d", resp.StatusCode) + } + } + } err := f.client.Call("Function.Invoke", req, response) if err != nil { e := f.tracer.NewError(err) diff --git a/module/apmlambda/lambda_test.go b/module/apmlambda/lambda_test.go new file mode 100644 index 000000000..f25bf3158 --- /dev/null +++ b/module/apmlambda/lambda_test.go @@ -0,0 +1,36 @@ +// 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 apmlambda + +import ( + "encoding/json" + "testing" + + "github.com/stretchr/testify/assert" + "go.elastic.co/apm/v2" + "go.elastic.co/fastjson" +) + +func TestCreatePartialTransactionJSON(t *testing.T) { + var w fastjson.Writer + tx := apm.DefaultTracer().StartTransaction("test", "function") + defer tx.End() + assert.NoError(t, createPartialTransactionJSON(tx, &w)) + assert.True(t, json.Valid(w.Bytes())) + assert.Equal(t, "test", string(w.Bytes())) +} From 896ddd46962f77b7fc52c26008f3dbe04898d4f9 Mon Sep 17 00:00:00 2001 From: Vishal Raj Date: Wed, 23 Nov 2022 13:18:12 +0800 Subject: [PATCH 2/3] Add header requirement for req ID --- module/apmlambda/lambda.go | 62 ++++++++++++++++++++++++-------------- 1 file changed, 39 insertions(+), 23 deletions(-) diff --git a/module/apmlambda/lambda.go b/module/apmlambda/lambda.go index 7863d772c..d1e695ef0 100644 --- a/module/apmlambda/lambda.go +++ b/module/apmlambda/lambda.go @@ -19,6 +19,7 @@ package apmlambda // import "go.elastic.co/apm/module/apmlambda/v2" import ( "bytes" + "fmt" "log" "net" "net/http" @@ -111,29 +112,8 @@ func (f *Function) Invoke(req *messages.InvokeRequest, response *messages.Invoke lambdaContext.Request = formatPayload(req.Payload) lambdaContext.Response = "" - if !ignoreTxnRegistration { - defer jsonw.Reset() - if err := createPartialTransactionJSON(tx, &jsonw); err != nil { - log.Printf("failed to create partial transaction for registration: %v", err) - } else { - resp, err := http.Post( - // TODO: @lahsivjar better way to get base URI - "http://localhost:8200/register/transaction", - "application/vnd.elastic.apm.transaction+json", - bytes.NewReader(jsonw.Bytes()), - ) - // Don't attempt registration for next invocations if network - // error or the registration endpoint is not found. - if err != nil || resp.StatusCode == 404 { - ignoreTxnRegistration = true - } - if err != nil { - log.Printf("failed to register transaction, req failed with error: %v", err) - } - if resp.StatusCode/100 != 2 { - log.Printf("failed to register transaction, req failed with status code: %d", resp.StatusCode) - } - } + if err := registerTxn(tx, req.RequestId); err != nil { + log.Printf("failed to register txn: %v", err) } err := f.client.Call("Function.Invoke", req, response) if err != nil { @@ -154,6 +134,42 @@ func (f *Function) Invoke(req *messages.InvokeRequest, response *messages.Invoke return nil } +func registerTxn(tx *apm.Transaction, requestID string) error { + if ignoreTxnRegistration { + return nil + } + + defer jsonw.Reset() + if err := createPartialTransactionJSON(tx, &jsonw); err != nil { + return fmt.Errorf("failed to create txn registration body: %v", err) + } + req, err := http.NewRequest( + http.MethodPost, + // TODO: @lahsivjar better way to get base URI + "http://localhost:8200/register/transaction", + bytes.NewReader(jsonw.Bytes()), + ) + if err != nil { + return fmt.Errorf("failed to create txn registration request: %v", err) + } + req.Header.Set("Content-Type", "application/vnd.elastic.apm.transaction+json") + req.Header.Set("x-elastic-aws-request-id", requestID) + + resp, err := http.DefaultClient.Do(req) + // Don't attempt registration for next invocations if network + // error or the registration endpoint is not found. + if err != nil || resp.StatusCode == 404 { + ignoreTxnRegistration = true + } + if err != nil { + return fmt.Errorf("failed to register transaction, req failed with error: %v", err) + } + if resp.StatusCode/100 != 2 { + return fmt.Errorf("failed to register transaction, req failed with status code: %d", resp.StatusCode) + } + return nil +} + type invokeResponseError struct { err *messages.InvokeResponse_Error } From e93c15ceaaf52d9af012fc1339e2db22daf0171d Mon Sep 17 00:00:00 2001 From: Vishal Raj Date: Tue, 18 Apr 2023 11:41:37 +0800 Subject: [PATCH 3/3] Update as per new specs --- module/apmlambda/lambda.go | 19 +++++++++++++------ tracer.go | 6 +++--- 2 files changed, 16 insertions(+), 9 deletions(-) diff --git a/module/apmlambda/lambda.go b/module/apmlambda/lambda.go index d1e695ef0..0ee5b46c9 100644 --- a/module/apmlambda/lambda.go +++ b/module/apmlambda/lambda.go @@ -57,8 +57,8 @@ var ( Response string `json:"response,omitempty"` } - jsonw fastjson.Writer - + jsonw fastjson.Writer + metadataBytes int ignoreTxnRegistration bool ) @@ -112,7 +112,7 @@ func (f *Function) Invoke(req *messages.InvokeRequest, response *messages.Invoke lambdaContext.Request = formatPayload(req.Payload) lambdaContext.Response = "" - if err := registerTxn(tx, req.RequestId); err != nil { + if err := registerTxn(f.tracer, tx, req.RequestId); err != nil { log.Printf("failed to register txn: %v", err) } err := f.client.Call("Function.Invoke", req, response) @@ -134,12 +134,19 @@ func (f *Function) Invoke(req *messages.InvokeRequest, response *messages.Invoke return nil } -func registerTxn(tx *apm.Transaction, requestID string) error { +func registerTxn(tracer *apm.Tracer, tx *apm.Transaction, requestID string) error { if ignoreTxnRegistration { return nil } - defer jsonw.Reset() + if metadataBytes == 0 { + jsonw.Reset() + mb := tracer.JSONRequestMetadata() + jsonw.RawBytes(tracer.JSONRequestMetadata()) + metadataBytes = len(mb) + } + defer jsonw.Rewind(metadataBytes) + if err := createPartialTransactionJSON(tx, &jsonw); err != nil { return fmt.Errorf("failed to create txn registration body: %v", err) } @@ -152,7 +159,7 @@ func registerTxn(tx *apm.Transaction, requestID string) error { if err != nil { return fmt.Errorf("failed to create txn registration request: %v", err) } - req.Header.Set("Content-Type", "application/vnd.elastic.apm.transaction+json") + req.Header.Set("Content-Type", "application/vnd.elastic.apm.transaction+ndjson") req.Header.Set("x-elastic-aws-request-id", requestID) resp, err := http.DefaultClient.Do(req) diff --git a/tracer.go b/tracer.go index 1b6271559..1628d78b3 100644 --- a/tracer.go +++ b/tracer.go @@ -1241,7 +1241,7 @@ func (t *Tracer) loop() { } sendStreamRequest <- gracePeriod if metadata == nil { - metadata = t.jsonRequestMetadata() + metadata = t.JSONRequestMetadata() } zlibWriter.Reset(&requestBuf) zlibWriter.Write(metadata) @@ -1317,10 +1317,10 @@ func (t *Tracer) loop() { } } -// jsonRequestMetadata returns a JSON-encoded metadata object that features +// JSONRequestMetadata returns a JSON-encoded metadata object that features // at the head of every request body. This is called exactly once, when the // first request is made. -func (t *Tracer) jsonRequestMetadata() []byte { +func (t *Tracer) JSONRequestMetadata() []byte { var json fastjson.Writer json.RawString(`{"metadata":`) t.encodeRequestMetadata(&json)