internal_txn.go 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903
  1. package newrelic
  2. import (
  3. "errors"
  4. "fmt"
  5. "net/http"
  6. "net/url"
  7. "reflect"
  8. "strings"
  9. "sync"
  10. "time"
  11. "github.com/newrelic/go-agent/internal"
  12. )
  13. type txnInput struct {
  14. W http.ResponseWriter
  15. Config Config
  16. Reply *internal.ConnectReply
  17. Consumer dataConsumer
  18. attrConfig *internal.AttributeConfig
  19. }
  20. type txn struct {
  21. txnInput
  22. // This mutex is required since the consumer may call the public API
  23. // interface functions from different routines.
  24. sync.Mutex
  25. // finished indicates whether or not End() has been called. After
  26. // finished has been set to true, no recording should occur.
  27. finished bool
  28. numPayloadsCreated uint32
  29. ignore bool
  30. // wroteHeader prevents capturing multiple response code errors if the
  31. // user erroneously calls WriteHeader multiple times.
  32. wroteHeader bool
  33. internal.TxnData
  34. }
  35. func newTxn(input txnInput, name string) *txn {
  36. txn := &txn{
  37. txnInput: input,
  38. }
  39. txn.Start = time.Now()
  40. txn.Name = name
  41. txn.Attrs = internal.NewAttributes(input.attrConfig)
  42. if input.Config.DistributedTracer.Enabled {
  43. txn.BetterCAT.Enabled = true
  44. txn.BetterCAT.Priority = internal.NewPriority()
  45. txn.BetterCAT.ID = internal.NewSpanID()
  46. // Calculate sampled at the beginning of the transaction (rather
  47. // than lazily at payload creation time) because it controls the
  48. // creation of span events.
  49. txn.BetterCAT.Sampled = txn.Reply.AdaptiveSampler.ComputeSampled(txn.BetterCAT.Priority.Float32(), txn.Start)
  50. if txn.BetterCAT.Sampled {
  51. txn.BetterCAT.Priority += 1.0
  52. }
  53. txn.SpanEventsEnabled = txn.Config.SpanEvents.Enabled && txn.Reply.CollectSpanEvents
  54. }
  55. txn.Attrs.Agent.Add(internal.AttributeHostDisplayName, txn.Config.HostDisplayName, nil)
  56. txn.TxnTrace.Enabled = txn.txnTracesEnabled()
  57. txn.TxnTrace.SegmentThreshold = txn.Config.TransactionTracer.SegmentThreshold
  58. txn.StackTraceThreshold = txn.Config.TransactionTracer.StackTraceThreshold
  59. txn.SlowQueriesEnabled = txn.slowQueriesEnabled()
  60. txn.SlowQueryThreshold = txn.Config.DatastoreTracer.SlowQuery.Threshold
  61. // Synthetics support is tied up with a transaction's Old CAT field,
  62. // CrossProcess. To support Synthetics with either BetterCAT or Old CAT,
  63. // Initialize the CrossProcess field of the transaction, passing in
  64. // the top-level configuration.
  65. doOldCAT := txn.Config.CrossApplicationTracer.Enabled
  66. noGUID := txn.Config.DistributedTracer.Enabled
  67. txn.CrossProcess.Init(doOldCAT, noGUID, input.Reply)
  68. return txn
  69. }
  70. type requestWrap struct{ request *http.Request }
  71. func (r requestWrap) Header() http.Header { return r.request.Header }
  72. func (r requestWrap) URL() *url.URL { return r.request.URL }
  73. func (r requestWrap) Method() string { return r.request.Method }
  74. func (r requestWrap) Transport() TransportType {
  75. if strings.HasPrefix(r.request.Proto, "HTTP") {
  76. if r.request.TLS != nil {
  77. return TransportHTTPS
  78. }
  79. return TransportHTTP
  80. }
  81. return TransportUnknown
  82. }
  83. func (txn *txn) SetWebRequest(r WebRequest) error {
  84. txn.Lock()
  85. defer txn.Unlock()
  86. if txn.finished {
  87. return errAlreadyEnded
  88. }
  89. // Any call to SetWebRequest should indicate a web transaction.
  90. txn.IsWeb = true
  91. if nil == r {
  92. return nil
  93. }
  94. if h := r.Header(); nil != h {
  95. txn.Queuing = internal.QueueDuration(h, txn.Start)
  96. if p := h.Get(DistributedTracePayloadHeader); p != "" {
  97. txn.acceptDistributedTracePayloadLocked(r.Transport(), p)
  98. }
  99. txn.CrossProcess.InboundHTTPRequest(h)
  100. }
  101. internal.RequestAgentAttributes(txn.Attrs, r.Method(), r.Header())
  102. if u := r.URL(); nil != u {
  103. txn.CleanURL = internal.SafeURL(u)
  104. }
  105. return nil
  106. }
  107. func (txn *txn) slowQueriesEnabled() bool {
  108. return txn.Config.DatastoreTracer.SlowQuery.Enabled &&
  109. txn.Reply.CollectTraces
  110. }
  111. func (txn *txn) txnTracesEnabled() bool {
  112. return txn.Config.TransactionTracer.Enabled &&
  113. txn.Reply.CollectTraces
  114. }
  115. func (txn *txn) txnEventsEnabled() bool {
  116. return txn.Config.TransactionEvents.Enabled &&
  117. txn.Reply.CollectAnalyticsEvents
  118. }
  119. func (txn *txn) errorEventsEnabled() bool {
  120. return txn.Config.ErrorCollector.CaptureEvents &&
  121. txn.Reply.CollectErrorEvents
  122. }
  123. func (txn *txn) freezeName() {
  124. if txn.ignore || ("" != txn.FinalName) {
  125. return
  126. }
  127. txn.FinalName = internal.CreateFullTxnName(txn.Name, txn.Reply, txn.IsWeb)
  128. if "" == txn.FinalName {
  129. txn.ignore = true
  130. }
  131. }
  132. func (txn *txn) getsApdex() bool {
  133. return txn.IsWeb
  134. }
  135. func (txn *txn) txnTraceThreshold() time.Duration {
  136. if txn.Config.TransactionTracer.Threshold.IsApdexFailing {
  137. return internal.ApdexFailingThreshold(txn.ApdexThreshold)
  138. }
  139. return txn.Config.TransactionTracer.Threshold.Duration
  140. }
  141. func (txn *txn) shouldSaveTrace() bool {
  142. return txn.CrossProcess.IsSynthetics() ||
  143. (txn.txnTracesEnabled() && (txn.Duration >= txn.txnTraceThreshold()))
  144. }
  145. func (txn *txn) MergeIntoHarvest(h *internal.Harvest) {
  146. var priority internal.Priority
  147. if txn.BetterCAT.Enabled {
  148. priority = txn.BetterCAT.Priority
  149. } else {
  150. priority = internal.NewPriority()
  151. }
  152. internal.CreateTxnMetrics(&txn.TxnData, h.Metrics)
  153. internal.MergeBreakdownMetrics(&txn.TxnData, h.Metrics)
  154. if txn.txnEventsEnabled() {
  155. // Allocate a new TxnEvent to prevent a reference to the large transaction.
  156. alloc := new(internal.TxnEvent)
  157. *alloc = txn.TxnData.TxnEvent
  158. h.TxnEvents.AddTxnEvent(alloc, priority)
  159. }
  160. internal.MergeTxnErrors(&h.ErrorTraces, txn.Errors, txn.TxnEvent)
  161. if txn.errorEventsEnabled() {
  162. for _, e := range txn.Errors {
  163. errEvent := &internal.ErrorEvent{
  164. ErrorData: *e,
  165. TxnEvent: txn.TxnEvent,
  166. }
  167. // Since the stack trace is not used in error events, remove the reference
  168. // to minimize memory.
  169. errEvent.Stack = nil
  170. h.ErrorEvents.Add(errEvent, priority)
  171. }
  172. }
  173. if txn.shouldSaveTrace() {
  174. h.TxnTraces.Witness(internal.HarvestTrace{
  175. TxnEvent: txn.TxnEvent,
  176. Trace: txn.TxnTrace,
  177. })
  178. }
  179. if nil != txn.SlowQueries {
  180. h.SlowSQLs.Merge(txn.SlowQueries, txn.TxnEvent)
  181. }
  182. if txn.BetterCAT.Sampled && txn.SpanEventsEnabled {
  183. h.SpanEvents.MergeFromTransaction(&txn.TxnData)
  184. }
  185. }
  186. // TransportType's name field is not mutable outside of its package
  187. // however, it still periodically needs to be used and assigned within
  188. // the this package. For testing purposes only.
  189. func getTransport(transport string) string {
  190. var retVal string
  191. switch transport {
  192. case TransportHTTP.name:
  193. retVal = TransportHTTP.name
  194. case TransportHTTPS.name:
  195. retVal = TransportHTTPS.name
  196. case TransportKafka.name:
  197. retVal = TransportKafka.name
  198. case TransportJMS.name:
  199. retVal = TransportJMS.name
  200. case TransportIronMQ.name:
  201. retVal = TransportIronMQ.name
  202. case TransportAMQP.name:
  203. retVal = TransportAMQP.name
  204. case TransportQueue.name:
  205. retVal = TransportQueue.name
  206. case TransportOther.name:
  207. retVal = TransportOther.name
  208. case TransportUnknown.name:
  209. default:
  210. retVal = TransportUnknown.name
  211. }
  212. return retVal
  213. }
  214. func responseCodeIsError(cfg *Config, code int) bool {
  215. if code < http.StatusBadRequest { // 400
  216. return false
  217. }
  218. for _, ignoreCode := range cfg.ErrorCollector.IgnoreStatusCodes {
  219. if code == ignoreCode {
  220. return false
  221. }
  222. }
  223. return true
  224. }
  225. func headersJustWritten(txn *txn, code int) {
  226. txn.Lock()
  227. defer txn.Unlock()
  228. if txn.finished {
  229. return
  230. }
  231. if txn.wroteHeader {
  232. return
  233. }
  234. txn.wroteHeader = true
  235. internal.ResponseHeaderAttributes(txn.Attrs, txn.W.Header())
  236. internal.ResponseCodeAttribute(txn.Attrs, code)
  237. if responseCodeIsError(&txn.Config, code) {
  238. e := internal.TxnErrorFromResponseCode(time.Now(), code)
  239. e.Stack = internal.GetStackTrace(1)
  240. txn.noticeErrorInternal(e)
  241. }
  242. }
  243. func (txn *txn) responseHeader() http.Header {
  244. txn.Lock()
  245. defer txn.Unlock()
  246. if txn.finished {
  247. return nil
  248. }
  249. if txn.wroteHeader {
  250. return nil
  251. }
  252. if !txn.CrossProcess.Enabled {
  253. return nil
  254. }
  255. if !txn.CrossProcess.IsInbound() {
  256. return nil
  257. }
  258. txn.freezeName()
  259. contentLength := internal.GetContentLengthFromHeader(txn.W.Header())
  260. appData, err := txn.CrossProcess.CreateAppData(txn.FinalName, txn.Queuing, time.Since(txn.Start), contentLength)
  261. if err != nil {
  262. txn.Config.Logger.Debug("error generating outbound response header", map[string]interface{}{
  263. "error": err,
  264. })
  265. return nil
  266. }
  267. return internal.AppDataToHTTPHeader(appData)
  268. }
  269. func addCrossProcessHeaders(txn *txn) {
  270. // responseHeader() checks the wroteHeader field and returns a nil map if the
  271. // header has been written, so we don't need a check here.
  272. for key, values := range txn.responseHeader() {
  273. for _, value := range values {
  274. txn.W.Header().Add(key, value)
  275. }
  276. }
  277. }
  278. func (txn *txn) Header() http.Header { return txn.W.Header() }
  279. func (txn *txn) Write(b []byte) (int, error) {
  280. // This is safe to call unconditionally, even if Write() is called multiple
  281. // times; see also the commentary in addCrossProcessHeaders().
  282. addCrossProcessHeaders(txn)
  283. n, err := txn.W.Write(b)
  284. headersJustWritten(txn, http.StatusOK)
  285. return n, err
  286. }
  287. func (txn *txn) WriteHeader(code int) {
  288. addCrossProcessHeaders(txn)
  289. txn.W.WriteHeader(code)
  290. headersJustWritten(txn, code)
  291. }
  292. func (txn *txn) End() error {
  293. txn.Lock()
  294. defer txn.Unlock()
  295. if txn.finished {
  296. return errAlreadyEnded
  297. }
  298. txn.finished = true
  299. r := recover()
  300. if nil != r {
  301. e := internal.TxnErrorFromPanic(time.Now(), r)
  302. e.Stack = internal.GetStackTrace(0)
  303. txn.noticeErrorInternal(e)
  304. }
  305. txn.Stop = time.Now()
  306. txn.Duration = txn.Stop.Sub(txn.Start)
  307. if children := internal.TracerRootChildren(&txn.TxnData); txn.Duration > children {
  308. txn.Exclusive = txn.Duration - children
  309. }
  310. txn.freezeName()
  311. // Finalise the CAT state.
  312. if err := txn.CrossProcess.Finalise(txn.Name, txn.Config.AppName); err != nil {
  313. txn.Config.Logger.Debug("error finalising the cross process state", map[string]interface{}{
  314. "error": err,
  315. })
  316. }
  317. // Assign apdexThreshold regardless of whether or not the transaction
  318. // gets apdex since it may be used to calculate the trace threshold.
  319. txn.ApdexThreshold = internal.CalculateApdexThreshold(txn.Reply, txn.FinalName)
  320. if txn.getsApdex() {
  321. if txn.HasErrors() {
  322. txn.Zone = internal.ApdexFailing
  323. } else {
  324. txn.Zone = internal.CalculateApdexZone(txn.ApdexThreshold, txn.Duration)
  325. }
  326. } else {
  327. txn.Zone = internal.ApdexNone
  328. }
  329. if txn.Config.Logger.DebugEnabled() {
  330. txn.Config.Logger.Debug("transaction ended", map[string]interface{}{
  331. "name": txn.FinalName,
  332. "duration_ms": txn.Duration.Seconds() * 1000.0,
  333. "ignored": txn.ignore,
  334. "app_connected": "" != txn.Reply.RunID,
  335. })
  336. }
  337. if !txn.ignore {
  338. txn.Consumer.Consume(txn.Reply.RunID, txn)
  339. }
  340. // Note that if a consumer uses `panic(nil)`, the panic will not
  341. // propagate.
  342. if nil != r {
  343. panic(r)
  344. }
  345. return nil
  346. }
  347. func (txn *txn) AddAttribute(name string, value interface{}) error {
  348. txn.Lock()
  349. defer txn.Unlock()
  350. if txn.Config.HighSecurity {
  351. return errHighSecurityEnabled
  352. }
  353. if !txn.Reply.SecurityPolicies.CustomParameters.Enabled() {
  354. return errSecurityPolicy
  355. }
  356. if txn.finished {
  357. return errAlreadyEnded
  358. }
  359. return internal.AddUserAttribute(txn.Attrs, name, value, internal.DestAll)
  360. }
  361. var (
  362. errorsLocallyDisabled = errors.New("errors locally disabled")
  363. errorsRemotelyDisabled = errors.New("errors remotely disabled")
  364. errNilError = errors.New("nil error")
  365. errAlreadyEnded = errors.New("transaction has already ended")
  366. errSecurityPolicy = errors.New("disabled by security policy")
  367. )
  368. const (
  369. highSecurityErrorMsg = "message removed by high security setting"
  370. securityPolicyErrorMsg = "message removed by security policy"
  371. )
  372. func (txn *txn) noticeErrorInternal(err internal.ErrorData) error {
  373. if !txn.Config.ErrorCollector.Enabled {
  374. return errorsLocallyDisabled
  375. }
  376. if !txn.Reply.CollectErrors {
  377. return errorsRemotelyDisabled
  378. }
  379. if nil == txn.Errors {
  380. txn.Errors = internal.NewTxnErrors(internal.MaxTxnErrors)
  381. }
  382. if txn.Config.HighSecurity {
  383. err.Msg = highSecurityErrorMsg
  384. }
  385. if !txn.Reply.SecurityPolicies.AllowRawExceptionMessages.Enabled() {
  386. err.Msg = securityPolicyErrorMsg
  387. }
  388. txn.Errors.Add(err)
  389. txn.TxnData.TxnEvent.HasError = true //mark transaction as having an error
  390. return nil
  391. }
  392. var (
  393. errTooManyErrorAttributes = fmt.Errorf("too many extra attributes: limit is %d",
  394. internal.AttributeErrorLimit)
  395. )
  396. func (txn *txn) NoticeError(err error) error {
  397. txn.Lock()
  398. defer txn.Unlock()
  399. if txn.finished {
  400. return errAlreadyEnded
  401. }
  402. if nil == err {
  403. return errNilError
  404. }
  405. e := internal.ErrorData{
  406. When: time.Now(),
  407. Msg: err.Error(),
  408. }
  409. if ec, ok := err.(ErrorClasser); ok {
  410. e.Klass = ec.ErrorClass()
  411. }
  412. if "" == e.Klass {
  413. e.Klass = reflect.TypeOf(err).String()
  414. }
  415. if st, ok := err.(StackTracer); ok {
  416. e.Stack = st.StackTrace()
  417. // Note that if the provided stack trace is excessive in length,
  418. // it will be truncated during JSON creation.
  419. }
  420. if nil == e.Stack {
  421. e.Stack = internal.GetStackTrace(2)
  422. }
  423. if ea, ok := err.(ErrorAttributer); ok && !txn.Config.HighSecurity && txn.Reply.SecurityPolicies.CustomParameters.Enabled() {
  424. unvetted := ea.ErrorAttributes()
  425. if len(unvetted) > internal.AttributeErrorLimit {
  426. return errTooManyErrorAttributes
  427. }
  428. e.ExtraAttributes = make(map[string]interface{})
  429. for key, val := range unvetted {
  430. val, errr := internal.ValidateUserAttribute(key, val)
  431. if nil != errr {
  432. return errr
  433. }
  434. e.ExtraAttributes[key] = val
  435. }
  436. }
  437. return txn.noticeErrorInternal(e)
  438. }
  439. func (txn *txn) SetName(name string) error {
  440. txn.Lock()
  441. defer txn.Unlock()
  442. if txn.finished {
  443. return errAlreadyEnded
  444. }
  445. txn.Name = name
  446. return nil
  447. }
  448. func (txn *txn) Ignore() error {
  449. txn.Lock()
  450. defer txn.Unlock()
  451. if txn.finished {
  452. return errAlreadyEnded
  453. }
  454. txn.ignore = true
  455. return nil
  456. }
  457. func (txn *txn) StartSegmentNow() SegmentStartTime {
  458. var s internal.SegmentStartTime
  459. txn.Lock()
  460. if !txn.finished {
  461. s = internal.StartSegment(&txn.TxnData, time.Now())
  462. }
  463. txn.Unlock()
  464. return SegmentStartTime{
  465. segment: segment{
  466. start: s,
  467. txn: txn,
  468. },
  469. }
  470. }
  471. type segment struct {
  472. start internal.SegmentStartTime
  473. txn *txn
  474. }
  475. func endSegment(s *Segment) error {
  476. txn := s.StartTime.txn
  477. if nil == txn {
  478. return nil
  479. }
  480. var err error
  481. txn.Lock()
  482. if txn.finished {
  483. err = errAlreadyEnded
  484. } else {
  485. err = internal.EndBasicSegment(&txn.TxnData, s.StartTime.start, time.Now(), s.Name)
  486. }
  487. txn.Unlock()
  488. return err
  489. }
  490. func endDatastore(s *DatastoreSegment) error {
  491. txn := s.StartTime.txn
  492. if nil == txn {
  493. return nil
  494. }
  495. txn.Lock()
  496. defer txn.Unlock()
  497. if txn.finished {
  498. return errAlreadyEnded
  499. }
  500. if txn.Config.HighSecurity {
  501. s.QueryParameters = nil
  502. }
  503. if !txn.Config.DatastoreTracer.QueryParameters.Enabled {
  504. s.QueryParameters = nil
  505. }
  506. if txn.Reply.SecurityPolicies.RecordSQL.IsSet() {
  507. s.QueryParameters = nil
  508. if !txn.Reply.SecurityPolicies.RecordSQL.Enabled() {
  509. s.ParameterizedQuery = ""
  510. }
  511. }
  512. if !txn.Config.DatastoreTracer.DatabaseNameReporting.Enabled {
  513. s.DatabaseName = ""
  514. }
  515. if !txn.Config.DatastoreTracer.InstanceReporting.Enabled {
  516. s.Host = ""
  517. s.PortPathOrID = ""
  518. }
  519. return internal.EndDatastoreSegment(internal.EndDatastoreParams{
  520. Tracer: &txn.TxnData,
  521. Start: s.StartTime.start,
  522. Now: time.Now(),
  523. Product: string(s.Product),
  524. Collection: s.Collection,
  525. Operation: s.Operation,
  526. ParameterizedQuery: s.ParameterizedQuery,
  527. QueryParameters: s.QueryParameters,
  528. Host: s.Host,
  529. PortPathOrID: s.PortPathOrID,
  530. Database: s.DatabaseName,
  531. })
  532. }
  533. func externalSegmentMethod(s *ExternalSegment) string {
  534. r := s.Request
  535. // Is this a client request?
  536. if nil != s.Response && nil != s.Response.Request {
  537. r = s.Response.Request
  538. // Golang's http package states that when a client's
  539. // Request has an empty string for Method, the
  540. // method is GET.
  541. if "" == r.Method {
  542. return "GET"
  543. }
  544. }
  545. if nil == r {
  546. return ""
  547. }
  548. return r.Method
  549. }
  550. func externalSegmentURL(s *ExternalSegment) (*url.URL, error) {
  551. if "" != s.URL {
  552. return url.Parse(s.URL)
  553. }
  554. r := s.Request
  555. if nil != s.Response && nil != s.Response.Request {
  556. r = s.Response.Request
  557. }
  558. if r != nil {
  559. return r.URL, nil
  560. }
  561. return nil, nil
  562. }
  563. func endExternal(s *ExternalSegment) error {
  564. txn := s.StartTime.txn
  565. if nil == txn {
  566. return nil
  567. }
  568. txn.Lock()
  569. defer txn.Unlock()
  570. if txn.finished {
  571. return errAlreadyEnded
  572. }
  573. m := externalSegmentMethod(s)
  574. u, err := externalSegmentURL(s)
  575. if nil != err {
  576. return err
  577. }
  578. return internal.EndExternalSegment(&txn.TxnData, s.StartTime.start, time.Now(), u, m, s.Response)
  579. }
  580. // oldCATOutboundHeaders generates the Old CAT and Synthetics headers, depending
  581. // on whether Old CAT is enabled or any Synthetics functionality has been
  582. // triggered in the agent.
  583. func oldCATOutboundHeaders(txn *txn) http.Header {
  584. txn.Lock()
  585. defer txn.Unlock()
  586. if txn.finished {
  587. return http.Header{}
  588. }
  589. metadata, err := txn.CrossProcess.CreateCrossProcessMetadata(txn.Name, txn.Config.AppName)
  590. if err != nil {
  591. txn.Config.Logger.Debug("error generating outbound headers", map[string]interface{}{
  592. "error": err,
  593. })
  594. // It's possible for CreateCrossProcessMetadata() to error and still have a
  595. // Synthetics header, so we'll still fall through to returning headers
  596. // based on whatever metadata was returned.
  597. }
  598. return internal.MetadataToHTTPHeader(metadata)
  599. }
  600. func outboundHeaders(s *ExternalSegment) http.Header {
  601. txn := s.StartTime.txn
  602. if nil == txn {
  603. return http.Header{}
  604. }
  605. hdr := oldCATOutboundHeaders(txn)
  606. // hdr may be empty, or it may contain headers. If DistributedTracer
  607. // is enabled, add more to the existing hdr
  608. if p := txn.CreateDistributedTracePayload().HTTPSafe(); "" != p {
  609. hdr.Add(DistributedTracePayloadHeader, p)
  610. return hdr
  611. }
  612. return hdr
  613. }
  614. const (
  615. maxSampledDistributedPayloads = 35
  616. )
  617. type shimPayload struct{}
  618. func (s shimPayload) Text() string { return "" }
  619. func (s shimPayload) HTTPSafe() string { return "" }
  620. func (txn *txn) CreateDistributedTracePayload() (payload DistributedTracePayload) {
  621. payload = shimPayload{}
  622. txn.Lock()
  623. defer txn.Unlock()
  624. if !txn.BetterCAT.Enabled {
  625. return
  626. }
  627. if txn.finished {
  628. txn.CreatePayloadException = true
  629. return
  630. }
  631. if "" == txn.Reply.PrimaryAppID {
  632. // Return a shimPayload if the application is not yet connected.
  633. return
  634. }
  635. txn.numPayloadsCreated++
  636. var p internal.Payload
  637. p.Type = internal.CallerType
  638. p.Account = txn.Reply.AccountID
  639. p.App = txn.Reply.PrimaryAppID
  640. p.TracedID = txn.BetterCAT.TraceID()
  641. p.Priority = txn.BetterCAT.Priority
  642. p.Timestamp.Set(time.Now())
  643. p.TransactionID = txn.BetterCAT.ID // Set the transaction ID to the transaction guid.
  644. if txn.Reply.AccountID != txn.Reply.TrustedAccountKey {
  645. p.TrustedAccountKey = txn.Reply.TrustedAccountKey
  646. }
  647. if txn.BetterCAT.Sampled && txn.SpanEventsEnabled {
  648. p.ID = txn.CurrentSpanIdentifier()
  649. }
  650. // limit the number of outbound sampled=true payloads to prevent too
  651. // many downstream sampled events.
  652. p.SetSampled(false)
  653. if txn.numPayloadsCreated < maxSampledDistributedPayloads {
  654. p.SetSampled(txn.BetterCAT.Sampled)
  655. }
  656. txn.CreatePayloadSuccess = true
  657. payload = p
  658. return
  659. }
  660. var (
  661. errOutboundPayloadCreated = errors.New("outbound payload already created")
  662. errAlreadyAccepted = errors.New("AcceptDistributedTracePayload has already been called")
  663. errInboundPayloadDTDisabled = errors.New("DistributedTracer must be enabled to accept an inbound payload")
  664. errTrustedAccountKey = errors.New("trusted account key missing or does not match")
  665. )
  666. func (txn *txn) AcceptDistributedTracePayload(t TransportType, p interface{}) error {
  667. txn.Lock()
  668. defer txn.Unlock()
  669. return txn.acceptDistributedTracePayloadLocked(t, p)
  670. }
  671. func (txn *txn) acceptDistributedTracePayloadLocked(t TransportType, p interface{}) error {
  672. if !txn.BetterCAT.Enabled {
  673. return errInboundPayloadDTDisabled
  674. }
  675. if txn.finished {
  676. txn.AcceptPayloadException = true
  677. return errAlreadyEnded
  678. }
  679. if txn.numPayloadsCreated > 0 {
  680. txn.AcceptPayloadCreateBeforeAccept = true
  681. return errOutboundPayloadCreated
  682. }
  683. if txn.BetterCAT.Inbound != nil {
  684. txn.AcceptPayloadIgnoredMultiple = true
  685. return errAlreadyAccepted
  686. }
  687. if nil == p {
  688. txn.AcceptPayloadNullPayload = true
  689. return nil
  690. }
  691. payload, err := internal.AcceptPayload(p)
  692. if nil != err {
  693. if _, ok := err.(internal.ErrPayloadParse); ok {
  694. txn.AcceptPayloadParseException = true
  695. } else if _, ok := err.(internal.ErrUnsupportedPayloadVersion); ok {
  696. txn.AcceptPayloadIgnoredVersion = true
  697. } else if _, ok := err.(internal.ErrPayloadMissingField); ok {
  698. txn.AcceptPayloadParseException = true
  699. } else {
  700. txn.AcceptPayloadException = true
  701. }
  702. return err
  703. }
  704. if nil == payload {
  705. return nil
  706. }
  707. // now that we have a parsed and alloc'd payload,
  708. // let's make sure it has the correct fields
  709. if err := payload.IsValid(); nil != err {
  710. txn.AcceptPayloadParseException = true
  711. return err
  712. }
  713. // and let's also do our trustedKey check
  714. receivedTrustKey := payload.TrustedAccountKey
  715. if "" == receivedTrustKey {
  716. receivedTrustKey = payload.Account
  717. }
  718. if receivedTrustKey != txn.Reply.TrustedAccountKey {
  719. txn.AcceptPayloadUntrustedAccount = true
  720. return errTrustedAccountKey
  721. }
  722. if 0 != payload.Priority {
  723. txn.BetterCAT.Priority = payload.Priority
  724. }
  725. // a nul payload.Sampled means the a field wasn't provided
  726. if nil != payload.Sampled {
  727. txn.BetterCAT.Sampled = *payload.Sampled
  728. } else {
  729. txn.BetterCAT.Sampled = txn.Reply.AdaptiveSampler.ComputeSampled(txn.BetterCAT.Priority.Float32(), time.Now())
  730. }
  731. txn.BetterCAT.Inbound = payload
  732. // TransportType's name field is not mutable outside of its package
  733. // so the only check needed is if the caller is using an empty TransportType
  734. txn.BetterCAT.Inbound.TransportType = t.name
  735. if t.name == "" {
  736. txn.BetterCAT.Inbound.TransportType = TransportUnknown.name
  737. txn.Config.Logger.Debug("Invalid transport type, defaulting to Unknown", map[string]interface{}{})
  738. }
  739. if tm := payload.Timestamp.Time(); txn.Start.After(tm) {
  740. txn.BetterCAT.Inbound.TransportDuration = txn.Start.Sub(tm)
  741. }
  742. txn.AcceptPayloadSuccess = true
  743. return nil
  744. }