mirror of
https://github.com/hanzomsg/nats-server.git
synced 2026-10-30 20:27:34 +00:00
An issue was reported (https://github.com/nats-io/nats-architecture-and-design/issues/392) stating that users don't want the `traceparent` header to be changed to `Xraceparent` by the server (in order to disable tracing) because it may break their applications that rely on its presence. The header will no longer be changed, including its case. The specification (https://www.w3.org/TR/trace-context/#header-name) says: ``` Vendors MUST expect the header in any case (upper, lower, mixed), and SHOULD send the header name in lowercase. ``` Since it was "SHOULD", we have no obligation to rewrite it in lower case. In the ADR issue, it was discussed to add a new header that would disable tracing (and will leave `traceparent` header name untouched) but the issue with that is that in a mixed environment (newer and older servers), the older servers would not know about the new header and would possibly emit traces when they should not. The approach in this PR is to add or update the content of `Nats-Trace-Dest` header to a specific invalid subject ("trace disabled") instead of changing the `Nats-Trace-Dest` and/or `traceparent` headers to `Xats-Trace-Dest` and `Xraceparent`. For newer server, when processing an inbound message that has the `Nats-Trace-Dest` header set to this specific value, regardless of the presence of `traceparent` or not, we understand it as a way to disable tracing, do not report any error and consider the tracing disabled (and process the message normally). For older servers, the presence of `Nats-Trace-Dest` takes precedence, so `traceparent` will be ignored, but since the trace destination is an invalid NATS subject, the error: `Destination "trace disabled" is not valid, won't be able to trace events` will be logged, and the message will not be traced. It is not ideal that an error message is printed, but this allows for backward compatibility. The message is still processed normally, so in this context the error can be safely ignored by the user. Signed-off-by: Ivan Kozlovic <[email protected]>
800 lines
22 KiB
Go
800 lines
22 KiB
Go
// Copyright 2024-2026 The NATS Authors
|
|
// Licensed 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 server
|
|
|
|
import (
|
|
"bytes"
|
|
"encoding/json"
|
|
"fmt"
|
|
"math/rand"
|
|
"strconv"
|
|
"strings"
|
|
"sync/atomic"
|
|
"time"
|
|
)
|
|
|
|
const (
|
|
MsgTraceDest = "Nats-Trace-Dest"
|
|
MsgTraceDestDisabled = "trace disabled" // This must be an invalid NATS subject
|
|
MsgTraceHop = "Nats-Trace-Hop"
|
|
MsgTraceOriginAccount = "Nats-Trace-Origin-Account"
|
|
MsgTraceOnly = "Nats-Trace-Only"
|
|
|
|
// External trace header. Note that this header is normally in lower
|
|
// case (https://www.w3.org/TR/trace-context/#header-name). Vendors
|
|
// MUST expect the header in any case (upper, lower, mixed), and
|
|
// SHOULD send the header name in lowercase. We used to change it
|
|
// to lower case, but no longer do that in 2.14.
|
|
traceParentHdr = "traceparent"
|
|
)
|
|
|
|
var (
|
|
traceDestHdrAsBytes = stringToBytes(MsgTraceDest)
|
|
traceDestDisabledAsBytes = stringToBytes(MsgTraceDestDisabled)
|
|
traceParentHdrAsBytes = stringToBytes(traceParentHdr)
|
|
crLFAsBytes = stringToBytes(CR_LF)
|
|
dashAsBytes = stringToBytes("-")
|
|
)
|
|
|
|
type MsgTraceType string
|
|
|
|
// Type of message trace events in the MsgTraceEvents list.
|
|
// This is needed to unmarshal the list.
|
|
const (
|
|
MsgTraceIngressType = "in"
|
|
MsgTraceSubjectMappingType = "sm"
|
|
MsgTraceStreamExportType = "se"
|
|
MsgTraceServiceImportType = "si"
|
|
MsgTraceJetStreamType = "js"
|
|
MsgTraceEgressType = "eg"
|
|
)
|
|
|
|
type MsgTraceEvent struct {
|
|
Server ServerInfo `json:"server"`
|
|
Request MsgTraceRequest `json:"request"`
|
|
Hops int `json:"hops,omitempty"`
|
|
Events MsgTraceEvents `json:"events"`
|
|
}
|
|
|
|
type MsgTraceRequest struct {
|
|
// We are not making this an http.Header so that header name case is preserved.
|
|
Header map[string][]string `json:"header,omitempty"`
|
|
MsgSize int `json:"msgsize,omitempty"`
|
|
}
|
|
|
|
type MsgTraceEvents []MsgTrace
|
|
|
|
type MsgTrace interface {
|
|
new() MsgTrace
|
|
typ() MsgTraceType
|
|
}
|
|
|
|
type MsgTraceBase struct {
|
|
Type MsgTraceType `json:"type"`
|
|
Timestamp time.Time `json:"ts"`
|
|
}
|
|
|
|
type MsgTraceIngress struct {
|
|
MsgTraceBase
|
|
Kind int `json:"kind"`
|
|
CID uint64 `json:"cid"`
|
|
Name string `json:"name,omitempty"`
|
|
Account string `json:"acc"`
|
|
Subject string `json:"subj"`
|
|
Error string `json:"error,omitempty"`
|
|
}
|
|
|
|
type MsgTraceSubjectMapping struct {
|
|
MsgTraceBase
|
|
MappedTo string `json:"to"`
|
|
}
|
|
|
|
type MsgTraceStreamExport struct {
|
|
MsgTraceBase
|
|
Account string `json:"acc"`
|
|
To string `json:"to"`
|
|
}
|
|
|
|
type MsgTraceServiceImport struct {
|
|
MsgTraceBase
|
|
Account string `json:"acc"`
|
|
From string `json:"from"`
|
|
To string `json:"to"`
|
|
}
|
|
|
|
type MsgTraceJetStream struct {
|
|
MsgTraceBase
|
|
Stream string `json:"stream"`
|
|
Subject string `json:"subject,omitempty"`
|
|
NoInterest bool `json:"nointerest,omitempty"`
|
|
Error string `json:"error,omitempty"`
|
|
}
|
|
|
|
type MsgTraceEgress struct {
|
|
MsgTraceBase
|
|
Kind int `json:"kind"`
|
|
CID uint64 `json:"cid"`
|
|
Name string `json:"name,omitempty"`
|
|
Hop string `json:"hop,omitempty"`
|
|
Account string `json:"acc,omitempty"`
|
|
Subscription string `json:"sub,omitempty"`
|
|
Queue string `json:"queue,omitempty"`
|
|
Error string `json:"error,omitempty"`
|
|
|
|
// This is for applications that unmarshal the trace events
|
|
// and want to link an egress to route/leaf/gateway with
|
|
// the MsgTraceEvent from that server.
|
|
Link *MsgTraceEvent `json:"-"`
|
|
}
|
|
|
|
// -------------------------------------------------------------
|
|
|
|
func (t MsgTraceBase) typ() MsgTraceType { return t.Type }
|
|
func (MsgTraceIngress) new() MsgTrace { return &MsgTraceIngress{} }
|
|
func (MsgTraceSubjectMapping) new() MsgTrace { return &MsgTraceSubjectMapping{} }
|
|
func (MsgTraceStreamExport) new() MsgTrace { return &MsgTraceStreamExport{} }
|
|
func (MsgTraceServiceImport) new() MsgTrace { return &MsgTraceServiceImport{} }
|
|
func (MsgTraceJetStream) new() MsgTrace { return &MsgTraceJetStream{} }
|
|
func (MsgTraceEgress) new() MsgTrace { return &MsgTraceEgress{} }
|
|
|
|
var msgTraceInterfaces = map[MsgTraceType]MsgTrace{
|
|
MsgTraceIngressType: MsgTraceIngress{},
|
|
MsgTraceSubjectMappingType: MsgTraceSubjectMapping{},
|
|
MsgTraceStreamExportType: MsgTraceStreamExport{},
|
|
MsgTraceServiceImportType: MsgTraceServiceImport{},
|
|
MsgTraceJetStreamType: MsgTraceJetStream{},
|
|
MsgTraceEgressType: MsgTraceEgress{},
|
|
}
|
|
|
|
func (t *MsgTraceEvents) UnmarshalJSON(data []byte) error {
|
|
var raw []json.RawMessage
|
|
err := json.Unmarshal(data, &raw)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
*t = make(MsgTraceEvents, len(raw))
|
|
var tt MsgTraceBase
|
|
for i, r := range raw {
|
|
if err = json.Unmarshal(r, &tt); err != nil {
|
|
return err
|
|
}
|
|
tr, ok := msgTraceInterfaces[tt.Type]
|
|
if !ok {
|
|
return fmt.Errorf("unknown trace type %v", tt.Type)
|
|
}
|
|
te := tr.new()
|
|
if err := json.Unmarshal(r, te); err != nil {
|
|
return err
|
|
}
|
|
(*t)[i] = te
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func getTraceAs[T MsgTrace](e any) *T {
|
|
v, ok := e.(*T)
|
|
if ok {
|
|
return v
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (t *MsgTraceEvent) Ingress() *MsgTraceIngress {
|
|
if len(t.Events) < 1 {
|
|
return nil
|
|
}
|
|
return getTraceAs[MsgTraceIngress](t.Events[0])
|
|
}
|
|
|
|
func (t *MsgTraceEvent) SubjectMapping() *MsgTraceSubjectMapping {
|
|
for _, e := range t.Events {
|
|
if e.typ() == MsgTraceSubjectMappingType {
|
|
return getTraceAs[MsgTraceSubjectMapping](e)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (t *MsgTraceEvent) StreamExports() []*MsgTraceStreamExport {
|
|
var se []*MsgTraceStreamExport
|
|
for _, e := range t.Events {
|
|
if e.typ() == MsgTraceStreamExportType {
|
|
se = append(se, getTraceAs[MsgTraceStreamExport](e))
|
|
}
|
|
}
|
|
return se
|
|
}
|
|
|
|
func (t *MsgTraceEvent) ServiceImports() []*MsgTraceServiceImport {
|
|
var si []*MsgTraceServiceImport
|
|
for _, e := range t.Events {
|
|
if e.typ() == MsgTraceServiceImportType {
|
|
si = append(si, getTraceAs[MsgTraceServiceImport](e))
|
|
}
|
|
}
|
|
return si
|
|
}
|
|
|
|
func (t *MsgTraceEvent) JetStream() *MsgTraceJetStream {
|
|
for _, e := range t.Events {
|
|
if e.typ() == MsgTraceJetStreamType {
|
|
return getTraceAs[MsgTraceJetStream](e)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (t *MsgTraceEvent) Egresses() []*MsgTraceEgress {
|
|
var eg []*MsgTraceEgress
|
|
for _, e := range t.Events {
|
|
if e.typ() == MsgTraceEgressType {
|
|
eg = append(eg, getTraceAs[MsgTraceEgress](e))
|
|
}
|
|
}
|
|
return eg
|
|
}
|
|
|
|
const (
|
|
errMsgTraceOnlyNoSupport = "Not delivered because remote does not support message tracing"
|
|
errMsgTraceNoSupport = "Message delivered but remote does not support message tracing so no trace event generated from there"
|
|
errMsgTraceNoEcho = "Not delivered because of no echo"
|
|
errMsgTracePubViolation = "Not delivered because publish denied for this subject"
|
|
errMsgTraceSubDeny = "Not delivered because subscription denies this subject"
|
|
errMsgTraceSubClosed = "Not delivered because subscription is closed"
|
|
errMsgTraceClientClosed = "Not delivered because client is closed"
|
|
errMsgTraceAutoSubExceeded = "Not delivered because auto-unsubscribe exceeded"
|
|
errMsgTraceFastProdNoStall = "Not delivered because fast producer not stalled and consumer is slow"
|
|
)
|
|
|
|
type msgTrace struct {
|
|
ready int32
|
|
srv *Server
|
|
acc *Account
|
|
// Origin account name, set only if acc is nil when acc lookup failed.
|
|
oan string
|
|
dest string
|
|
event *MsgTraceEvent
|
|
js *MsgTraceJetStream
|
|
hop string
|
|
nhop string
|
|
tonly bool // Will only trace the message, not do delivery.
|
|
ct compressionType
|
|
}
|
|
|
|
// This will be false outside of the tests, so when building the server binary,
|
|
// any code where you see `if msgTraceRunInTests` statement will be compiled
|
|
// out, so this will have no performance penalty.
|
|
var (
|
|
msgTraceRunInTests bool
|
|
msgTraceCheckSupport bool
|
|
)
|
|
|
|
// Returns the message trace object, if message is being traced,
|
|
// and `true` if we want to only trace, not actually deliver the message.
|
|
func (c *client) isMsgTraceEnabled() (*msgTrace, bool) {
|
|
t := c.pa.trace
|
|
if t == nil {
|
|
return nil, false
|
|
}
|
|
return t, t.tonly
|
|
}
|
|
|
|
// For LEAF/ROUTER/GATEWAY, return false if the remote does not support
|
|
// message tracing (important if the tracing requests trace-only).
|
|
func (c *client) msgTraceSupport() bool {
|
|
// Exclude client connection from the protocol check.
|
|
return c.kind == CLIENT || c.opts.Protocol >= MsgTraceProto
|
|
}
|
|
|
|
func getConnName(c *client) string {
|
|
switch c.kind {
|
|
case ROUTER:
|
|
if n := c.route.remoteName; n != _EMPTY_ {
|
|
return n
|
|
}
|
|
case GATEWAY:
|
|
if n := c.gw.remoteName; n != _EMPTY_ {
|
|
return n
|
|
}
|
|
case LEAF:
|
|
if n := c.leaf.remoteServer; n != _EMPTY_ {
|
|
return n
|
|
}
|
|
}
|
|
return c.opts.Name
|
|
}
|
|
|
|
func getCompressionType(cts string) compressionType {
|
|
if cts == _EMPTY_ {
|
|
return noCompression
|
|
}
|
|
cts = strings.ToLower(cts)
|
|
if strings.Contains(cts, "snappy") || strings.Contains(cts, "s2") {
|
|
return snappyCompression
|
|
}
|
|
if strings.Contains(cts, "gzip") {
|
|
return gzipCompression
|
|
}
|
|
return unsupportedCompression
|
|
}
|
|
|
|
func (c *client) initMsgTrace() *msgTrace {
|
|
// The code in the "if" statement is only running in test mode.
|
|
if msgTraceRunInTests {
|
|
// Check the type of client that tries to initialize a trace struct.
|
|
if !(c.kind == CLIENT || c.kind == ROUTER || c.kind == GATEWAY || c.kind == LEAF) {
|
|
panic(fmt.Sprintf("Unexpected client type %q trying to initialize msgTrace", c.kindString()))
|
|
}
|
|
// In some tests, we want to make a server behave like an old server
|
|
// and so even if a trace header is received, we want the server to
|
|
// simply ignore it.
|
|
if msgTraceCheckSupport {
|
|
if c.srv == nil || c.srv.getServerProto() < MsgTraceProto {
|
|
return nil
|
|
}
|
|
}
|
|
}
|
|
if c.pa.hdr <= 0 {
|
|
return nil
|
|
}
|
|
hdr := c.msgBuf[:c.pa.hdr]
|
|
headers, external := genHeaderMapIfTraceHeadersPresent(hdr)
|
|
if len(headers) == 0 {
|
|
return nil
|
|
}
|
|
// Little helper to give us the first value of a given header, or _EMPTY_
|
|
// if key is not present.
|
|
getHdrVal := func(key string) string {
|
|
vv, ok := headers[key]
|
|
if !ok {
|
|
return _EMPTY_
|
|
}
|
|
return vv[0]
|
|
}
|
|
var (
|
|
dest string
|
|
traceOnly bool
|
|
)
|
|
// Check for traceOnly only if not external.
|
|
if !external {
|
|
if to := getHdrVal(MsgTraceOnly); to != _EMPTY_ {
|
|
tos := strings.ToLower(to)
|
|
switch tos {
|
|
case "1", "true", "on":
|
|
traceOnly = true
|
|
}
|
|
}
|
|
dest = getHdrVal(MsgTraceDest)
|
|
// Check the destination to see if this is a valid public subject.
|
|
if !IsValidPublishSubject(dest) {
|
|
// We still have to return a msgTrace object (if traceOnly is set)
|
|
// because if we don't, the message will end-up being delivered to
|
|
// applications, which may break them. We report the error in any case.
|
|
c.Errorf("Destination %q is not valid, won't be able to trace events", dest)
|
|
if !traceOnly {
|
|
// We can bail, tracing will be disabled for this message.
|
|
return nil
|
|
}
|
|
}
|
|
}
|
|
var (
|
|
// Account to use when sending the trace event
|
|
acc *Account
|
|
// Ingress' account name
|
|
ian string
|
|
// Origin account name
|
|
oan string
|
|
// The hop "id", taken from headers only when not from CLIENT
|
|
hop string
|
|
)
|
|
if c.kind == ROUTER || c.kind == GATEWAY || c.kind == LEAF {
|
|
// The ingress account name will always be c.pa.account, but `acc` may
|
|
// be different if we have an origin account header.
|
|
if c.kind == LEAF {
|
|
ian = c.acc.GetName()
|
|
} else {
|
|
ian = string(c.pa.account)
|
|
}
|
|
// The remote will have set the origin account header only if the
|
|
// message changed account (think of service imports).
|
|
oan = getHdrVal(MsgTraceOriginAccount)
|
|
if oan == _EMPTY_ {
|
|
// For LEAF or ROUTER with pinned-account, we can use the c.acc.
|
|
if c.kind == LEAF || (c.kind == ROUTER && len(c.route.accName) > 0) {
|
|
acc = c.acc
|
|
} else {
|
|
// We will lookup account with c.pa.account (or ian).
|
|
oan = ian
|
|
}
|
|
}
|
|
// Unless we already got the account, we need to look it up.
|
|
if acc == nil {
|
|
// We don't want to do account resolving here.
|
|
if acci, ok := c.srv.accounts.Load(oan); ok {
|
|
acc = acci.(*Account)
|
|
// Since we have looked-up the account, we don't need oan, so
|
|
// clear it in case it was set.
|
|
oan = _EMPTY_
|
|
} else {
|
|
// We still have to return a msgTrace object (if traceOnly is set)
|
|
// because if we don't, the message will end-up being delivered to
|
|
// applications, which may break them. We report the error in any case.
|
|
c.Errorf("Account %q was not found, won't be able to trace events", oan)
|
|
if !traceOnly {
|
|
// We can bail, tracing will be disabled for this message.
|
|
return nil
|
|
}
|
|
}
|
|
}
|
|
// Check the hop header
|
|
hop = getHdrVal(MsgTraceHop)
|
|
} else {
|
|
acc = c.acc
|
|
ian = acc.GetName()
|
|
}
|
|
// If external, we need to have the account's trace destination set,
|
|
// otherwise, we are not enabling tracing.
|
|
if external {
|
|
var sampling int
|
|
if acc != nil {
|
|
dest, sampling = acc.getTraceDestAndSampling()
|
|
}
|
|
if dest == _EMPTY_ {
|
|
// No account destination, no tracing for external trace headers.
|
|
return nil
|
|
}
|
|
// Check sampling, but only from origin server.
|
|
if c.kind == CLIENT && !sample(sampling) {
|
|
// Need to disable tracing so that if the message is routed, it won't
|
|
// trigger a trace there.
|
|
c.msgBuf = c.setHeader(MsgTraceDest, MsgTraceDestDisabled, c.msgBuf)
|
|
return nil
|
|
}
|
|
}
|
|
c.pa.trace = &msgTrace{
|
|
srv: c.srv,
|
|
acc: acc,
|
|
oan: oan,
|
|
dest: dest,
|
|
ct: getCompressionType(getHdrVal(acceptEncodingHeader)),
|
|
hop: hop,
|
|
event: &MsgTraceEvent{
|
|
Request: MsgTraceRequest{
|
|
Header: headers,
|
|
MsgSize: c.pa.size,
|
|
},
|
|
Events: append(MsgTraceEvents(nil), &MsgTraceIngress{
|
|
MsgTraceBase: MsgTraceBase{
|
|
Type: MsgTraceIngressType,
|
|
Timestamp: time.Now(),
|
|
},
|
|
Kind: c.kind,
|
|
CID: c.cid,
|
|
Name: getConnName(c),
|
|
Account: ian,
|
|
Subject: string(c.pa.subject),
|
|
}),
|
|
},
|
|
tonly: traceOnly,
|
|
}
|
|
return c.pa.trace
|
|
}
|
|
|
|
func sample(sampling int) bool {
|
|
// Option parsing should ensure that sampling is [1..100], but consider
|
|
// any value outside of this range to be 100%.
|
|
if sampling <= 0 || sampling >= 100 {
|
|
return true
|
|
}
|
|
return rand.Int31n(100) <= int32(sampling)
|
|
}
|
|
|
|
// This function will return the header as a map (instead of http.Header because
|
|
// we want to preserve the header names' case) and a boolean that indicates if
|
|
// the headers have been lifted due to the presence of the external trace header
|
|
// only.
|
|
// Note that because of the traceParentHdr, the search is done in a case
|
|
// insensitive way. We used to rewrite it in lower case but no longer do since v2.14.
|
|
func genHeaderMapIfTraceHeadersPresent(hdr []byte) (map[string][]string, bool) {
|
|
|
|
var (
|
|
_keys = [64][]byte{}
|
|
_vals = [64][]byte{}
|
|
m map[string][]string
|
|
traceDestHdrFound bool
|
|
traceParentHdrFound bool
|
|
)
|
|
// Skip the hdrLine
|
|
if !bytes.HasPrefix(hdr, stringToBytes(hdrLine)) {
|
|
return nil, false
|
|
}
|
|
|
|
keys := _keys[:0]
|
|
vals := _vals[:0]
|
|
|
|
for i := len(hdrLine); i < len(hdr); {
|
|
// Search for key/val delimiter
|
|
del := bytes.IndexByte(hdr[i:], ':')
|
|
if del < 0 {
|
|
break
|
|
}
|
|
keyStart := i
|
|
key := hdr[keyStart : keyStart+del]
|
|
i += del + 1
|
|
for i < len(hdr) && (hdr[i] == ' ' || hdr[i] == '\t') {
|
|
i++
|
|
}
|
|
valStart := i
|
|
nl := bytes.Index(hdr[valStart:], crLFAsBytes)
|
|
if nl < 0 {
|
|
break
|
|
}
|
|
valEnd := valStart + nl
|
|
for valEnd > valStart && (hdr[valEnd-1] == ' ' || hdr[valEnd-1] == '\t') {
|
|
valEnd--
|
|
}
|
|
val := hdr[valStart:valEnd]
|
|
if len(key) > 0 && len(val) > 0 {
|
|
vals = append(vals, val)
|
|
|
|
// We search for our special keys only if not already found.
|
|
|
|
// Check for the external trace header.
|
|
// Search needs to be case insensitive.
|
|
if !traceParentHdrFound && bytes.EqualFold(key, traceParentHdrAsBytes) {
|
|
// We will now check if the value has sampling or not.
|
|
// TODO(ik): Not sure if this header can have multiple values
|
|
// or not, and if so, what would be the rule to check for
|
|
// sampling. What is done here is to check them all until we
|
|
// found one with sampling.
|
|
tk := bytes.Split(val, dashAsBytes)
|
|
if len(tk) == 4 && len([]byte(tk[3])) == 2 {
|
|
if hexVal, err := strconv.ParseInt(bytesToString(tk[3]), 16, 8); err == nil {
|
|
if hexVal&0x1 == 0x1 {
|
|
traceParentHdrFound = true
|
|
}
|
|
}
|
|
}
|
|
} else if !traceDestHdrFound && bytes.Equal(key, traceDestHdrAsBytes) {
|
|
// This is the Nats-Trace-Dest header, check the value to see
|
|
// if it indicates that the trace was disabled.
|
|
if bytes.Equal(val, traceDestDisabledAsBytes) {
|
|
return nil, false
|
|
}
|
|
traceDestHdrFound = true
|
|
}
|
|
// Add to the keys and preserve the key's case
|
|
keys = append(keys, key)
|
|
}
|
|
i += nl + 2
|
|
}
|
|
if !traceDestHdrFound && !traceParentHdrFound {
|
|
return nil, false
|
|
}
|
|
m = make(map[string][]string, len(keys))
|
|
for i, k := range keys {
|
|
hname := string(k)
|
|
m[hname] = append(m[hname], string(vals[i]))
|
|
}
|
|
return m, !traceDestHdrFound && traceParentHdrFound
|
|
}
|
|
|
|
// Special case where we create a trace event before parsing the message.
|
|
// This is for cases where the connection will be closed when detecting
|
|
// an error during early message processing (for instance max payload).
|
|
func (c *client) initAndSendIngressErrEvent(hdr []byte, dest string, ingressError error) {
|
|
if ingressError == nil {
|
|
return
|
|
}
|
|
ct := getAcceptEncoding(hdr)
|
|
t := &msgTrace{
|
|
srv: c.srv,
|
|
acc: c.acc,
|
|
dest: dest,
|
|
ct: ct,
|
|
event: &MsgTraceEvent{
|
|
Request: MsgTraceRequest{MsgSize: c.pa.size},
|
|
Events: append(MsgTraceEvents(nil), &MsgTraceIngress{
|
|
MsgTraceBase: MsgTraceBase{
|
|
Type: MsgTraceIngressType,
|
|
Timestamp: time.Now(),
|
|
},
|
|
Kind: c.kind,
|
|
CID: c.cid,
|
|
Name: getConnName(c),
|
|
Error: ingressError.Error(),
|
|
}),
|
|
},
|
|
}
|
|
t.sendEvent()
|
|
}
|
|
|
|
// Returns `true` if message tracing is enabled and we are tracing only,
|
|
// that is, we are not going to deliver the inbound message, returns
|
|
// `false` otherwise (no tracing, or tracing and message delivery).
|
|
func (t *msgTrace) traceOnly() bool {
|
|
return t != nil && t.tonly
|
|
}
|
|
|
|
func (t *msgTrace) setOriginAccountHeaderIfNeeded(c *client, acc *Account, msg []byte) []byte {
|
|
var oan string
|
|
// If t.acc is set, only check that, not t.oan.
|
|
if t.acc != nil {
|
|
if t.acc != acc {
|
|
oan = t.acc.GetName()
|
|
}
|
|
} else if t.oan != acc.GetName() {
|
|
oan = t.oan
|
|
}
|
|
if oan != _EMPTY_ {
|
|
msg = c.setHeader(MsgTraceOriginAccount, oan, msg)
|
|
}
|
|
return msg
|
|
}
|
|
|
|
func (t *msgTrace) setHopHeader(c *client, msg []byte) []byte {
|
|
e := t.event
|
|
e.Hops++
|
|
if len(t.hop) > 0 {
|
|
t.nhop = fmt.Sprintf("%s.%d", t.hop, e.Hops)
|
|
} else {
|
|
t.nhop = fmt.Sprintf("%d", e.Hops)
|
|
}
|
|
return c.setHeader(MsgTraceHop, t.nhop, msg)
|
|
}
|
|
|
|
func (t *msgTrace) setIngressError(err string) {
|
|
if i := t.event.Ingress(); i != nil {
|
|
i.Error = err
|
|
}
|
|
}
|
|
|
|
func (t *msgTrace) addSubjectMappingEvent(subj []byte) {
|
|
if t == nil {
|
|
return
|
|
}
|
|
t.event.Events = append(t.event.Events, &MsgTraceSubjectMapping{
|
|
MsgTraceBase: MsgTraceBase{
|
|
Type: MsgTraceSubjectMappingType,
|
|
Timestamp: time.Now(),
|
|
},
|
|
MappedTo: string(subj),
|
|
})
|
|
}
|
|
|
|
func (t *msgTrace) addEgressEvent(dc *client, sub *subscription, err string) {
|
|
if t == nil {
|
|
return
|
|
}
|
|
e := &MsgTraceEgress{
|
|
MsgTraceBase: MsgTraceBase{
|
|
Type: MsgTraceEgressType,
|
|
Timestamp: time.Now(),
|
|
},
|
|
Kind: dc.kind,
|
|
CID: dc.cid,
|
|
Name: getConnName(dc),
|
|
Hop: t.nhop,
|
|
Error: err,
|
|
}
|
|
t.nhop = _EMPTY_
|
|
// Specific to CLIENT connections...
|
|
if dc.kind == CLIENT {
|
|
// Set the subscription's subject and possibly queue name.
|
|
e.Subscription = string(sub.subject)
|
|
if len(sub.queue) > 0 {
|
|
e.Queue = string(sub.queue)
|
|
}
|
|
}
|
|
if dc.kind == CLIENT || dc.kind == LEAF {
|
|
if i := t.event.Ingress(); i != nil {
|
|
// If the Ingress' account is different from the destination's
|
|
// account, add the account name into the Egress trace event.
|
|
// This would happen with service imports.
|
|
if dcAccName := dc.acc.GetName(); dcAccName != i.Account {
|
|
e.Account = dcAccName
|
|
}
|
|
}
|
|
}
|
|
t.event.Events = append(t.event.Events, e)
|
|
}
|
|
|
|
func (t *msgTrace) addStreamExportEvent(dc *client, to []byte) {
|
|
if t == nil {
|
|
return
|
|
}
|
|
dc.mu.Lock()
|
|
accName := dc.acc.GetName()
|
|
dc.mu.Unlock()
|
|
t.event.Events = append(t.event.Events, &MsgTraceStreamExport{
|
|
MsgTraceBase: MsgTraceBase{
|
|
Type: MsgTraceStreamExportType,
|
|
Timestamp: time.Now(),
|
|
},
|
|
Account: accName,
|
|
To: string(to),
|
|
})
|
|
}
|
|
|
|
func (t *msgTrace) addServiceImportEvent(accName, from, to string) {
|
|
if t == nil {
|
|
return
|
|
}
|
|
t.event.Events = append(t.event.Events, &MsgTraceServiceImport{
|
|
MsgTraceBase: MsgTraceBase{
|
|
Type: MsgTraceServiceImportType,
|
|
Timestamp: time.Now(),
|
|
},
|
|
Account: accName,
|
|
From: from,
|
|
To: to,
|
|
})
|
|
}
|
|
|
|
func (t *msgTrace) addJetStreamEvent(streamName string) {
|
|
if t == nil {
|
|
return
|
|
}
|
|
t.js = &MsgTraceJetStream{
|
|
MsgTraceBase: MsgTraceBase{
|
|
Type: MsgTraceJetStreamType,
|
|
Timestamp: time.Now(),
|
|
},
|
|
Stream: streamName,
|
|
}
|
|
t.event.Events = append(t.event.Events, t.js)
|
|
}
|
|
|
|
func (t *msgTrace) updateJetStreamEvent(subject string, noInterest bool) {
|
|
if t == nil {
|
|
return
|
|
}
|
|
// JetStream event should have been created in addJetStreamEvent
|
|
if t.js == nil {
|
|
return
|
|
}
|
|
t.js.Subject = subject
|
|
t.js.NoInterest = noInterest
|
|
// Update the timestamp since this is more accurate than when it
|
|
// was first added in addJetStreamEvent().
|
|
t.js.Timestamp = time.Now()
|
|
}
|
|
|
|
func (t *msgTrace) sendEventFromJetStream(err error) {
|
|
if t == nil {
|
|
return
|
|
}
|
|
// JetStream event should have been created in addJetStreamEvent
|
|
if t.js == nil {
|
|
return
|
|
}
|
|
if err != nil {
|
|
t.js.Error = err.Error()
|
|
}
|
|
t.sendEvent()
|
|
}
|
|
|
|
func (t *msgTrace) sendEvent() {
|
|
if t == nil {
|
|
return
|
|
}
|
|
if t.js != nil {
|
|
ready := atomic.AddInt32(&t.ready, 1) == 2
|
|
if !ready {
|
|
return
|
|
}
|
|
}
|
|
t.srv.sendInternalAccountSysMsg(t.acc, t.dest, &t.event.Server, t.event, t.ct)
|
|
}
|