hanzomsg/nats-server
1
0
Fork 0
mirror of https://github.com/hanzomsg/nats-server.git synced 2026-10-30 20:27:34 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
main
nats-server/server/msgtrace_test.go
T
Ivan Kozlovic c331e7fde4 (2.14) [CHANGED] Message Tracing: Do not rewrite traceparent header 
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]>
2026-01-26 12:07:15 -07:00

5266 lines
162 KiB
Go
Raw Permalink Blame History

// 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.
//go:build !skip_msgtrace_tests
package server
import (
"bytes"
"compress/gzip"
"encoding/json"
"fmt"
"io"
"net"
"strings"
"sync/atomic"
"testing"
"time"
"github.com/klauspost/compress/s2"
"github.com/nats-io/jwt/v2"
"github.com/nats-io/nats.go"
"github.com/nats-io/nkeys"
)
func init() {
msgTraceRunInTests = true
}
func TestMsgTraceConnName(t *testing.T) {
c := &client{kind: ROUTER, route: &route{remoteName: "somename"}}
c.opts.Name = "someid"
// If route.remoteName is set, it will take precedence.
val := getConnName(c)
require_Equal[string](t, val, "somename")
// When not set, we revert to c.opts.Name
c.route.remoteName = _EMPTY_
val = getConnName(c)
require_Equal[string](t, val, "someid")
// Now same for GW.
c.route = nil
c.gw = &gateway{remoteName: "somename"}
c.kind = GATEWAY
val = getConnName(c)
require_Equal[string](t, val, "somename")
// Revert to c.opts.Name
c.gw.remoteName = _EMPTY_
val = getConnName(c)
require_Equal[string](t, val, "someid")
// For LeafNode now
c.gw = nil
c.leaf = &leaf{remoteServer: "somename"}
c.kind = LEAF
val = getConnName(c)
require_Equal[string](t, val, "somename")
// But if not set...
c.leaf.remoteServer = _EMPTY_
val = getConnName(c)
require_Equal[string](t, val, "someid")
c.leaf = nil
c.kind = CLIENT
val = getConnName(c)
require_Equal[string](t, val, "someid")
}
func TestMsgTraceGenHeaderMap(t *testing.T) {
traceparentDifferentCase := "TrAcEpArEnT"
for _, test := range []struct {
name string
header []byte
expected map[string][]string
external bool
}{
{"missing header line", []byte("Nats-Trace-Dest: val\r\n"), nil, false},
{"no trace header present", []byte(hdrLine + "Header1: val1\r\nHeader2: val2\r\n"), nil, false},
{"trace header with some prefix", []byte(hdrLine + "Some-Prefix-" + MsgTraceDest + ": some value\r\n"), nil, false},
{"trace header with some suffix", []byte(hdrLine + MsgTraceDest + "-Some-Suffix: some value\r\n"), nil, false},
{"trace header with space before colon", []byte(hdrLine + MsgTraceDest + " : some value\r\n"), nil, false},
{"trace header with missing cr_lf for value", []byte(hdrLine + MsgTraceDest + ": bogus"), nil, false},
{"trace header with empty value", []byte(hdrLine + MsgTraceDest + ": \r\n"), nil, false},
{"external trace header with some prefix", []byte(hdrLine + "Some-Prefix-" + traceParentHdr + ": 00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01\r\n"), nil, false},
{"external trace header with some suffix", []byte(hdrLine + traceParentHdr + "-Some-Suffix: 00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01\r\n"), nil, false},
{"external header with space before colon", []byte(hdrLine + traceParentHdr + " : 00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01\r\n"), nil, false},
{"external header with missing cr_lf for value", []byte(hdrLine + traceParentHdr + ": 00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"), nil, false},
{"external header with empty value", []byte(hdrLine + traceParentHdr + ": \r\n"), nil, false},
{"trace header first", []byte(hdrLine + MsgTraceDest + ": some.dest\r\nSome-Header: some value\r\n"),
map[string][]string{"Some-Header": {"some value"}, MsgTraceDest: {"some.dest"}}, false},
{"trace header last", []byte(hdrLine + "Some-Header: some value\r\n" + MsgTraceDest + ": some.dest\r\n"),
map[string][]string{"Some-Header": {"some value"}, MsgTraceDest: {"some.dest"}}, false},
{"trace header multiple values", []byte(hdrLine + MsgTraceDest + ": some.dest\r\nSome-Header: some value\r\n" + MsgTraceDest + ": some.dest.2"),
map[string][]string{"Some-Header": {"some value"}, MsgTraceDest: {"some.dest", "some.dest.2"}}, false},
{"trace header and some empty key", []byte(hdrLine + MsgTraceDest + ": some.dest\r\n: bogus\r\nSome-Header: some value\r\n"),
map[string][]string{"Some-Header": {"some value"}, MsgTraceDest: {"some.dest"}}, false},
{"trace header and some header missing cr_lf for value", []byte(hdrLine + MsgTraceDest + ": some.dest\r\nSome-Header: bogus"),
map[string][]string{MsgTraceDest: {"some.dest"}}, false},
{"trace header and trims value", []byte(hdrLine + MsgTraceDest + ": some.dest \r\n"),
map[string][]string{MsgTraceDest: {"some.dest"}}, false},
{"trace header and external after", []byte(hdrLine + MsgTraceDest + ": some.dest\r\n" + traceParentHdr + ": 00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01\r\nSome-Header: some value\r\n"),
map[string][]string{"Some-Header": {"some value"}, MsgTraceDest: {"some.dest"}, traceParentHdr: {"00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"}}, false},
{"trace header and external before", []byte(hdrLine + traceParentHdr + ": 00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01\r\n" + MsgTraceDest + ": some.dest\r\nSome-Header: some value\r\n"),
map[string][]string{"Some-Header": {"some value"}, MsgTraceDest: {"some.dest"}, traceParentHdr: {"00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"}}, false},
{"external malformed", []byte(hdrLine + traceParentHdr + ": 00-4bf92f3577b34da6a3ce929d0e0e4736-01\r\n"), nil, false},
{"external first and sampling", []byte(hdrLine + traceParentHdr + ": 00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01\r\nSome-Header: some value\r\n"),
map[string][]string{"Some-Header": {"some value"}, traceParentHdr: {"00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"}}, true},
{"external middle and sampling", []byte(hdrLine + "Some-Header: some value1\r\n" + traceParentHdr + ": 00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01\r\nSome-Header: some value2\r\n"),
map[string][]string{"Some-Header": {"some value1", "some value2"}, traceParentHdr: {"00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"}}, true},
{"external last and sampling", []byte(hdrLine + "Some-Header: some value\r\n" + traceParentHdr + ": 00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01\r\n"),
map[string][]string{"Some-Header": {"some value"}, traceParentHdr: {"00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"}}, true},
{"external sampling with not just 01", []byte(hdrLine + traceParentHdr + ": 00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-27\r\nSome-Header: some value\r\n"),
map[string][]string{"Some-Header": {"some value"}, traceParentHdr: {"00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-27"}}, true},
{"external trims value", []byte(hdrLine + traceParentHdr + ": 00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01 \r\nSome-Header: some value\r\n"),
map[string][]string{"Some-Header": {"some value"}, traceParentHdr: {"00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"}}, true},
{"external with different case and sampling", []byte(hdrLine + traceparentDifferentCase + ": 00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01\r\nSome-Header: some value\r\n"),
map[string][]string{"Some-Header": {"some value"}, traceparentDifferentCase: {"00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"}}, true},
{"external first and not sampling", []byte(hdrLine + traceParentHdr + ": 00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-00\r\nSome-Header: some value\r\n"), nil, false},
{"external middle and not sampling", []byte(hdrLine + "Some-Header: some value1\r\n" + traceParentHdr + ": 00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-00\r\nSome-Header: some value2\r\n"), nil, false},
{"external last and not sampling", []byte(hdrLine + "Some-Header: some value\r\n" + traceParentHdr + ": 00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-00\r\n"), nil, false},
{"external not sampling with not just 00", []byte(hdrLine + traceParentHdr + ": 00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-22\r\nSome-Header: some value\r\n"), nil, false},
} {
t.Run(test.name, func(t *testing.T) {
m, ext := genHeaderMapIfTraceHeadersPresent(test.header)
if test.external != ext {
t.Fatalf("Expected external to be %v, got %v", test.external, ext)
}
if len(test.expected) != len(m) {
t.Fatalf("Expected map to be of size %v, got %v", len(test.expected), len(m))
}
// If external, we should find traceParentHdr
if test.external {
headerName := traceParentHdr
if _, ok := m[headerName]; !ok {
// There is a test where we use different case, so check for that one too.
headerName = traceparentDifferentCase
if _, ok := m[headerName]; !ok {
t.Fatalf("Expected traceparent header to be present, it was not: %+v", m)
}
}
// We no longer rewrite the header, so take that into consideration.
if !bytes.Contains(test.header, []byte(headerName)) {
t.Fatalf("Header should have been rewritten to have the traceparent in lower case: %s", test.header)
}
}
for k, vv := range m {
evv, ok := test.expected[k]
if !ok {
t.Fatalf("Did not find header %q in resulting map: %+v", k, m)
}
for i, v := range vv {
if evv[i] != v {
t.Fatalf("Expected value %v of key %q to be %q, got %q", i, k, evv[i], v)
}
}
}
})
}
}
func TestMsgTraceBasic(t *testing.T) {
conf := createConfFile(t, []byte(`
listen: 127.0.0.1:-1
mappings = {
foo: bar
}
`))
s, _ := RunServerWithConfig(conf)
defer s.Shutdown()
nc := natsConnect(t, s.ClientURL())
defer nc.Close()
cid, err := nc.GetClientID()
require_NoError(t, err)
traceSub := natsSubSync(t, nc, "my.trace.subj")
natsFlush(t, nc)
// Send trace message to a dummy subject to check that resulting trace's
// SubjectMapping and Egress are nil.
msg := nats.NewMsg("dummy")
msg.Header.Set(MsgTraceDest, traceSub.Subject)
msg.Header.Set(MsgTraceOnly, "true")
msg.Data = []byte("hello!")
err = nc.PublishMsg(msg)
require_NoError(t, err)
traceMsg := natsNexMsg(t, traceSub, time.Second)
var e MsgTraceEvent
json.Unmarshal(traceMsg.Data, &e)
require_Equal[string](t, e.Server.Name, s.Name())
// We don't remove the headers, so we will find the tracing header there.
require_True(t, e.Request.Header != nil)
require_Equal[int](t, len(e.Request.Header), 2)
// The message size is 6 + whatever size for the 2 trace headers.
// Let's just make sure that size is > 20...
require_True(t, e.Request.MsgSize > 20)
ingress := e.Ingress()
require_True(t, ingress != nil)
require_True(t, ingress.Kind == CLIENT)
require_True(t, ingress.Timestamp != time.Time{})
require_Equal[uint64](t, ingress.CID, cid)
require_Equal[string](t, ingress.Name, _EMPTY_)
require_Equal[string](t, ingress.Account, globalAccountName)
require_Equal[string](t, ingress.Subject, "dummy")
require_Equal[string](t, ingress.Error, _EMPTY_)
require_True(t, e.SubjectMapping() == nil)
require_True(t, e.StreamExports() == nil)
require_True(t, e.ServiceImports() == nil)
require_True(t, e.JetStream() == nil)
require_True(t, e.Egresses() == nil)
// Now setup subscriptions that generate interest on the subject.
nc2 := natsConnect(t, s.ClientURL(), nats.Name("sub1And2"))
defer nc2.Close()
sub1 := natsSubSync(t, nc2, "bar")
sub2 := natsSubSync(t, nc2, "bar")
natsFlush(t, nc2)
nc2CID, _ := nc2.GetClientID()
nc3 := natsConnect(t, s.ClientURL())
defer nc3.Close()
sub3 := natsSubSync(t, nc3, "*")
natsFlush(t, nc3)
for _, test := range []struct {
name string
deliverMsg bool
}{
{"just trace", false},
{"deliver msg", true},
} {
t.Run(test.name, func(t *testing.T) {
msg = nats.NewMsg("foo")
msg.Header.Set("Some-App-Header", "some value")
msg.Header.Set(MsgTraceDest, traceSub.Subject)
if !test.deliverMsg {
msg.Header.Set(MsgTraceOnly, "true")
}
msg.Data = []byte("hello!")
err = nc.PublishMsg(msg)
require_NoError(t, err)
checkAppMsg := func(sub *nats.Subscription, expected bool) {
if expected {
appMsg := natsNexMsg(t, sub, time.Second)
require_Equal[string](t, string(appMsg.Data), "hello!")
// We don't remove message trace header, so we should have
// 2 headers (the app + trace destination)
require_True(t, len(appMsg.Header) == 2)
require_Equal[string](t, appMsg.Header.Get("Some-App-Header"), "some value")
require_Equal[string](t, appMsg.Header.Get(MsgTraceDest), traceSub.Subject)
}
// Check that no (more) messages are received.
if msg, err := sub.NextMsg(100 * time.Millisecond); err != nats.ErrTimeout {
t.Fatalf("Did not expect application message, got %s", msg.Data)
}
}
for _, sub := range []*nats.Subscription{sub1, sub2, sub3} {
checkAppMsg(sub, test.deliverMsg)
}
traceMsg = natsNexMsg(t, traceSub, time.Second)
e = MsgTraceEvent{}
json.Unmarshal(traceMsg.Data, &e)
require_Equal[string](t, e.Server.Name, s.Name())
require_True(t, e.Request.Header != nil)
// We should have the app header and the trace header(s) too.
expected := 2
if !test.deliverMsg {
// The "trace-only" header is added.
expected++
}
require_Equal[int](t, len(e.Request.Header), expected)
require_Equal[string](t, e.Request.Header["Some-App-Header"][0], "some value")
// The message size is 6 + whatever size for the 2 trace headers.
// Let's just make sure that size is > 20...
require_True(t, e.Request.MsgSize > 20)
ingress := e.Ingress()
require_True(t, ingress.Kind == CLIENT)
require_True(t, ingress.Timestamp != time.Time{})
require_Equal[string](t, ingress.Account, globalAccountName)
require_Equal[string](t, ingress.Subject, "foo")
sm := e.SubjectMapping()
require_True(t, sm != nil)
require_True(t, sm.Timestamp != time.Time{})
require_Equal[string](t, sm.MappedTo, "bar")
egress := e.Egresses()
require_Equal[int](t, len(egress), 3)
var sub1And2 int
for _, eg := range egress {
// All Egress should be clients
require_True(t, eg.Kind == CLIENT)
require_True(t, eg.Timestamp != time.Time{})
// For nc2CID, we should have two egress
if eg.CID == nc2CID {
// Check name
require_Equal[string](t, eg.Name, "sub1And2")
require_Equal[string](t, eg.Subscription, "bar")
sub1And2++
} else {
// No name set
require_Equal[string](t, eg.Name, _EMPTY_)
require_Equal[string](t, eg.Subscription, "*")
}
}
require_Equal[int](t, sub1And2, 2)
})
}
}
func TestMsgTraceIngressMaxPayloadError(t *testing.T) {
o := DefaultOptions()
o.MaxPayload = 1024
s := RunServer(o)
defer s.Shutdown()
nc := natsConnect(t, s.ClientURL())
defer nc.Close()
traceSub := natsSubSync(t, nc, "my.trace.subj")
natsSub(t, nc, "foo", func(_ *nats.Msg) {})
natsFlush(t, nc)
// Ensure the subscription is known by the server we're connected to.
checkSubInterest(t, s, globalAccountName, "my.trace.subj", time.Second)
checkSubInterest(t, s, globalAccountName, "foo", time.Second)
for _, test := range []struct {
name string
deliverMsg bool
}{
{"just trace", false},
{"deliver msg", true},
} {
t.Run(test.name, func(t *testing.T) {
nc2, err := net.Dial("tcp", fmt.Sprintf("127.0.0.1:%d", o.Port))
require_NoError(t, err)
defer nc2.Close()
nc2.Write([]byte("CONNECT {\"protocol\":1,\"headers\":true,\"no_responders\":true}\r\n"))
var traceOnlyHdr string
if !test.deliverMsg {
traceOnlyHdr = fmt.Sprintf("%s:true\r\n", MsgTraceOnly)
}
hdr := fmt.Sprintf("%s%s:%s\r\n%s\r\n", hdrLine, MsgTraceDest, traceSub.Subject, traceOnlyHdr)
hPub := fmt.Sprintf("HPUB foo %d 2048\r\n%sAAAAAAAAAAAAAAAAAA...", len(hdr), hdr)
nc2.Write([]byte(hPub))
traceMsg := natsNexMsg(t, traceSub, time.Second)
var e MsgTraceEvent
json.Unmarshal(traceMsg.Data, &e)
require_Equal[string](t, e.Server.Name, s.Name())
require_True(t, e.Request.Header == nil)
require_True(t, e.Ingress() != nil)
require_Contains(t, e.Ingress().Error, ErrMaxPayload.Error())
require_True(t, e.Egresses() == nil)
})
}
}
func TestMsgTraceIngressErrors(t *testing.T) {
conf := createConfFile(t, []byte(`
port: -1
accounts {
A {
users: [
{
user: a
password: pwd
permissions {
subscribe: ["my.trace.subj", "foo"]
publish {
allow: ["foo", "bar.>"]
deny: ["bar.baz"]
}
}
}
]
}
}
`))
s, _ := RunServerWithConfig(conf)
defer s.Shutdown()
nc := natsConnect(t, s.ClientURL(), nats.UserInfo("a", "pwd"))
defer nc.Close()
traceSub := natsSubSync(t, nc, "my.trace.subj")
natsSub(t, nc, "foo", func(_ *nats.Msg) {})
natsFlush(t, nc)
// Ensure the subscription is known by the server we're connected to.
checkSubInterest(t, s, "A", "my.trace.subj", time.Second)
checkSubInterest(t, s, "A", "foo", time.Second)
for _, test := range []struct {
name string
deliverMsg bool
}{
{"just trace", false},
{"deliver msg", true},
} {
t.Run(test.name, func(t *testing.T) {
nc2 := natsConnect(t, s.ClientURL(), nats.UserInfo("a", "pwd"))
defer nc2.Close()
sendMsg := func(subj, reply, errTxt string) {
msg := nats.NewMsg(subj)
msg.Header.Set(MsgTraceDest, traceSub.Subject)
if !test.deliverMsg {
msg.Header.Set(MsgTraceOnly, "true")
}
msg.Reply = reply
msg.Data = []byte("hello")
nc2.PublishMsg(msg)
traceMsg := natsNexMsg(t, traceSub, time.Second)
var e MsgTraceEvent
json.Unmarshal(traceMsg.Data, &e)
require_Equal[string](t, e.Server.Name, s.Name())
require_True(t, e.Request.Header != nil)
require_Contains(t, e.Ingress().Error, errTxt)
require_True(t, e.Egresses() == nil)
}
// Send to a subject that causes permission violation
sendMsg("bar.baz", _EMPTY_, "Permissions Violation for Publish to")
// Send to a subject that is reserved for GW replies
sendMsg(gwReplyPrefix+"foo", _EMPTY_, "Permissions Violation for Publish to")
// Send with a Reply that is reserved
sendMsg("foo", replyPrefix+"bar", "Permissions Violation for Publish with Reply of")
})
}
}
func TestMsgTraceEgressErrors(t *testing.T) {
conf := createConfFile(t, []byte(`
port: -1
accounts {
A {
users: [
{
user: a
password: pwd
permissions {
subscribe: {
allow: ["my.trace.subj", "foo", "bar.>"]
deny: "bar.bat"
}
publish {
allow: ["foo", "bar.>"]
deny: ["bar.baz"]
}
}
}
]
}
}
`))
s, _ := RunServerWithConfig(conf)
defer s.Shutdown()
nc := natsConnect(t, s.ClientURL(), nats.UserInfo("a", "pwd"))
defer nc.Close()
traceSub := natsSubSync(t, nc, "my.trace.subj")
natsFlush(t, nc)
// Ensure the subscription is known by the server we're connected to.
checkSubInterest(t, s, "A", "my.trace.subj", time.Second)
for _, test := range []struct {
name string
deliverMsg bool
}{
{"just trace", false},
{"deliver msg", true},
} {
t.Run(test.name, func(t *testing.T) {
sendMsg := func(pubc *nats.Conn, subj, errTxt string) {
t.Helper()
msg := nats.NewMsg(subj)
msg.Header.Set(MsgTraceDest, traceSub.Subject)
if !test.deliverMsg {
msg.Header.Set(MsgTraceOnly, "true")
}
msg.Data = []byte("hello")
pubc.PublishMsg(msg)
traceMsg := natsNexMsg(t, traceSub, time.Second)
var e MsgTraceEvent
json.Unmarshal(traceMsg.Data, &e)
require_Equal[string](t, e.Server.Name, s.Name())
egress := e.Egresses()
require_Equal[int](t, len(egress), 1)
require_Contains(t, egress[0].Error, errTxt)
}
// Test no-echo.
nc2 := natsConnect(t, s.ClientURL(), nats.UserInfo("a", "pwd"), nats.NoEcho())
defer nc2.Close()
natsSubSync(t, nc2, "foo")
sendMsg(nc2, "foo", errMsgTraceNoEcho)
nc2.Close()
// Test deny sub.
nc2 = natsConnect(t, s.ClientURL(), nats.UserInfo("a", "pwd"))
defer nc2.Close()
natsSubSync(t, nc2, "bar.>")
sendMsg(nc2, "bar.bat", errMsgTraceSubDeny)
nc2.Close()
// Test sub closed
nc2 = natsConnect(t, s.ClientURL(), nats.UserInfo("a", "pwd"))
defer nc2.Close()
natsSubSync(t, nc2, "bar.>")
natsFlush(t, nc2)
// Aritifially change the closed status of the subscription
cid, err := nc2.GetClientID()
require_NoError(t, err)
c := s.GetClient(cid)
c.mu.Lock()
for _, sub := range c.subs {
if string(sub.subject) == "bar.>" {
sub.close()
}
}
c.mu.Unlock()
sendMsg(nc2, "bar.bar", errMsgTraceSubClosed)
nc2.Close()
// The following applies only when doing delivery.
if test.deliverMsg {
// Test auto-unsub exceeded
nc2 = natsConnect(t, s.ClientURL(), nats.UserInfo("a", "pwd"))
defer nc2.Close()
sub := natsSubSync(t, nc2, "bar.>")
err := sub.AutoUnsubscribe(10)
require_NoError(t, err)
natsFlush(t, nc2)
// Modify sub.nm to be already over the 10 limit
cid, err := nc2.GetClientID()
require_NoError(t, err)
c := s.GetClient(cid)
c.mu.Lock()
for _, sub := range c.subs {
if string(sub.subject) == "bar.>" {
sub.nm = 20
}
}
c.mu.Unlock()
sendMsg(nc2, "bar.bar", errMsgTraceAutoSubExceeded)
nc2.Close()
// Test client closed
nc2 = natsConnect(t, s.ClientURL(), nats.UserInfo("a", "pwd"))
defer nc2.Close()
natsSubSync(t, nc2, "bar.>")
cid, err = nc2.GetClientID()
require_NoError(t, err)
c = s.GetClient(cid)
c.mu.Lock()
c.out.stc = make(chan struct{})
c.mu.Unlock()
msg := nats.NewMsg("bar.bar")
msg.Header.Set(MsgTraceDest, traceSub.Subject)
if !test.deliverMsg {
msg.Header.Set(MsgTraceOnly, "true")
}
msg.Data = []byte("hello")
nc2.PublishMsg(msg)
// This needs to be less than default stall time, which now is 2ms.
time.Sleep(1 * time.Millisecond)
c.mu.Lock()
c.flags.set(closeConnection)
c.mu.Unlock()
traceMsg := natsNexMsg(t, traceSub, time.Second)
var e MsgTraceEvent
json.Unmarshal(traceMsg.Data, &e)
require_Equal[string](t, e.Server.Name, s.Name())
egress := e.Egresses()
require_Equal[int](t, len(egress), 1)
require_Contains(t, egress[0].Error, errMsgTraceClientClosed)
c.mu.Lock()
c.flags.clear(closeConnection)
c.mu.Unlock()
nc2.Close()
}
})
}
}
func TestMsgTraceWithQueueSub(t *testing.T) {
o := DefaultOptions()
s := RunServer(o)
defer s.Shutdown()
nc := natsConnect(t, s.ClientURL())
defer nc.Close()
traceSub := natsSubSync(t, nc, "my.trace.subj")
natsFlush(t, nc)
nc2 := natsConnect(t, s.ClientURL(), nats.Name("sub1"))
defer nc2.Close()
sub1 := natsQueueSubSync(t, nc2, "foo", "bar")
natsFlush(t, nc2)
nc3 := natsConnect(t, s.ClientURL(), nats.Name("sub2"))
defer nc3.Close()
sub2 := natsQueueSubSync(t, nc3, "foo", "bar")
sub3 := natsQueueSubSync(t, nc3, "*", "baz")
natsFlush(t, nc3)
// Ensure the subscription is known by the server we're connected to.
checkSubInterest(t, s, globalAccountName, "my.trace.subj", time.Second)
checkSubInterest(t, s, globalAccountName, "foo", time.Second)
for _, test := range []struct {
name string
deliverMsg bool
}{
{"just trace", false},
{"deliver msg", true},
} {
t.Run(test.name, func(t *testing.T) {
msg := nats.NewMsg("foo")
msg.Header.Set(MsgTraceDest, traceSub.Subject)
if !test.deliverMsg {
msg.Header.Set(MsgTraceOnly, "true")
}
if !test.deliverMsg {
msg.Data = []byte("hello1")
} else {
msg.Data = []byte("hello2")
}
err := nc.PublishMsg(msg)
require_NoError(t, err)
if test.deliverMsg {
// Only one should have got the message...
msg1, err1 := sub1.NextMsg(100 * time.Millisecond)
msg2, err2 := sub2.NextMsg(100 * time.Millisecond)
if err1 == nil && err2 == nil {
t.Fatalf("Only one message should have been received")
}
var val string
if msg1 != nil {
val = string(msg1.Data)
} else {
val = string(msg2.Data)
}
require_Equal[string](t, val, "hello2")
// Queue baz should also have received the message
msg := natsNexMsg(t, sub3, time.Second)
require_Equal[string](t, string(msg.Data), "hello2")
}
// Check that no (more) messages are received.
for _, sub := range []*nats.Subscription{sub1, sub2, sub3} {
if msg, err := sub.NextMsg(100 * time.Millisecond); err != nats.ErrTimeout {
t.Fatalf("Expected no message, got %s", msg.Data)
}
}
traceMsg := natsNexMsg(t, traceSub, time.Second)
var e MsgTraceEvent
json.Unmarshal(traceMsg.Data, &e)
require_Equal[string](t, e.Server.Name, s.Name())
ingress := e.Ingress()
require_True(t, ingress != nil)
require_True(t, ingress.Kind == CLIENT)
require_Equal[string](t, ingress.Subject, "foo")
egress := e.Egresses()
require_Equal[int](t, len(egress), 2)
var qbar, qbaz int
for _, eg := range egress {
switch eg.Queue {
case "bar":
require_Equal[string](t, eg.Subscription, "foo")
qbar++
case "baz":
require_Equal[string](t, eg.Subscription, "*")
qbaz++
default:
t.Fatalf("Wrong queue name: %q", eg.Queue)
}
}
require_Equal[int](t, qbar, 1)
require_Equal[int](t, qbaz, 1)
})
}
}
func TestMsgTraceWithRoutes(t *testing.T) {
tmpl := `
port: -1
accounts {
A { users: [{user:A, password: pwd}] }
B { users: [{user:B, password: pwd}] }
}
cluster {
name: "local"
port: -1
accounts: ["A"]
%s
}
`
conf1 := createConfFile(t, []byte(fmt.Sprintf(tmpl, _EMPTY_)))
s1, o1 := RunServerWithConfig(conf1)
defer s1.Shutdown()
conf2 := createConfFile(t, []byte(fmt.Sprintf(tmpl, fmt.Sprintf("routes: [\"nats://127.0.0.1:%d\"]", o1.Cluster.Port))))
s2, _ := RunServerWithConfig(conf2)
defer s2.Shutdown()
checkClusterFormed(t, s1, s2)
checkDummy := func(user string) {
nc := natsConnect(t, s1.ClientURL(), nats.UserInfo(user, "pwd"))
defer nc.Close()
traceSub := natsSubSync(t, nc, "my.trace.subj")
natsFlush(t, nc)
// Send trace message to a dummy subject to check that resulting trace
// is as expected.
msg := nats.NewMsg("dummy")
msg.Header.Set(MsgTraceDest, traceSub.Subject)
msg.Header.Set(MsgTraceOnly, "true")
msg.Data = []byte("hello!")
err := nc.PublishMsg(msg)
require_NoError(t, err)
traceMsg := natsNexMsg(t, traceSub, time.Second)
var e MsgTraceEvent
json.Unmarshal(traceMsg.Data, &e)
require_Equal[string](t, e.Server.Name, s1.Name())
ingress := e.Ingress()
require_True(t, ingress != nil)
require_True(t, ingress.Kind == CLIENT)
// "user" is same than account name in this test.
require_Equal[string](t, ingress.Account, user)
require_Equal[string](t, ingress.Subject, "dummy")
require_True(t, e.SubjectMapping() == nil)
require_True(t, e.Egresses() == nil)
// We should also not get an event from the remote server.
if msg, err := traceSub.NextMsg(100 * time.Millisecond); err != nats.ErrTimeout {
t.Fatalf("Expected no message, got %s", msg.Data)
}
}
checkDummy("A")
checkDummy("B")
for _, test := range []struct {
name string
acc string
}{
{"pinned account", "A"},
{"reg account", "B"},
} {
t.Run(test.name, func(t *testing.T) {
acc := test.acc
// Now create subscriptions on both s1 and s2
nc2 := natsConnect(t, s2.ClientURL(), nats.UserInfo(acc, "pwd"), nats.Name("sub2"))
defer nc2.Close()
sub2 := natsQueueSubSync(t, nc2, "foo.*", "my_queue")
nc3 := natsConnect(t, s2.ClientURL(), nats.UserInfo(acc, "pwd"), nats.Name("sub3"))
defer nc3.Close()
sub3 := natsQueueSubSync(t, nc3, "*.*", "my_queue_2")
checkSubInterest(t, s1, acc, "foo.bar", time.Second)
nc1 := natsConnect(t, s1.ClientURL(), nats.UserInfo(acc, "pwd"), nats.Name("sub1"))
defer nc1.Close()
sub1 := natsSubSync(t, nc1, "*.bar")
nct := natsConnect(t, s1.ClientURL(), nats.UserInfo(acc, "pwd"), nats.Name("tracer"))
defer nct.Close()
traceSub := natsSubSync(t, nct, "my.trace.subj")
for _, test := range []struct {
name string
deliverMsg bool
}{
{"just trace", false},
{"deliver msg", true},
} {
t.Run(test.name, func(t *testing.T) {
msg := nats.NewMsg("foo.bar")
msg.Header.Set(MsgTraceDest, traceSub.Subject)
if !test.deliverMsg {
msg.Header.Set(MsgTraceOnly, "true")
}
msg.Data = []byte("hello!")
err := nct.PublishMsg(msg)
require_NoError(t, err)
checkAppMsg := func(sub *nats.Subscription, expected bool) {
if expected {
appMsg := natsNexMsg(t, sub, time.Second)
require_Equal[string](t, string(appMsg.Data), "hello!")
}
// Check that no (more) messages are received.
if msg, err := sub.NextMsg(100 * time.Millisecond); err != nats.ErrTimeout {
t.Fatalf("Did not expect application message, got %s", msg.Data)
}
}
for _, sub := range []*nats.Subscription{sub1, sub2, sub3} {
checkAppMsg(sub, test.deliverMsg)
}
check := func() {
traceMsg := natsNexMsg(t, traceSub, time.Second)
var e MsgTraceEvent
json.Unmarshal(traceMsg.Data, &e)
ingress := e.Ingress()
require_True(t, ingress != nil)
switch ingress.Kind {
case CLIENT:
require_Equal[string](t, e.Server.Name, s1.Name())
require_Equal[string](t, ingress.Account, acc)
require_Equal[string](t, ingress.Subject, "foo.bar")
egress := e.Egresses()
require_Equal[int](t, len(egress), 2)
for _, eg := range egress {
if eg.Kind == CLIENT {
require_Equal[string](t, eg.Name, "sub1")
require_Equal[string](t, eg.Subscription, "*.bar")
require_Equal[string](t, eg.Queue, _EMPTY_)
} else {
require_True(t, eg.Kind == ROUTER)
require_Equal[string](t, eg.Name, s2.Name())
require_Equal[string](t, eg.Subscription, _EMPTY_)
require_Equal[string](t, eg.Queue, _EMPTY_)
}
}
case ROUTER:
require_Equal[string](t, e.Server.Name, s2.Name())
require_Equal[string](t, ingress.Account, acc)
require_Equal[string](t, ingress.Subject, "foo.bar")
egress := e.Egresses()
require_Equal[int](t, len(egress), 2)
var gotSub2, gotSub3 int
for _, eg := range egress {
require_True(t, eg.Kind == CLIENT)
switch eg.Name {
case "sub2":
require_Equal[string](t, eg.Subscription, "foo.*")
require_Equal[string](t, eg.Queue, "my_queue")
gotSub2++
case "sub3":
require_Equal[string](t, eg.Subscription, "*.*")
require_Equal[string](t, eg.Queue, "my_queue_2")
gotSub3++
default:
t.Fatalf("Unexpected egress name: %+v", eg)
}
}
require_Equal[int](t, gotSub2, 1)
require_Equal[int](t, gotSub3, 1)
default:
t.Fatalf("Unexpected ingress: %+v", ingress)
}
}
// We should get 2 events. Order is not guaranteed.
check()
check()
// Make sure we are not receiving more traces
if tm, err := traceSub.NextMsg(250 * time.Millisecond); err == nil {
t.Fatalf("Should not have received trace message: %s", tm.Data)
}
})
}
})
}
}
func TestMsgTraceWithRouteToOldServer(t *testing.T) {
msgTraceCheckSupport = true
defer func() { msgTraceCheckSupport = false }()
tmpl := `
port: -1
cluster {
name: "local"
port: -1
pool_size: -1
%s
}
`
conf1 := createConfFile(t, []byte(fmt.Sprintf(tmpl, _EMPTY_)))
s1, o1 := RunServerWithConfig(conf1)
defer s1.Shutdown()
conf2 := createConfFile(t, []byte(fmt.Sprintf(tmpl, fmt.Sprintf("routes: [\"nats://127.0.0.1:%d\"]", o1.Cluster.Port))))
o2 := LoadConfig(conf2)
// Make this server behave like an older server
o2.overrideProto = setServerProtoForTest(MsgTraceProto - 1)
s2 := RunServer(o2)
defer s2.Shutdown()
checkClusterFormed(t, s1, s2)
// Now create subscriptions on both s1 and s2
nc2 := natsConnect(t, s2.ClientURL(), nats.Name("sub2"))
defer nc2.Close()
sub2 := natsSubSync(t, nc2, "foo")
checkSubInterest(t, s1, globalAccountName, "foo", time.Second)
nc1 := natsConnect(t, s1.ClientURL(), nats.Name("sub1"))
defer nc1.Close()
sub1 := natsSubSync(t, nc1, "foo")
nct := natsConnect(t, s1.ClientURL(), nats.Name("tracer"))
defer nct.Close()
traceSub := natsSubSync(t, nct, "my.trace.subj")
// Ensure the subscription is known by the server we're connected to.
checkSubInterest(t, s1, globalAccountName, "my.trace.subj", time.Second)
checkSubInterest(t, s2, globalAccountName, "my.trace.subj", time.Second)
for _, test := range []struct {
name string
deliverMsg bool
}{
{"just trace", false},
{"deliver msg", true},
} {
t.Run(test.name, func(t *testing.T) {
msg := nats.NewMsg("foo")
msg.Header.Set(MsgTraceDest, traceSub.Subject)
if !test.deliverMsg {
msg.Header.Set(MsgTraceOnly, "true")
}
msg.Data = []byte("hello!")
err := nct.PublishMsg(msg)
require_NoError(t, err)
checkAppMsg := func(sub *nats.Subscription, expected bool) {
if expected {
appMsg := natsNexMsg(t, sub, time.Second)
require_Equal[string](t, string(appMsg.Data), "hello!")
}
// Check that no (more) messages are received.
if msg, err := sub.NextMsg(100 * time.Millisecond); err != nats.ErrTimeout {
t.Fatalf("Did not expect application message, got %s", msg.Data)
}
}
// Even if a server does not support tracing, as long as the header
// TraceOnly is not set, the message should be forwarded to the remote.
for _, sub := range []*nats.Subscription{sub1, sub2} {
checkAppMsg(sub, test.deliverMsg)
}
traceMsg := natsNexMsg(t, traceSub, time.Second)
var e MsgTraceEvent
json.Unmarshal(traceMsg.Data, &e)
ingress := e.Ingress()
require_True(t, ingress != nil)
require_True(t, ingress.Kind == CLIENT)
require_Equal[string](t, e.Server.Name, s1.Name())
egress := e.Egresses()
require_Equal[int](t, len(egress), 2)
for _, ci := range egress {
switch ci.Kind {
case CLIENT:
require_Equal[string](t, ci.Name, "sub1")
case ROUTER:
require_Equal[string](t, ci.Name, s2.Name())
if test.deliverMsg {
require_Contains(t, ci.Error, errMsgTraceNoSupport)
} else {
require_Contains(t, ci.Error, errMsgTraceOnlyNoSupport)
}
default:
t.Fatalf("Unexpected egress: %+v", ci)
}
}
// We should not get a second trace
if msg, err := traceSub.NextMsg(100 * time.Millisecond); err != nats.ErrTimeout {
t.Fatalf("Did not expect other trace, got %s", msg.Data)
}
})
}
}
func TestMsgTraceWithLeafNode(t *testing.T) {
for _, mainTest := range []struct {
name string
fromHub bool
leafUseLocalAcc bool
}{
{"from hub", true, false},
{"from leaf", false, false},
{"from hub with local account", true, true},
{"from leaf with local account", false, true},
} {
t.Run(mainTest.name, func(t *testing.T) {
confHub := createConfFile(t, []byte(`
port: -1
server_name: "A"
accounts {
A { users: [{user: "a", password: "pwd"}]}
B { users: [{user: "b", password: "pwd"}]}
}
leafnodes {
port: -1
}
`))
hub, ohub := RunServerWithConfig(confHub)
defer hub.Shutdown()
var accs string
var lacc string
if mainTest.leafUseLocalAcc {
accs = `accounts { B { users: [{user: "b", password: "pwd"}]} }`
lacc = `account: B`
}
confLeaf := createConfFile(t, []byte(fmt.Sprintf(`
port: -1
server_name: "B"
%s
leafnodes {
remotes [
{
url: "nats://a:[email protected]:%d"
%s
}
]
}
`, accs, ohub.LeafNode.Port, lacc)))
leaf, _ := RunServerWithConfig(confLeaf)
defer leaf.Shutdown()
checkLeafNodeConnected(t, hub)
checkLeafNodeConnected(t, leaf)
var s1, s2 *Server
if mainTest.fromHub {
s1, s2 = hub, leaf
} else {
s1, s2 = leaf, hub
}
// Now create subscriptions on both s1 and s2
opts := []nats.Option{nats.Name("sub2")}
var user string
// If fromHub, then it means that s2 is the leaf.
if mainTest.fromHub {
if mainTest.leafUseLocalAcc {
user = "b"
}
} else {
// s2 is the hub, always connect with user "a'"
user = "a"
}
if user != _EMPTY_ {
opts = append(opts, nats.UserInfo(user, "pwd"))
}
nc2 := natsConnect(t, s2.ClientURL(), opts...)
defer nc2.Close()
sub2 := natsSubSync(t, nc2, "foo")
if mainTest.fromHub {
checkSubInterest(t, s1, "A", "foo", time.Second)
} else if mainTest.leafUseLocalAcc {
checkSubInterest(t, s1, "B", "foo", time.Second)
} else {
checkSubInterest(t, s1, globalAccountName, "foo", time.Second)
}
user = _EMPTY_
opts = []nats.Option{nats.Name("sub1")}
if mainTest.fromHub {
// s1 is the hub, so we need user "a"
user = "a"
} else if mainTest.leafUseLocalAcc {
// s1 is the leaf, we need user "b" if leafUseLocalAcc
user = "b"
}
if user != _EMPTY_ {
opts = append(opts, nats.UserInfo(user, "pwd"))
}
nc1 := natsConnect(t, s1.ClientURL(), opts...)
defer nc1.Close()
sub1 := natsSubSync(t, nc1, "foo")
opts = []nats.Option{nats.Name("tracer")}
if user != _EMPTY_ {
opts = append(opts, nats.UserInfo(user, "pwd"))
}
nct := natsConnect(t, s1.ClientURL(), opts...)
defer nct.Close()
traceSub := natsSubSync(t, nct, "my.trace.subj")
for _, test := range []struct {
name string
deliverMsg bool
}{
{"just trace", false},
{"deliver msg", true},
} {
t.Run(test.name, func(t *testing.T) {
msg := nats.NewMsg("foo")
msg.Header.Set(MsgTraceDest, traceSub.Subject)
if !test.deliverMsg {
msg.Header.Set(MsgTraceOnly, "true")
}
msg.Data = []byte("hello!")
err := nct.PublishMsg(msg)
require_NoError(t, err)
checkAppMsg := func(sub *nats.Subscription, expected bool) {
if expected {
appMsg := natsNexMsg(t, sub, time.Second)
require_Equal[string](t, string(appMsg.Data), "hello!")
}
// Check that no (more) messages are received.
if msg, err := sub.NextMsg(100 * time.Millisecond); err != nats.ErrTimeout {
t.Fatalf("Did not expect application message, got %s", msg.Data)
}
}
for _, sub := range []*nats.Subscription{sub1, sub2} {
checkAppMsg(sub, test.deliverMsg)
}
check := func() {
traceMsg := natsNexMsg(t, traceSub, time.Second)
var e MsgTraceEvent
json.Unmarshal(traceMsg.Data, &e)
ingress := e.Ingress()
require_True(t, ingress != nil)
switch ingress.Kind {
case CLIENT:
require_Equal[string](t, e.Server.Name, s1.Name())
egress := e.Egresses()
require_Equal[int](t, len(egress), 2)
for _, eg := range egress {
switch eg.Kind {
case CLIENT:
require_Equal[string](t, eg.Name, "sub1")
case LEAF:
require_Equal[string](t, eg.Name, s2.Name())
require_Equal[string](t, eg.Error, _EMPTY_)
default:
t.Fatalf("Unexpected egress: %+v", eg)
}
}
case LEAF:
require_Equal[string](t, e.Server.Name, s2.Name())
require_True(t, ingress.Kind == LEAF)
require_Equal(t, ingress.Name, s1.Name())
egress := e.Egresses()
require_Equal[int](t, len(egress), 1)
eg := egress[0]
require_True(t, eg.Kind == CLIENT)
require_Equal[string](t, eg.Name, "sub2")
default:
t.Fatalf("Unexpected ingress: %+v", ingress)
}
}
check()
check()
// Make sure we are not receiving more traces
if tm, err := traceSub.NextMsg(250 * time.Millisecond); err == nil {
t.Fatalf("Should not have received trace message: %s", tm.Data)
}
})
}
})
}
}
func TestMsgTraceWithLeafNodeToOldServer(t *testing.T) {
msgTraceCheckSupport = true
defer func() { msgTraceCheckSupport = false }()
for _, mainTest := range []struct {
name string
fromHub bool
}{
{"from hub", true},
{"from leaf", false},
} {
t.Run(mainTest.name, func(t *testing.T) {
confHub := createConfFile(t, []byte(`
port: -1
server_name: "A"
leafnodes {
port: -1
}
`))
ohub := LoadConfig(confHub)
if !mainTest.fromHub {
// Make this server behave like an older server
ohub.overrideProto = setServerProtoForTest(MsgTraceProto - 1)
}
hub := RunServer(ohub)
defer hub.Shutdown()
confLeaf := createConfFile(t, []byte(fmt.Sprintf(`
port: -1
server_name: "B"
leafnodes {
remotes [{url: "nats://127.0.0.1:%d"}]
}
`, ohub.LeafNode.Port)))
oleaf := LoadConfig(confLeaf)
if mainTest.fromHub {
// Make this server behave like an older server
oleaf.overrideProto = setServerProtoForTest(MsgTraceProto - 1)
}
leaf := RunServer(oleaf)
defer leaf.Shutdown()
checkLeafNodeConnected(t, hub)
checkLeafNodeConnected(t, leaf)
var s1, s2 *Server
if mainTest.fromHub {
s1, s2 = hub, leaf
} else {
s1, s2 = leaf, hub
}
// Now create subscriptions on both s1 and s2
nc2 := natsConnect(t, s2.ClientURL(), nats.Name("sub2"))
defer nc2.Close()
sub2 := natsSubSync(t, nc2, "foo")
checkSubInterest(t, s1, globalAccountName, "foo", time.Second)
nc1 := natsConnect(t, s1.ClientURL(), nats.Name("sub1"))
defer nc1.Close()
sub1 := natsSubSync(t, nc1, "foo")
nct := natsConnect(t, s1.ClientURL(), nats.Name("tracer"))
defer nct.Close()
traceSub := natsSubSync(t, nct, "my.trace.subj")
for _, test := range []struct {
name string
deliverMsg bool
}{
{"just trace", false},
{"deliver msg", true},
} {
t.Run(test.name, func(t *testing.T) {
msg := nats.NewMsg("foo")
msg.Header.Set(MsgTraceDest, traceSub.Subject)
if !test.deliverMsg {
msg.Header.Set(MsgTraceOnly, "true")
}
msg.Data = []byte("hello!")
err := nct.PublishMsg(msg)
require_NoError(t, err)
checkAppMsg := func(sub *nats.Subscription, expected bool) {
if expected {
appMsg := natsNexMsg(t, sub, time.Second)
require_Equal[string](t, string(appMsg.Data), "hello!")
}
// Check that no (more) messages are received.
if msg, err := sub.NextMsg(100 * time.Millisecond); err != nats.ErrTimeout {
t.Fatalf("Did not expect application message, got %s", msg.Data)
}
}
// Even if a server does not support tracing, as long as the header
// TraceOnly is not set, the message should be forwarded to the remote.
for _, sub := range []*nats.Subscription{sub1, sub2} {
checkAppMsg(sub, test.deliverMsg)
}
traceMsg := natsNexMsg(t, traceSub, time.Second)
var e MsgTraceEvent
json.Unmarshal(traceMsg.Data, &e)
ingress := e.Ingress()
require_True(t, ingress != nil)
require_True(t, ingress.Kind == CLIENT)
require_Equal[string](t, e.Server.Name, s1.Name())
egress := e.Egresses()
require_Equal[int](t, len(egress), 2)
for _, ci := range egress {
switch ci.Kind {
case CLIENT:
require_Equal[string](t, ci.Name, "sub1")
case LEAF:
require_Equal[string](t, ci.Name, s2.Name())
if test.deliverMsg {
require_Contains(t, ci.Error, errMsgTraceNoSupport)
} else {
require_Contains(t, ci.Error, errMsgTraceOnlyNoSupport)
}
default:
t.Fatalf("Unexpected egress: %+v", ci)
}
}
// We should not get a second trace
if msg, err := traceSub.NextMsg(100 * time.Millisecond); err != nats.ErrTimeout {
t.Fatalf("Did not expect other trace, got %s", msg.Data)
}
})
}
})
}
}
func TestMsgTraceWithLeafNodeDaisyChain(t *testing.T) {
confHub := createConfFile(t, []byte(`
port: -1
server_name: "A"
accounts {
A { users: [{user: "a", password: "pwd"}]}
}
leafnodes {
port: -1
}
`))
hub, ohub := RunServerWithConfig(confHub)
defer hub.Shutdown()
confLeaf1 := createConfFile(t, []byte(fmt.Sprintf(`
port: -1
server_name: "B"
accounts {
B { users: [{user: "b", password: "pwd"}]}
}
leafnodes {
port: -1
remotes [{url: "nats://a:[email protected]:%d", account: B}]
}
`, ohub.LeafNode.Port)))
leaf1, oleaf1 := RunServerWithConfig(confLeaf1)
defer leaf1.Shutdown()
confLeaf2 := createConfFile(t, []byte(fmt.Sprintf(`
port: -1
server_name: "C"
accounts {
C { users: [{user: "c", password: "pwd"}]}
}
leafnodes {
remotes [{url: "nats://b:[email protected]:%d", account: C}]
}
`, oleaf1.LeafNode.Port)))
leaf2, _ := RunServerWithConfig(confLeaf2)
defer leaf2.Shutdown()
checkLeafNodeConnected(t, hub)
checkLeafNodeConnectedCount(t, leaf1, 2)
checkLeafNodeConnected(t, leaf2)
nct := natsConnect(t, hub.ClientURL(), nats.UserInfo("a", "pwd"), nats.Name("Tracer"))
defer nct.Close()
traceSub := natsSubSync(t, nct, "my.trace.subj")
natsFlush(t, nct)
// Make sure that subject interest travels down to leaf2
checkSubInterest(t, leaf2, "C", traceSub.Subject, time.Second)
nc1 := natsConnect(t, leaf1.ClientURL(), nats.UserInfo("b", "pwd"), nats.Name("sub1"))
defer nc1.Close()
nc2 := natsConnect(t, leaf2.ClientURL(), nats.UserInfo("c", "pwd"), nats.Name("sub2"))
defer nc2.Close()
sub2 := natsQueueSubSync(t, nc2, "foo.bar", "my_queue")
natsFlush(t, nc2)
// Check the subject interest makes it to leaf1
checkSubInterest(t, leaf1, "B", "foo.bar", time.Second)
// Now create the sub on leaf1
sub1 := natsSubSync(t, nc1, "foo.*")
natsFlush(t, nc1)
// Check that subject interest registered on "hub"
checkSubInterest(t, hub, "A", "foo.bar", time.Second)
for _, test := range []struct {
name string
deliverMsg bool
}{
{"just trace", false},
{"deliver msg", true},
} {
t.Run(test.name, func(t *testing.T) {
msg := nats.NewMsg("foo.bar")
msg.Header.Set(MsgTraceDest, traceSub.Subject)
if !test.deliverMsg {
msg.Header.Set(MsgTraceOnly, "true")
}
msg.Data = []byte("hello!")
err := nct.PublishMsg(msg)
require_NoError(t, err)
checkAppMsg := func(sub *nats.Subscription, expected bool) {
if expected {
appMsg := natsNexMsg(t, sub, time.Second)
require_Equal[string](t, string(appMsg.Data), "hello!")
}
// Check that no (more) messages are received.
if msg, err := sub.NextMsg(100 * time.Millisecond); err != nats.ErrTimeout {
t.Fatalf("Did not expect application message, got %s", msg.Data)
}
}
for _, sub := range []*nats.Subscription{sub1, sub2} {
checkAppMsg(sub, test.deliverMsg)
}
check := func() {
traceMsg := natsNexMsg(t, traceSub, time.Second)
var e MsgTraceEvent
json.Unmarshal(traceMsg.Data, &e)
ingress := e.Ingress()
require_True(t, ingress != nil)
switch ingress.Kind {
case CLIENT:
require_Equal[string](t, e.Server.Name, hub.Name())
require_Equal[string](t, ingress.Name, "Tracer")
require_Equal[string](t, ingress.Account, "A")
require_Equal[string](t, ingress.Subject, "foo.bar")
egress := e.Egresses()
require_Equal[int](t, len(egress), 1)
eg := egress[0]
require_True(t, eg.Kind == LEAF)
require_Equal[string](t, eg.Name, leaf1.Name())
require_Equal[string](t, eg.Account, _EMPTY_)
require_Equal[string](t, eg.Subscription, _EMPTY_)
case LEAF:
switch e.Server.Name {
case leaf1.Name():
require_Equal(t, ingress.Name, hub.Name())
require_Equal(t, ingress.Account, "B")
require_Equal[string](t, ingress.Subject, "foo.bar")
egress := e.Egresses()
require_Equal[int](t, len(egress), 2)
for _, eg := range egress {
switch eg.Kind {
case CLIENT:
require_Equal[string](t, eg.Name, "sub1")
require_Equal[string](t, eg.Subscription, "foo.*")
require_Equal[string](t, eg.Queue, _EMPTY_)
case LEAF:
require_Equal[string](t, eg.Name, leaf2.Name())
require_Equal[string](t, eg.Account, _EMPTY_)
require_Equal[string](t, eg.Subscription, _EMPTY_)
require_Equal[string](t, eg.Queue, _EMPTY_)
default:
t.Fatalf("Unexpected egress: %+v", eg)
}
}
case leaf2.Name():
require_Equal(t, ingress.Name, leaf1.Name())
require_Equal(t, ingress.Account, "C")
require_Equal[string](t, ingress.Subject, "foo.bar")
egress := e.Egresses()
require_Equal[int](t, len(egress), 1)
eg := egress[0]
require_True(t, eg.Kind == CLIENT)
require_Equal[string](t, eg.Name, "sub2")
require_Equal[string](t, eg.Subscription, "foo.bar")
require_Equal[string](t, eg.Queue, "my_queue")
default:
t.Fatalf("Unexpected ingress: %+v", ingress)
}
default:
t.Fatalf("Unexpected ingress: %+v", ingress)
}
}
check()
check()
check()
// Make sure we are not receiving more traces
if tm, err := traceSub.NextMsg(250 * time.Millisecond); err == nil {
t.Fatalf("Should not have received trace message: %s", tm.Data)
}
})
}
}
func TestMsgTraceWithGateways(t *testing.T) {
o2 := testDefaultOptionsForGateway("B")
o2.NoSystemAccount = false
s2 := runGatewayServer(o2)
defer s2.Shutdown()
o1 := testGatewayOptionsFromToWithServers(t, "A", "B", s2)
o1.NoSystemAccount = false
s1 := runGatewayServer(o1)
defer s1.Shutdown()
o3 := testGatewayOptionsFromToWithServers(t, "C", "B", s2)
o3.NoSystemAccount = false
s3 := runGatewayServer(o3)
defer s3.Shutdown()
waitForOutboundGateways(t, s1, 2, time.Second)
waitForInboundGateways(t, s1, 2, time.Second)
waitForInboundGateways(t, s2, 2, time.Second)
waitForOutboundGateways(t, s2, 2, time.Second)
waitForInboundGateways(t, s3, 2, time.Second)
waitForOutboundGateways(t, s3, 2, time.Second)
nc2 := natsConnect(t, s2.ClientURL(), nats.Name("sub2"))
defer nc2.Close()
sub2 := natsQueueSubSync(t, nc2, "foo.*", "my_queue_2")
nc22 := natsConnect(t, s2.ClientURL(), nats.Name("sub22"))
defer nc22.Close()
sub22 := natsQueueSubSync(t, nc22, "*.*", "my_queue_22")
nc3 := natsConnect(t, s3.ClientURL(), nats.Name("sub3"))
defer nc3.Close()
sub3 := natsQueueSubSync(t, nc3, "foo.*", "my_queue_3")
nc32 := natsConnect(t, s3.ClientURL(), nats.Name("sub32"))
defer nc32.Close()
sub32 := natsQueueSubSync(t, nc32, "*.*", "my_queue_32")
nc1 := natsConnect(t, s1.ClientURL(), nats.Name("sub1"))
defer nc1.Close()
sub1 := natsSubSync(t, nc1, "*.bar")
nct := natsConnect(t, s1.ClientURL(), nats.Name("tracer"))
defer nct.Close()
traceSub := natsSubSync(t, nct, "my.trace.subj")
// Ensure the subscription is known by the server we're connected to.
require_NoError(t, nct.Flush())
time.Sleep(100 * time.Millisecond)
for _, test := range []struct {
name string
deliverMsg bool
}{
{"just trace", false},
{"deliver msg", true},
} {
t.Run(test.name, func(t *testing.T) {
msg := nats.NewMsg("foo.bar")
msg.Header.Set(MsgTraceDest, traceSub.Subject)
if !test.deliverMsg {
msg.Header.Set(MsgTraceOnly, "true")
}
msg.Data = []byte("hello!")
err := nct.PublishMsg(msg)
require_NoError(t, err)
checkAppMsg := func(sub *nats.Subscription, expected bool) {
if expected {
appMsg := natsNexMsg(t, sub, time.Second)
require_Equal(t, string(appMsg.Data), "hello!")
}
// Check that no (more) messages are received.
if msg, err := sub.NextMsg(100 * time.Millisecond); err != nats.ErrTimeout {
t.Fatalf("Did not expect application message, got %s", msg.Data)
}
}
for _, sub := range []*nats.Subscription{sub1, sub2, sub22, sub3, sub32} {
checkAppMsg(sub, test.deliverMsg)
}
var previousHop string
check := func() {
traceMsg := natsNexMsg(t, traceSub, time.Second)
var e MsgTraceEvent
json.Unmarshal(traceMsg.Data, &e)
ingress := e.Ingress()
require_True(t, ingress != nil)
switch ingress.Kind {
case CLIENT:
require_Equal(t, e.Server.Name, s1.Name())
require_Equal(t, ingress.Account, globalAccountName)
require_Equal(t, ingress.Subject, "foo.bar")
egress := e.Egresses()
require_Equal(t, len(egress), 3)
for _, eg := range egress {
switch eg.Kind {
case CLIENT:
require_Equal(t, eg.Name, "sub1")
require_Equal(t, eg.Subscription, "*.bar")
require_Equal(t, eg.Queue, _EMPTY_)
case GATEWAY:
if eg.Name != s2.Name() && eg.Name != s3.Name() {
t.Fatalf("Expected name to be %q or %q, got %q", s2.Name(), s3.Name(), eg.Name)
}
require_Equal(t, eg.Error, _EMPTY_)
require_Equal(t, eg.Subscription, _EMPTY_)
require_Equal(t, eg.Queue, _EMPTY_)
default:
t.Fatalf("Unexpected egress: %+v", eg)
}
}
case GATEWAY:
require_True(t, e.Request.Header != nil)
require_Len(t, len(e.Request.Header[MsgTraceHop]), 1)
hop := e.Request.Header[MsgTraceHop][0]
require_True(t, hop == "1" || hop == "2")
if previousHop == _EMPTY_ {
previousHop = hop
} else if hop == previousHop {
t.Fatalf("Expected different hop value, got the same %q", hop)
}
var sub2Name, queue2Name, sub3Name, queue3Name string
switch e.Server.Name {
case s2.Name():
require_Equal(t, e.Server.Cluster, "B")
sub2Name, sub3Name = "sub2", "sub22"
queue2Name, queue3Name = "my_queue_2", "my_queue_22"
case s3.Name():
require_Equal(t, e.Server.Cluster, "C")
sub2Name, sub3Name = "sub3", "sub32"
queue2Name, queue3Name = "my_queue_3", "my_queue_32"
default:
t.Fatalf("Unexpected server name %q", e.Server.Name)
}
require_Equal(t, ingress.Account, globalAccountName)
require_Equal(t, ingress.Subject, "foo.bar")
egress := e.Egresses()
require_Equal(t, len(egress), 2)
var gotSub2, gotSub3 int
for _, eg := range egress {
require_True(t, eg.Kind == CLIENT)
switch eg.Name {
case sub2Name:
require_Equal(t, eg.Subscription, "foo.*")
require_Equal(t, eg.Queue, queue2Name)
gotSub2++
case sub3Name:
require_Equal(t, eg.Subscription, "*.*")
require_Equal(t, eg.Queue, queue3Name)
gotSub3++
default:
t.Fatalf("Unexpected egress name: %+v", eg)
}
}
require_Equal(t, gotSub2, 1)
require_Equal(t, gotSub3, 1)
default:
t.Fatalf("Unexpected ingress: %+v", ingress)
}
}
// We should get 3 events
for range 3 {
check()
}
// Make sure we are not receiving more traces
if tm, err := traceSub.NextMsg(250 * time.Millisecond); err == nil {
t.Fatalf("Should not have received trace message: %s", tm.Data)
}
})
}
}
func TestMsgTraceWithGatewayToOldServer(t *testing.T) {
msgTraceCheckSupport = true
defer func() { msgTraceCheckSupport = false }()
o2 := testDefaultOptionsForGateway("B")
o2.NoSystemAccount = false
// Make this server behave like an older server
o2.overrideProto = setServerProtoForTest(MsgTraceProto - 1)
s2 := runGatewayServer(o2)
defer s2.Shutdown()
o1 := testGatewayOptionsFromToWithServers(t, "A", "B", s2)
o1.NoSystemAccount = false
s1 := runGatewayServer(o1)
defer s1.Shutdown()
waitForOutboundGateways(t, s1, 1, time.Second)
waitForInboundGateways(t, s2, 1, time.Second)
waitForOutboundGateways(t, s2, 1, time.Second)
nc2 := natsConnect(t, s2.ClientURL(), nats.Name("sub2"))
defer nc2.Close()
sub2 := natsSubSync(t, nc2, "foo")
nc1 := natsConnect(t, s1.ClientURL(), nats.Name("sub1"))
defer nc1.Close()
sub1 := natsSubSync(t, nc1, "foo")
nct := natsConnect(t, s1.ClientURL(), nats.Name("tracer"))
defer nct.Close()
traceSub := natsSubSync(t, nct, "my.trace.subj")
// Ensure the subscription is known by the server we're connected to.
require_NoError(t, nct.Flush())
time.Sleep(100 * time.Millisecond)
for _, test := range []struct {
name string
deliverMsg bool
}{
{"just trace", false},
{"deliver msg", true},
} {
t.Run(test.name, func(t *testing.T) {
msg := nats.NewMsg("foo")
msg.Header.Set(MsgTraceDest, traceSub.Subject)
if !test.deliverMsg {
msg.Header.Set(MsgTraceOnly, "true")
}
msg.Data = []byte("hello!")
err := nct.PublishMsg(msg)
require_NoError(t, err)
checkAppMsg := func(sub *nats.Subscription, expected bool) {
if expected {
appMsg := natsNexMsg(t, sub, time.Second)
require_Equal[string](t, string(appMsg.Data), "hello!")
}
// Check that no (more) messages are received.
if msg, err := sub.NextMsg(100 * time.Millisecond); err != nats.ErrTimeout {
t.Fatalf("Did not expect application message, got %s", msg.Data)
}
}
// Even if a server does not support tracing, as long as the header
// TraceOnly is not set, the message should be forwarded to the remote.
for _, sub := range []*nats.Subscription{sub1, sub2} {
checkAppMsg(sub, test.deliverMsg)
}
traceMsg := natsNexMsg(t, traceSub, time.Second)
var e MsgTraceEvent
json.Unmarshal(traceMsg.Data, &e)
ingress := e.Ingress()
require_True(t, ingress != nil)
switch ingress.Kind {
case CLIENT:
require_Equal[string](t, e.Server.Name, s1.Name())
egress := e.Egresses()
require_Equal[int](t, len(egress), 2)
for _, ci := range egress {
switch ci.Kind {
case CLIENT:
require_Equal[string](t, ci.Name, "sub1")
case GATEWAY:
require_Equal[string](t, ci.Name, s2.Name())
if test.deliverMsg {
require_Contains(t, ci.Error, errMsgTraceNoSupport)
} else {
require_Contains(t, ci.Error, errMsgTraceOnlyNoSupport)
}
default:
t.Fatalf("Unexpected egress: %+v", ci)
}
}
default:
t.Fatalf("Unexpected ingress: %+v", ingress)
}
// We should not get a second trace
if msg, err := traceSub.NextMsg(100 * time.Millisecond); err != nats.ErrTimeout {
t.Fatalf("Did not expect other trace, got %s", msg.Data)
}
})
}
}
func TestMsgTraceServiceImport(t *testing.T) {
tmpl := `
listen: 127.0.0.1:-1
accounts {
A {
users: [{user: a, password: pwd}]
exports: [ { service: ">", allow_trace: %v} ]
mappings = {
bar: bozo
}
}
B {
users: [{user: b, password: pwd}]
imports: [ { service: {account: "A", subject:">"} } ]
exports: [ { service: ">", allow_trace: %v} ]
}
C {
users: [{user: c, password: pwd}]
exports: [ { service: ">", allow_trace: %v } ]
}
D {
users: [{user: d, password: pwd}]
imports: [
{ service: {account: "B", subject:"bar"}, to: baz }
{ service: {account: "C", subject:">"} }
]
mappings = {
bat: baz
}
}
}
`
conf := createConfFile(t, []byte(fmt.Sprintf(tmpl, false, false, false)))
s, _ := RunServerWithConfig(conf)
defer s.Shutdown()
nc := natsConnect(t, s.ClientURL(), nats.UserInfo("d", "pwd"), nats.Name("Requestor"))
defer nc.Close()
traceSub := natsSubSync(t, nc, "my.trace.subj")
sub := natsSubSync(t, nc, "my.service.response.inbox")
nc2 := natsConnect(t, s.ClientURL(), nats.UserInfo("a", "pwd"), nats.Name("ServiceA"))
defer nc2.Close()
recv := int32(0)
natsQueueSub(t, nc2, "*", "my_queue", func(m *nats.Msg) {
atomic.AddInt32(&recv, 1)
m.Respond(m.Data)
})
natsFlush(t, nc2)
nc3 := natsConnect(t, s.ClientURL(), nats.UserInfo("c", "pwd"), nats.Name("ServiceC"))
defer nc3.Close()
natsSub(t, nc3, "baz", func(m *nats.Msg) {
atomic.AddInt32(&recv, 1)
m.Respond(m.Data)
})
natsFlush(t, nc3)
for mainIter, mainTest := range []struct {
name string
allow bool
}{
{"not allowed", false},
{"allowed", true},
{"not allowed again", false},
} {
atomic.StoreInt32(&recv, 0)
t.Run(mainTest.name, func(t *testing.T) {
for _, test := range []struct {
name string
deliverMsg bool
}{
{"just trace", false},
{"deliver msg", true},
} {
t.Run(test.name, func(t *testing.T) {
msg := nats.NewMsg("bat")
msg.Header.Set(MsgTraceDest, traceSub.Subject)
if !test.deliverMsg {
msg.Header.Set(MsgTraceOnly, "true")
}
if !test.deliverMsg {
msg.Data = []byte("request1")
} else {
msg.Data = []byte("request2")
}
msg.Reply = sub.Subject
err := nc.PublishMsg(msg)
require_NoError(t, err)
if test.deliverMsg {
for i := 0; i < 2; i++ {
appMsg := natsNexMsg(t, sub, time.Second)
require_Equal[string](t, string(appMsg.Data), "request2")
}
}
// Check that no (more) messages are received.
if msg, err := sub.NextMsg(100 * time.Millisecond); msg != nil || (err != nats.ErrTimeout && err != nats.ErrNoResponders) {
t.Fatalf("Did not expect application message, got msg=%v err=%v", msg, err)
}
if !test.deliverMsg {
// Just to make sure that message was not delivered to service
// responders, wait a bit and check the recv value.
time.Sleep(50 * time.Millisecond)
if n := atomic.LoadInt32(&recv); n != 0 {
t.Fatalf("Expected no message to be received, but service callback fired %d times", n)
}
}
traceMsg := natsNexMsg(t, traceSub, time.Second)
var e MsgTraceEvent
json.Unmarshal(traceMsg.Data, &e)
require_Equal[string](t, e.Server.Name, s.Name())
ingress := e.Ingress()
require_True(t, ingress != nil)
require_True(t, ingress.Kind == CLIENT)
require_Equal[string](t, ingress.Account, "D")
require_Equal[string](t, ingress.Subject, "bat")
sm := e.SubjectMapping()
require_True(t, sm != nil)
require_Equal[string](t, sm.MappedTo, "baz")
simps := e.ServiceImports()
require_True(t, simps != nil)
var expectedServices int
if mainTest.allow {
expectedServices = 3
} else {
expectedServices = 2
}
require_Equal[int](t, len(simps), expectedServices)
for _, si := range simps {
require_True(t, si.Timestamp != time.Time{})
switch si.Account {
case "C":
require_Equal[string](t, si.From, "baz")
require_Equal[string](t, si.To, "baz")
case "B":
require_Equal[string](t, si.From, "baz")
require_Equal[string](t, si.To, "bar")
case "A":
if !mainTest.allow {
t.Fatalf("Without allow_trace, we should not see service for account A")
}
require_Equal[string](t, si.From, "bar")
require_Equal[string](t, si.To, "bozo")
default:
t.Fatalf("Unexpected account name: %s", si.Account)
}
}
egress := e.Egresses()
if !mainTest.allow {
require_Equal[int](t, len(egress), 0)
} else {
require_Equal[int](t, len(egress), 2)
var gotA, gotC bool
for _, eg := range egress {
// All Egress should be clients
require_True(t, eg.Kind == CLIENT)
// We should have one for ServiceA and one for ServiceC
if eg.Name == "ServiceA" {
require_Equal[string](t, eg.Account, "A")
require_Equal[string](t, eg.Subscription, "*")
require_Equal[string](t, eg.Queue, "my_queue")
gotA = true
} else if eg.Name == "ServiceC" {
require_Equal[string](t, eg.Account, "C")
require_Equal[string](t, eg.Queue, _EMPTY_)
gotC = true
}
}
if !gotA {
t.Fatalf("Did not get Egress for serviceA: %+v", egress)
}
if !gotC {
t.Fatalf("Did not get Egress for serviceC: %+v", egress)
}
}
// Make sure we properly remove the responses.
checkResp := func(an string) {
acc, err := s.lookupAccount(an)
require_NoError(t, err)
checkFor(t, time.Second, 15*time.Millisecond, func() error {
if n := acc.NumPendingAllResponses(); n != 0 {
return fmt.Errorf("Still %d responses pending for account %q on server %s", n, acc, s)
}
return nil
})
}
for _, acc := range []string{"A", "B", "C", "D"} {
checkResp(acc)
}
})
}
switch mainIter {
case 0:
reloadUpdateConfig(t, s, conf, fmt.Sprintf(tmpl, true, true, true))
case 1:
reloadUpdateConfig(t, s, conf, fmt.Sprintf(tmpl, false, false, false))
}
})
}
}
func TestMsgTraceServiceImportWithSuperCluster(t *testing.T) {
for _, mainTest := range []struct {
name string
allowStr string
allow bool
}{
{"allowed", "true", true},
{"not allowed", "false", false},
} {
t.Run(mainTest.name, func(t *testing.T) {
tmpl := `
listen: 127.0.0.1:-1
server_name: %s
jetstream: {max_mem_store: 256MB, max_file_store: 2GB, store_dir: '%s'}
cluster {
name: %s
listen: 127.0.0.1:%d
routes = [%s]
}
accounts {
A {
users: [{user: a, password: pwd}]
exports: [ { service: ">", allow_trace: ` + mainTest.allowStr + ` } ]
mappings = {
bar: bozo
}
trace_dest: "a.trace.subj"
}
B {
users: [{user: b, password: pwd}]
imports: [ { service: {account: "A", subject:">"} } ]
exports: [ { service: ">" , allow_trace: ` + mainTest.allowStr + ` } ]
trace_dest: "b.trace.subj"
}
C {
users: [{user: c, password: pwd}]
exports: [ { service: ">" , allow_trace: ` + mainTest.allowStr + ` } ]
trace_dest: "c.trace.subj"
}
D {
users: [{user: d, password: pwd}]
imports: [
{ service: {account: "B", subject:"bar"}, to: baz }
{ service: {account: "C", subject:">"} }
]
mappings = {
bat: baz
}
trace_dest: "d.trace.subj"
}
$SYS { users = [ { user: "admin", pass: "s3cr3t!" } ] }
}
`
sc := createJetStreamSuperClusterWithTemplate(t, tmpl, 3, 2)
defer sc.shutdown()
sfornc := sc.clusters[0].servers[0]
nc := natsConnect(t, sfornc.ClientURL(), nats.UserInfo("d", "pwd"), nats.Name("Requestor"))
defer nc.Close()
traceSub := natsSubSync(t, nc, "my.trace.subj")
sub := natsSubSync(t, nc, "my.service.response.inbox")
sfornc2 := sc.clusters[0].servers[1]
nc2 := natsConnect(t, sfornc2.ClientURL(), nats.UserInfo("a", "pwd"), nats.Name("ServiceA"))
defer nc2.Close()
subSvcA := natsQueueSubSync(t, nc2, "*", "my_queue")
natsFlush(t, nc2)
sfornc3 := sc.clusters[1].servers[0]
nc3 := natsConnect(t, sfornc3.ClientURL(), nats.UserInfo("c", "pwd"), nats.Name("ServiceC"))
defer nc3.Close()
subSvcC := natsSubSync(t, nc3, "baz")
natsFlush(t, nc3)
// Create a subscription for each account trace destination to make
// sure that we are not sending it there.
var accSubs []*nats.Subscription
for _, user := range []string{"a", "b", "c", "d"} {
nc := natsConnect(t, sfornc.ClientURL(), nats.UserInfo(user, "pwd"))
defer nc.Close()
accSubs = append(accSubs, natsSubSync(t, nc, user+".trace.subj"))
}
for _, test := range []struct {
name string
deliverMsg bool
}{
{"just trace", false},
{"deliver msg", true},
} {
t.Run(test.name, func(t *testing.T) {
msg := nats.NewMsg("bat")
msg.Header.Set(MsgTraceDest, traceSub.Subject)
if !test.deliverMsg {
msg.Header.Set(MsgTraceOnly, "true")
}
// We add the traceParentHdr header to make sure that it is
// deactivated in addition to the Nats-Trace-Dest header too
// when needed.
traceParentHdrVal := "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"
msg.Header.Set(traceParentHdr, traceParentHdrVal)
if !test.deliverMsg {
msg.Data = []byte("request1")
} else {
msg.Data = []byte("request2")
}
msg.Reply = sub.Subject
err := nc.PublishMsg(msg)
require_NoError(t, err)
if test.deliverMsg {
processSvc := func(sub *nats.Subscription) {
t.Helper()
appMsg := natsNexMsg(t, sub, time.Second)
if hv := appMsg.Header.Get(traceParentHdr); hv != traceParentHdrVal {
t.Fatalf("Expecting header with %q, but got %q", traceParentHdrVal, hv)
}
// This test causes a message to be routed to the
// service responders. When not allowing, we need
// to make sure that the trace header has been
// disabled. Not receiving the trace event from
// the remote is not enough to verify since the
// trace would not reach the origin server because
// the origin account header will not be present.
var expected string
if mainTest.allow {
expected = traceSub.Subject
} else {
expected = MsgTraceDestDisabled
}
if hv := appMsg.Header.Get(MsgTraceDest); hv != expected {
t.Fatalf("Expecting header with %q, but got %q", expected, hv)
}
appMsg.Respond(appMsg.Data)
}
processSvc(subSvcA)
processSvc(subSvcC)
for i := 0; i < 2; i++ {
appMsg := natsNexMsg(t, sub, time.Second)
require_Equal[string](t, string(appMsg.Data), "request2")
}
}
// Check that no (more) messages are received.
if msg, err := sub.NextMsg(100 * time.Millisecond); msg != nil || (err != nats.ErrTimeout && err != nats.ErrNoResponders) {
t.Fatalf("Did not expect application message, got msg=%v err=%v", msg, err)
}
if !test.deliverMsg {
// Just to make sure that message was not delivered to service
// responders, wait a bit and check the recv value.
time.Sleep(50 * time.Millisecond)
for _, sub := range []*nats.Subscription{subSvcA, subSvcC} {
if msg, err := sub.NextMsg(250 * time.Millisecond); err == nil {
t.Fatalf("Expected no message to be received, but service subscription got %s", msg.Data)
}
}
}
check := func() {
traceMsg := natsNexMsg(t, traceSub, time.Second)
var e MsgTraceEvent
json.Unmarshal(traceMsg.Data, &e)
ingress := e.Ingress()
require_True(t, ingress != nil)
switch ingress.Kind {
case CLIENT:
require_Equal[string](t, e.Server.Name, sfornc.Name())
require_Equal[string](t, ingress.Account, "D")
require_Equal[string](t, ingress.Subject, "bat")
sm := e.SubjectMapping()
require_True(t, sm != nil)
require_Equal[string](t, sm.MappedTo, "baz")
simps := e.ServiceImports()
require_True(t, simps != nil)
var expectedServices int
if mainTest.allow {
expectedServices = 3
} else {
expectedServices = 2
}
require_Equal[int](t, len(simps), expectedServices)
for _, si := range simps {
switch si.Account {
case "C":
require_Equal[string](t, si.From, "baz")
require_Equal[string](t, si.To, "baz")
case "B":
require_Equal[string](t, si.From, "baz")
require_Equal[string](t, si.To, "bar")
case "A":
if !mainTest.allow {
t.Fatalf("Without allow_trace, we should not see service for account A")
}
require_Equal[string](t, si.From, "bar")
require_Equal[string](t, si.To, "bozo")
default:
t.Fatalf("Unexpected account name: %s", si.Account)
}
}
egress := e.Egresses()
if !mainTest.allow {
require_Equal[int](t, len(egress), 0)
} else {
require_Equal[int](t, len(egress), 2)
for _, eg := range egress {
switch eg.Kind {
case ROUTER:
require_Equal[string](t, eg.Name, sfornc2.Name())
require_Equal[string](t, eg.Account, _EMPTY_)
case GATEWAY:
require_Equal[string](t, eg.Name, sfornc3.Name())
require_Equal[string](t, eg.Account, _EMPTY_)
}
}
}
case ROUTER:
require_Equal[string](t, e.Server.Name, sfornc2.Name())
require_Equal[string](t, ingress.Account, "A")
require_Equal[string](t, ingress.Subject, "bozo")
egress := e.Egresses()
require_Equal[int](t, len(egress), 1)
eg := egress[0]
require_True(t, eg.Kind == CLIENT)
require_Equal[string](t, eg.Name, "ServiceA")
require_Equal[string](t, eg.Account, _EMPTY_)
require_Equal[string](t, eg.Subscription, "*")
require_Equal[string](t, eg.Queue, "my_queue")
case GATEWAY:
require_Equal[string](t, e.Server.Name, sfornc3.Name())
require_Equal[string](t, ingress.Account, "C")
require_Equal[string](t, ingress.Subject, "baz")
egress := e.Egresses()
require_Equal[int](t, len(egress), 1)
eg := egress[0]
require_True(t, eg.Kind == CLIENT)
require_Equal[string](t, eg.Name, "ServiceC")
require_Equal[string](t, eg.Account, _EMPTY_)
require_Equal[string](t, eg.Subscription, "baz")
require_Equal[string](t, eg.Queue, _EMPTY_)
default:
t.Fatalf("Unexpected ingress: %+v", ingress)
}
}
// We should receive 3 events when allowed, a single when not.
check()
if mainTest.allow {
check()
check()
}
// Make sure we are not receiving more traces
if tm, err := traceSub.NextMsg(250 * time.Millisecond); err == nil {
t.Fatalf("Should not have received trace message: %s", tm.Data)
}
// Make sure that we never receive on any of the account
// trace destination's sub.
for _, sub := range accSubs {
if tm, err := sub.NextMsg(100 * time.Millisecond); err == nil {
t.Fatalf("Should not have received trace message on account's trace sub, got %s", tm.Data)
}
}
// Make sure we properly remove the responses.
checkResp := func(an string) {
for _, s := range []*Server{sfornc, sfornc2, sfornc3} {
acc, err := s.lookupAccount(an)
require_NoError(t, err)
checkFor(t, time.Second, 15*time.Millisecond, func() error {
if n := acc.NumPendingAllResponses(); n != 0 {
return fmt.Errorf("Still %d responses pending for account %q on server %s", n, acc, s)
}
return nil
})
}
}
for _, acc := range []string{"A", "B", "C", "D"} {
checkResp(acc)
}
})
}
})
}
}
func TestMsgTraceServiceImportWithLeafNodeHub(t *testing.T) {
confHub := createConfFile(t, []byte(`
listen: 127.0.0.1:-1
server_name: "S1"
accounts {
A {
users: [{user: a, password: pwd}]
exports: [ { service: ">", allow_trace: true } ]
mappings = {
bar: bozo
}
}
B {
users: [{user: b, password: pwd}]
imports: [ { service: {account: "A", subject:">"} } ]
exports: [ { service: ">", allow_trace: true } ]
}
C {
users: [{user: c, password: pwd}]
exports: [ { service: ">", allow_trace: true } ]
}
D {
users: [{user: d, password: pwd}]
imports: [
{ service: {account: "B", subject:"bar"}, to: baz }
{ service: {account: "C", subject:">"} }
]
mappings = {
bat: baz
}
}
$SYS { users = [ { user: "admin", pass: "s3cr3t!" } ] }
}
leafnodes {
port: -1
}
`))
hub, ohub := RunServerWithConfig(confHub)
defer hub.Shutdown()
confLeaf := createConfFile(t, []byte(fmt.Sprintf(`
listen: 127.0.0.1:-1
server_name: "S2"
leafnodes {
remotes [{url: "nats://d:[email protected]:%d"}]
}
`, ohub.LeafNode.Port)))
leaf, _ := RunServerWithConfig(confLeaf)
defer leaf.Shutdown()
checkLeafNodeConnectedCount(t, hub, 1)
checkLeafNodeConnectedCount(t, leaf, 1)
nc2 := natsConnect(t, hub.ClientURL(), nats.UserInfo("a", "pwd"), nats.Name("ServiceA"))
defer nc2.Close()
recv := int32(0)
natsQueueSub(t, nc2, "*", "my_queue", func(m *nats.Msg) {
atomic.AddInt32(&recv, 1)
m.Respond(m.Data)
})
natsFlush(t, nc2)
nc3 := natsConnect(t, hub.ClientURL(), nats.UserInfo("c", "pwd"), nats.Name("ServiceC"))
defer nc3.Close()
natsSub(t, nc3, "baz", func(m *nats.Msg) {
atomic.AddInt32(&recv, 1)
m.Respond(m.Data)
})
natsFlush(t, nc3)
nc := natsConnect(t, leaf.ClientURL(), nats.Name("Requestor"))
defer nc.Close()
traceSub := natsSubSync(t, nc, "my.trace.subj")
sub := natsSubSync(t, nc, "my.service.response.inbox")
checkSubInterest(t, leaf, globalAccountName, "bat", time.Second)
for _, test := range []struct {
name string
deliverMsg bool
}{
{"just trace", false},
{"deliver msg", true},
} {
t.Run(test.name, func(t *testing.T) {
msg := nats.NewMsg("bat")
msg.Header.Set(MsgTraceDest, traceSub.Subject)
if !test.deliverMsg {
msg.Header.Set(MsgTraceOnly, "true")
}
if !test.deliverMsg {
msg.Data = []byte("request1")
} else {
msg.Data = []byte("request2")
}
msg.Reply = sub.Subject
err := nc.PublishMsg(msg)
require_NoError(t, err)
if test.deliverMsg {
for i := 0; i < 2; i++ {
appMsg := natsNexMsg(t, sub, time.Second)
require_Equal[string](t, string(appMsg.Data), "request2")
}
}
// Check that no (more) messages are received.
if msg, err := sub.NextMsg(100 * time.Millisecond); msg != nil || err != nats.ErrTimeout {
t.Fatalf("Did not expect application message, got msg=%v err=%v", msg, err)
}
if !test.deliverMsg {
// Just to make sure that message was not delivered to service
// responders, wait a bit and check the recv value.
time.Sleep(50 * time.Millisecond)
if n := atomic.LoadInt32(&recv); n != 0 {
t.Fatalf("Expected no message to be received, but service callback fired %d times", n)
}
}
check := func() {
traceMsg := natsNexMsg(t, traceSub, time.Second)
var e MsgTraceEvent
json.Unmarshal(traceMsg.Data, &e)
ingress := e.Ingress()
require_True(t, ingress != nil)
switch ingress.Kind {
case CLIENT:
require_Equal[string](t, e.Server.Name, "S2")
require_Equal[string](t, ingress.Account, globalAccountName)
require_Equal[string](t, ingress.Subject, "bat")
require_True(t, e.SubjectMapping() == nil)
egress := e.Egresses()
require_Equal[int](t, len(egress), 1)
eg := egress[0]
require_True(t, eg.Kind == LEAF)
require_Equal[string](t, eg.Name, "S1")
require_Equal[string](t, eg.Account, _EMPTY_)
case LEAF:
require_Equal[string](t, e.Server.Name, hub.Name())
require_Equal[string](t, ingress.Name, leaf.Name())
require_Equal[string](t, ingress.Account, "D")
require_Equal[string](t, ingress.Subject, "bat")
sm := e.SubjectMapping()
require_True(t, sm != nil)
require_Equal[string](t, sm.MappedTo, "baz")
simps := e.ServiceImports()
require_True(t, simps != nil)
require_Equal[int](t, len(simps), 3)
for _, si := range simps {
switch si.Account {
case "C":
require_Equal[string](t, si.From, "baz")
require_Equal[string](t, si.To, "baz")
case "B":
require_Equal[string](t, si.From, "baz")
require_Equal[string](t, si.To, "bar")
case "A":
require_Equal[string](t, si.From, "bar")
require_Equal[string](t, si.To, "bozo")
default:
t.Fatalf("Unexpected account name: %s", si.Account)
}
}
egress := e.Egresses()
require_Equal[int](t, len(egress), 2)
for _, eg := range egress {
require_True(t, eg.Kind == CLIENT)
switch eg.Account {
case "C":
require_Equal[string](t, eg.Name, "ServiceC")
require_Equal[string](t, eg.Subscription, "baz")
require_Equal[string](t, eg.Queue, _EMPTY_)
case "A":
require_Equal[string](t, eg.Name, "ServiceA")
require_Equal[string](t, eg.Subscription, "*")
require_Equal[string](t, eg.Queue, "my_queue")
default:
t.Fatalf("Unexpected egress: %+v", eg)
}
}
default:
t.Fatalf("Unexpected ingress: %+v", ingress)
}
}
// We should receive 2 events.
check()
check()
// Make sure we are not receiving more traces
if tm, err := traceSub.NextMsg(250 * time.Millisecond); err == nil {
t.Fatalf("Should not have received trace message: %s", tm.Data)
}
// Make sure we properly remove the responses.
checkResp := func(an string) {
acc, err := hub.lookupAccount(an)
require_NoError(t, err)
checkFor(t, time.Second, 15*time.Millisecond, func() error {
if n := acc.NumPendingAllResponses(); n != 0 {
return fmt.Errorf("Still %d responses for account %q pending on %s", n, an, hub)
}
return nil
})
}
for _, acc := range []string{"A", "B", "C", "D"} {
checkResp(acc)
}
})
}
}
func TestMsgTraceServiceImportWithLeafNodeLeaf(t *testing.T) {
confHub := createConfFile(t, []byte(`
listen: 127.0.0.1:-1
server_name: "S1"
accounts {
A {
users: [{user: a, password: pwd}]
exports: [ { service: "bar", allow_trace: true } ]
}
B {
users: [{user: b, password: pwd}]
imports: [{ service: {account: "A", subject:"bar"}, to: baz }]
}
$SYS { users = [ { user: "admin", pass: "s3cr3t!" } ] }
}
leafnodes {
port: -1
}
`))
hub, ohub := RunServerWithConfig(confHub)
defer hub.Shutdown()
confLeaf := createConfFile(t, []byte(fmt.Sprintf(`
listen: 127.0.0.1:-1
server_name: "S2"
accounts {
A {
users: [{user: a, password: pwd}]
exports: [ { service: "bar"} ]
}
B { users: [{user: b, password: pwd}] }
$SYS { users = [ { user: "admin", pass: "s3cr3t!" } ] }
}
leafnodes {
remotes [
{
url: "nats://a:[email protected]:%d"
account: A
}
{
url: "nats://b:[email protected]:%d"
account: B
}
]
}
`, ohub.LeafNode.Port, ohub.LeafNode.Port)))
leaf, _ := RunServerWithConfig(confLeaf)
defer leaf.Shutdown()
checkLeafNodeConnectedCount(t, hub, 2)
checkLeafNodeConnectedCount(t, leaf, 2)
nc2 := natsConnect(t, leaf.ClientURL(), nats.UserInfo("a", "pwd"), nats.Name("ServiceA"))
defer nc2.Close()
recv := int32(0)
natsQueueSub(t, nc2, "*", "my_queue", func(m *nats.Msg) {
atomic.AddInt32(&recv, 1)
m.Respond(m.Data)
})
natsFlush(t, nc2)
nc := natsConnect(t, hub.ClientURL(), nats.UserInfo("b", "pwd"), nats.Name("Requestor"))
defer nc.Close()
traceSub := natsSubSync(t, nc, "my.trace.subj")
sub := natsSubSync(t, nc, "my.service.response.inbox")
// Check that hub has a subscription interest on "baz"
checkSubInterest(t, hub, "A", "baz", time.Second)
// And check that the leaf has the sub interest on the trace subject
checkSubInterest(t, leaf, "B", traceSub.Subject, time.Second)
for _, test := range []struct {
name string
deliverMsg bool
}{
{"just trace", false},
{"deliver msg", true},
} {
t.Run(test.name, func(t *testing.T) {
msg := nats.NewMsg("baz")
msg.Header.Set(MsgTraceDest, traceSub.Subject)
if !test.deliverMsg {
msg.Header.Set(MsgTraceOnly, "true")
}
if !test.deliverMsg {
msg.Data = []byte("request1")
} else {
msg.Data = []byte("request2")
}
msg.Reply = sub.Subject
err := nc.PublishMsg(msg)
require_NoError(t, err)
if test.deliverMsg {
appMsg := natsNexMsg(t, sub, time.Second)
require_Equal[string](t, string(appMsg.Data), "request2")
}
// Check that no (more) messages are received.
if msg, err := sub.NextMsg(100 * time.Millisecond); msg != nil || (err != nats.ErrTimeout && err != nats.ErrNoResponders) {
t.Fatalf("Did not expect application message, got msg=%v err=%v", msg, err)
}
if !test.deliverMsg {
// Just to make sure that message was not delivered to service
// responders, wait a bit and check the recv value.
time.Sleep(50 * time.Millisecond)
if n := atomic.LoadInt32(&recv); n != 0 {
t.Fatalf("Expected no message to be received, but service callback fired %d times", n)
}
}
check := func() {
traceMsg := natsNexMsg(t, traceSub, time.Second)
var e MsgTraceEvent
json.Unmarshal(traceMsg.Data, &e)
ingress := e.Ingress()
require_True(t, ingress != nil)
switch ingress.Kind {
case CLIENT:
require_Equal[string](t, e.Server.Name, "S1")
require_Equal[string](t, ingress.Name, "Requestor")
require_Equal[string](t, ingress.Account, "B")
require_Equal[string](t, ingress.Subject, "baz")
require_True(t, e.SubjectMapping() == nil)
simps := e.ServiceImports()
require_True(t, simps != nil)
require_Equal[int](t, len(simps), 1)
si := simps[0]
require_Equal[string](t, si.Account, "A")
require_Equal[string](t, si.From, "baz")
require_Equal[string](t, si.To, "bar")
egress := e.Egresses()
require_Equal[int](t, len(egress), 1)
eg := egress[0]
require_True(t, eg.Kind == LEAF)
require_Equal[string](t, eg.Name, "S2")
require_Equal[string](t, eg.Account, "A")
require_Equal[string](t, eg.Subscription, _EMPTY_)
case LEAF:
require_Equal[string](t, e.Server.Name, leaf.Name())
require_Equal[string](t, ingress.Name, hub.Name())
require_Equal[string](t, ingress.Account, "A")
require_Equal[string](t, ingress.Subject, "bar")
require_True(t, e.SubjectMapping() == nil)
require_True(t, e.ServiceImports() == nil)
egress := e.Egresses()
require_Equal[int](t, len(egress), 1)
eg := egress[0]
require_True(t, eg.Kind == CLIENT)
require_Equal[string](t, eg.Name, "ServiceA")
require_Equal[string](t, eg.Subscription, "*")
require_Equal[string](t, eg.Queue, "my_queue")
default:
t.Fatalf("Unexpected ingress: %+v", ingress)
}
}
// We should receive 2 events.
check()
check()
// Make sure we are not receiving more traces
if tm, err := traceSub.NextMsg(250 * time.Millisecond); err == nil {
t.Fatalf("Should not have received trace message: %s", tm.Data)
}
// Make sure we properly remove the responses.
checkResp := func(an string) {
acc, err := leaf.lookupAccount(an)
require_NoError(t, err)
checkFor(t, time.Second, 15*time.Millisecond, func() error {
if n := acc.NumPendingAllResponses(); n != 0 {
return fmt.Errorf("Still %d responses for account %q pending on %s", n, an, leaf)
}
return nil
})
}
for _, acc := range []string{"A", "B"} {
checkResp(acc)
}
})
}
}
func TestMsgTraceStreamExport(t *testing.T) {
tmpl := `
listen: 127.0.0.1:-1
accounts {
A {
users: [{user: a, password: pwd}]
exports: [
{ stream: "info.*.*.>"}
]
}
B {
users: [{user: b, password: pwd}]
imports: [ { stream: {account: "A", subject:"info.*.*.>"}, to: "B.info.$2.$1.>", allow_trace: %v } ]
}
C {
users: [{user: c, password: pwd}]
imports: [ { stream: {account: "A", subject:"info.*.*.>"}, to: "C.info.$1.$2.>", allow_trace: %v } ]
}
}
`
conf := createConfFile(t, []byte(fmt.Sprintf(tmpl, false, false)))
s, _ := RunServerWithConfig(conf)
defer s.Shutdown()
nc := natsConnect(t, s.ClientURL(), nats.UserInfo("a", "pwd"), nats.Name("Tracer"))
defer nc.Close()
traceSub := natsSubSync(t, nc, "my.trace.subj")
nc2 := natsConnect(t, s.ClientURL(), nats.UserInfo("b", "pwd"), nats.Name("sub1"))
defer nc2.Close()
sub1 := natsSubSync(t, nc2, "B.info.*.*.>")
natsFlush(t, nc2)
nc3 := natsConnect(t, s.ClientURL(), nats.UserInfo("c", "pwd"), nats.Name("sub2"))
defer nc3.Close()
sub2 := natsQueueSubSync(t, nc3, "C.info.>", "my_queue")
natsFlush(t, nc3)
for mainIter, mainTest := range []struct {
name string
allow bool
}{
{"not allowed", false},
{"allowed", true},
{"not allowed again", false},
} {
t.Run(mainTest.name, func(t *testing.T) {
for _, test := range []struct {
name string
deliverMsg bool
}{
{"just trace", false},
{"deliver msg", true},
} {
t.Run(test.name, func(t *testing.T) {
msg := nats.NewMsg("info.11.22.bar")
msg.Header.Set(MsgTraceDest, traceSub.Subject)
if !test.deliverMsg {
msg.Header.Set(MsgTraceOnly, "true")
}
msg.Data = []byte("hello")
err := nc.PublishMsg(msg)
require_NoError(t, err)
if test.deliverMsg {
appMsg := natsNexMsg(t, sub1, time.Second)
require_Equal[string](t, appMsg.Subject, "B.info.22.11.bar")
appMsg = natsNexMsg(t, sub2, time.Second)
require_Equal[string](t, appMsg.Subject, "C.info.11.22.bar")
}
// Check that no (more) messages are received.
for _, sub := range []*nats.Subscription{sub1, sub2} {
if msg, err := sub.NextMsg(100 * time.Millisecond); msg != nil || err != nats.ErrTimeout {
t.Fatalf("Did not expect application message, got msg=%v err=%v", msg, err)
}
}
traceMsg := natsNexMsg(t, traceSub, time.Second)
var e MsgTraceEvent
json.Unmarshal(traceMsg.Data, &e)
require_Equal[string](t, e.Server.Name, s.Name())
ingress := e.Ingress()
require_True(t, ingress != nil)
require_True(t, ingress.Kind == CLIENT)
require_Equal[string](t, ingress.Account, "A")
require_Equal[string](t, ingress.Subject, "info.11.22.bar")
require_True(t, e.SubjectMapping() == nil)
require_True(t, e.ServiceImports() == nil)
stexps := e.StreamExports()
require_True(t, stexps != nil)
require_Equal[int](t, len(stexps), 2)
for _, se := range stexps {
require_True(t, se.Timestamp != time.Time{})
switch se.Account {
case "B":
require_Equal[string](t, se.To, "B.info.22.11.bar")
case "C":
require_Equal[string](t, se.To, "C.info.11.22.bar")
default:
t.Fatalf("Unexpected stream export: %+v", se)
}
}
egress := e.Egresses()
if mainTest.allow {
require_Equal[int](t, len(egress), 2)
for _, eg := range egress {
require_True(t, eg.Kind == CLIENT)
switch eg.Account {
case "B":
require_Equal[string](t, eg.Name, "sub1")
require_Equal[string](t, eg.Subscription, "info.*.*.>")
require_Equal[string](t, eg.Queue, _EMPTY_)
case "C":
require_Equal[string](t, eg.Name, "sub2")
require_Equal[string](t, eg.Subscription, "info.*.*.>")
require_Equal[string](t, eg.Queue, "my_queue")
default:
t.Fatalf("Unexpected egress: %+v", eg)
}
}
} else {
require_Equal[int](t, len(egress), 0)
}
})
}
switch mainIter {
case 0:
reloadUpdateConfig(t, s, conf, fmt.Sprintf(tmpl, true, true))
case 1:
reloadUpdateConfig(t, s, conf, fmt.Sprintf(tmpl, false, false))
}
})
}
}
func TestMsgTraceStreamExportWithSuperCluster(t *testing.T) {
for _, mainTest := range []struct {
name string
allowStr string
allow bool
}{
{"allowed", "true", true},
{"not allowed", "false", false},
} {
t.Run(mainTest.name, func(t *testing.T) {
tmpl := `
listen: 127.0.0.1:-1
server_name: %s
jetstream: {max_mem_store: 256MB, max_file_store: 2GB, store_dir: '%s'}
cluster {
name: %s
listen: 127.0.0.1:%d
routes = [%s]
}
accounts {
A {
users: [{user: a, password: pwd}]
exports: [
{ stream: "info.*.*.>"}
]
}
B {
users: [{user: b, password: pwd}]
imports: [ { stream: {account: "A", subject:"info.*.*.>"}, to: "B.info.$2.$1.>", allow_trace: ` + mainTest.allowStr + ` } ]
}
C {
users: [{user: c, password: pwd}]
imports: [ { stream: {account: "A", subject:"info.*.*.>"}, to: "C.info.$1.$2.>", allow_trace: ` + mainTest.allowStr + ` } ]
}
$SYS { users = [ { user: "admin", pass: "s3cr3t!" } ] }
}
`
sc := createJetStreamSuperClusterWithTemplate(t, tmpl, 2, 2)
defer sc.shutdown()
sfornc := sc.clusters[0].servers[0]
nc := natsConnect(t, sfornc.ClientURL(), nats.UserInfo("a", "pwd"), nats.Name("Tracer"))
defer nc.Close()
traceSub := natsSubSync(t, nc, "my.trace.subj")
sfornc2 := sc.clusters[0].servers[1]
nc2 := natsConnect(t, sfornc2.ClientURL(), nats.UserInfo("b", "pwd"), nats.Name("sub1"))
defer nc2.Close()
sub1 := natsSubSync(t, nc2, "B.info.*.*.>")
natsFlush(t, nc2)
checkSubInterest(t, sfornc2, "A", traceSub.Subject, time.Second)
sfornc3 := sc.clusters[1].servers[0]
nc3 := natsConnect(t, sfornc3.ClientURL(), nats.UserInfo("c", "pwd"), nats.Name("sub2"))
defer nc3.Close()
sub2 := natsQueueSubSync(t, nc3, "C.info.>", "my_queue")
natsFlush(t, nc3)
checkSubInterest(t, sfornc, "A", "info.1.2.3.4", time.Second)
for _, s := range sc.clusters[0].servers {
checkForRegisteredQSubInterest(t, s, "C2", "A", "info.1.2.3", 1, time.Second)
}
for _, test := range []struct {
name string
deliverMsg bool
}{
{"just trace", false},
{"deliver msg", true},
} {
t.Run(test.name, func(t *testing.T) {
msg := nats.NewMsg("info.11.22.bar")
msg.Header.Set(MsgTraceDest, traceSub.Subject)
if !test.deliverMsg {
msg.Header.Set(MsgTraceOnly, "true")
}
msg.Data = []byte("hello")
err := nc.PublishMsg(msg)
require_NoError(t, err)
if test.deliverMsg {
appMsg := natsNexMsg(t, sub1, time.Second)
require_Equal[string](t, appMsg.Subject, "B.info.22.11.bar")
appMsg = natsNexMsg(t, sub2, time.Second)
require_Equal[string](t, appMsg.Subject, "C.info.11.22.bar")
}
// Check that no (more) messages are received.
for _, sub := range []*nats.Subscription{sub1, sub2} {
if msg, err := sub.NextMsg(100 * time.Millisecond); msg != nil || err != nats.ErrTimeout {
t.Fatalf("Did not expect application message, got msg=%v err=%v", msg, err)
}
}
check := func() {
traceMsg := natsNexMsg(t, traceSub, time.Second)
var e MsgTraceEvent
json.Unmarshal(traceMsg.Data, &e)
ingress := e.Ingress()
require_True(t, ingress != nil)
switch ingress.Kind {
case CLIENT:
require_Equal[string](t, e.Server.Name, sfornc.Name())
require_Equal[string](t, ingress.Name, "Tracer")
require_Equal[string](t, ingress.Account, "A")
require_Equal[string](t, ingress.Subject, "info.11.22.bar")
require_True(t, e.SubjectMapping() == nil)
require_True(t, e.ServiceImports() == nil)
require_True(t, e.StreamExports() == nil)
egress := e.Egresses()
require_Equal[int](t, len(egress), 2)
for _, eg := range egress {
switch eg.Kind {
case ROUTER:
require_Equal[string](t, eg.Name, sfornc2.Name())
require_Equal[string](t, eg.Account, _EMPTY_)
case GATEWAY:
require_Equal[string](t, eg.Name, sfornc3.Name())
require_Equal[string](t, eg.Account, _EMPTY_)
default:
t.Fatalf("Unexpected egress: %+v", eg)
}
}
case ROUTER:
require_Equal[string](t, e.Server.Name, sfornc2.Name())
require_Equal[string](t, ingress.Name, sfornc.Name())
require_Equal[string](t, ingress.Account, "A")
require_Equal[string](t, ingress.Subject, "info.11.22.bar")
require_True(t, e.SubjectMapping() == nil)
require_True(t, e.ServiceImports() == nil)
stexps := e.StreamExports()
require_True(t, stexps != nil)
require_Equal[int](t, len(stexps), 1)
se := stexps[0]
require_Equal[string](t, se.Account, "B")
require_Equal[string](t, se.To, "B.info.22.11.bar")
egress := e.Egresses()
if mainTest.allow {
require_Equal[int](t, len(egress), 1)
eg := egress[0]
require_True(t, eg.Kind == CLIENT)
require_Equal[string](t, eg.Name, "sub1")
require_Equal[string](t, eg.Subscription, "info.*.*.>")
require_Equal[string](t, eg.Queue, _EMPTY_)
} else {
require_Equal[int](t, len(egress), 0)
}
case GATEWAY:
require_Equal[string](t, e.Server.Name, sfornc3.Name())
require_Equal[string](t, ingress.Name, sfornc.Name())
require_Equal[string](t, ingress.Account, "A")
require_Equal[string](t, ingress.Subject, "info.11.22.bar")
require_True(t, e.SubjectMapping() == nil)
require_True(t, e.ServiceImports() == nil)
stexps := e.StreamExports()
require_True(t, stexps != nil)
require_Equal[int](t, len(stexps), 1)
se := stexps[0]
require_Equal[string](t, se.Account, "C")
require_Equal[string](t, se.To, "C.info.11.22.bar")
egress := e.Egresses()
if mainTest.allow {
require_Equal[int](t, len(egress), 1)
eg := egress[0]
require_True(t, eg.Kind == CLIENT)
require_Equal[string](t, eg.Name, "sub2")
require_Equal[string](t, eg.Subscription, "info.*.*.>")
require_Equal[string](t, eg.Queue, "my_queue")
} else {
require_Equal[int](t, len(egress), 0)
}
default:
t.Fatalf("Unexpected ingress: %+v", ingress)
}
}
// We expect 3 events
check()
check()
check()
// Make sure we are not receiving more traces
if tm, err := traceSub.NextMsg(250 * time.Millisecond); err == nil {
t.Fatalf("Should not have received trace message: %s", tm.Data)
}
})
}
})
}
}
func TestMsgTraceStreamExportWithLeafNode_Hub(t *testing.T) {
confHub := createConfFile(t, []byte(`
listen: 127.0.0.1:-1
server_name: "S1"
accounts {
A {
users: [{user: a, password: pwd}]
exports: [
{ stream: "info.*.*.>"}
]
}
B {
users: [{user: b, password: pwd}]
imports: [ { stream: {account: "A", subject:"info.*.*.>"}, to: "B.info.$2.$1.>", allow_trace: true } ]
}
C {
users: [{user: c, password: pwd}]
imports: [ { stream: {account: "A", subject:"info.*.*.>"}, to: "C.info.$1.$2.>", allow_trace: true } ]
}
}
leafnodes {
port: -1
}
`))
hub, ohub := RunServerWithConfig(confHub)
defer hub.Shutdown()
confLeaf := createConfFile(t, []byte(fmt.Sprintf(`
listen: 127.0.0.1:-1
server_name: "S2"
accounts {
LEAF { users: [{user: leaf, password: pwd}] }
}
leafnodes {
remotes [
{ url: "nats://a:[email protected]:%d", account: "LEAF" }
]
}
`, ohub.LeafNode.Port)))
leaf, _ := RunServerWithConfig(confLeaf)
defer leaf.Shutdown()
checkLeafNodeConnectedCount(t, hub, 1)
checkLeafNodeConnectedCount(t, leaf, 1)
nc := natsConnect(t, leaf.ClientURL(), nats.UserInfo("leaf", "pwd"), nats.Name("Tracer"))
defer nc.Close()
traceSub := natsSubSync(t, nc, "my.trace.subj")
checkSubInterest(t, hub, "A", traceSub.Subject, time.Second)
nc2 := natsConnect(t, hub.ClientURL(), nats.UserInfo("b", "pwd"), nats.Name("sub1"))
defer nc2.Close()
sub1 := natsSubSync(t, nc2, "B.info.*.*.>")
natsFlush(t, nc2)
nc3 := natsConnect(t, hub.ClientURL(), nats.UserInfo("c", "pwd"), nats.Name("sub2"))
defer nc3.Close()
sub2 := natsQueueSubSync(t, nc3, "C.info.>", "my_queue")
natsFlush(t, nc3)
acc, err := leaf.LookupAccount("LEAF")
require_NoError(t, err)
checkFor(t, time.Second, 50*time.Millisecond, func() error {
acc.mu.RLock()
sl := acc.sl
acc.mu.RUnlock()
r := sl.Match("info.1.2.3")
ok := len(r.psubs) > 0
if ok && (len(r.qsubs) == 0 || len(r.qsubs[0]) == 0) {
ok = false
}
if !ok {
return fmt.Errorf("Subscription interest not yet propagated")
}
return nil
})
for _, test := range []struct {
name string
deliverMsg bool
}{
{"just trace", false},
{"deliver msg", true},
} {
t.Run(test.name, func(t *testing.T) {
msg := nats.NewMsg("info.11.22.bar")
msg.Header.Set(MsgTraceDest, traceSub.Subject)
if !test.deliverMsg {
msg.Header.Set(MsgTraceOnly, "true")
}
msg.Data = []byte("hello")
err := nc.PublishMsg(msg)
require_NoError(t, err)
if test.deliverMsg {
appMsg := natsNexMsg(t, sub1, time.Second)
require_Equal[string](t, appMsg.Subject, "B.info.22.11.bar")
appMsg = natsNexMsg(t, sub2, time.Second)
require_Equal[string](t, appMsg.Subject, "C.info.11.22.bar")
}
// Check that no (more) messages are received.
for _, sub := range []*nats.Subscription{sub1, sub2} {
if msg, err := sub.NextMsg(100 * time.Millisecond); msg != nil || err != nats.ErrTimeout {
t.Fatalf("Did not expect application message, got msg=%v err=%v", msg, err)
}
}
check := func() {
traceMsg := natsNexMsg(t, traceSub, time.Second)
var e MsgTraceEvent
json.Unmarshal(traceMsg.Data, &e)
ingress := e.Ingress()
require_True(t, ingress != nil)
switch ingress.Kind {
case CLIENT:
require_Equal[string](t, e.Server.Name, leaf.Name())
require_Equal[string](t, ingress.Name, "Tracer")
require_Equal[string](t, ingress.Account, "LEAF")
require_Equal[string](t, ingress.Subject, "info.11.22.bar")
require_True(t, e.SubjectMapping() == nil)
require_True(t, e.ServiceImports() == nil)
require_True(t, e.StreamExports() == nil)
egress := e.Egresses()
require_Equal[int](t, len(egress), 1)
eg := egress[0]
require_True(t, eg.Kind == LEAF)
require_Equal[string](t, eg.Name, hub.Name())
require_Equal[string](t, eg.Account, _EMPTY_)
require_Equal[string](t, eg.Subscription, _EMPTY_)
require_Equal[string](t, eg.Queue, _EMPTY_)
case LEAF:
require_Equal[string](t, e.Server.Name, hub.Name())
require_Equal[string](t, ingress.Name, leaf.Name())
require_Equal[string](t, ingress.Account, "A")
require_Equal[string](t, ingress.Subject, "info.11.22.bar")
require_True(t, e.SubjectMapping() == nil)
require_True(t, e.ServiceImports() == nil)
stexps := e.StreamExports()
require_True(t, stexps != nil)
require_Equal[int](t, len(stexps), 2)
for _, se := range stexps {
switch se.Account {
case "B":
require_Equal[string](t, se.To, "B.info.22.11.bar")
case "C":
require_Equal[string](t, se.To, "C.info.11.22.bar")
default:
t.Fatalf("Unexpected stream export: %+v", se)
}
}
egress := e.Egresses()
require_Equal[int](t, len(egress), 2)
for _, eg := range egress {
require_True(t, eg.Kind == CLIENT)
switch eg.Account {
case "B":
require_Equal[string](t, eg.Name, "sub1")
require_Equal[string](t, eg.Subscription, "info.*.*.>")
require_Equal[string](t, eg.Queue, _EMPTY_)
case "C":
require_Equal[string](t, eg.Name, "sub2")
require_Equal[string](t, eg.Subscription, "info.*.*.>")
require_Equal[string](t, eg.Queue, "my_queue")
default:
t.Fatalf("Unexpected egress: %+v", eg)
}
}
default:
t.Fatalf("Unexpected ingress: %+v", ingress)
}
}
// We expect 2 events
check()
check()
// Make sure we are not receiving more traces
if tm, err := traceSub.NextMsg(250 * time.Millisecond); err == nil {
t.Fatalf("Should not have received trace message: %s", tm.Data)
}
})
}
}
func TestMsgTraceStreamExportWithLeafNode_Leaf(t *testing.T) {
confHub := createConfFile(t, []byte(`
listen: 127.0.0.1:-1
server_name: "S1"
accounts {
HUB { users: [{user: hub, password: pwd}] }
}
leafnodes {
port: -1
}
`))
hub, ohub := RunServerWithConfig(confHub)
defer hub.Shutdown()
confLeaf := createConfFile(t, []byte(fmt.Sprintf(`
listen: 127.0.0.1:-1
server_name: "S2"
accounts {
A {
users: [{user: a, password: pwd}]
exports: [
{ stream: "info.*.*.>"}
]
}
B {
users: [{user: b, password: pwd}]
imports: [ { stream: {account: "A", subject:"info.*.*.>"}, to: "B.info.$2.$1.>", allow_trace: true } ]
}
C {
users: [{user: c, password: pwd}]
imports: [ { stream: {account: "A", subject:"info.*.*.>"}, to: "C.info.$1.$2.>", allow_trace: true } ]
}
}
leafnodes {
remotes [
{ url: "nats://hub:[email protected]:%d", account: "A" }
]
}
`, ohub.LeafNode.Port)))
leaf, _ := RunServerWithConfig(confLeaf)
defer leaf.Shutdown()
checkLeafNodeConnectedCount(t, hub, 1)
checkLeafNodeConnectedCount(t, leaf, 1)
nc := natsConnect(t, hub.ClientURL(), nats.UserInfo("hub", "pwd"), nats.Name("Tracer"))
defer nc.Close()
traceSub := natsSubSync(t, nc, "my.trace.subj")
checkSubInterest(t, leaf, "A", traceSub.Subject, time.Second)
nc2 := natsConnect(t, leaf.ClientURL(), nats.UserInfo("b", "pwd"), nats.Name("sub1"))
defer nc2.Close()
sub1 := natsSubSync(t, nc2, "B.info.*.*.>")
natsFlush(t, nc2)
nc3 := natsConnect(t, leaf.ClientURL(), nats.UserInfo("c", "pwd"), nats.Name("sub2"))
defer nc3.Close()
sub2 := natsQueueSubSync(t, nc3, "C.info.>", "my_queue")
natsFlush(t, nc3)
acc, err := hub.LookupAccount("HUB")
require_NoError(t, err)
checkFor(t, time.Second, 50*time.Millisecond, func() error {
acc.mu.RLock()
sl := acc.sl
acc.mu.RUnlock()
r := sl.Match("info.1.2.3")
ok := len(r.psubs) > 0
if ok && (len(r.qsubs) == 0 || len(r.qsubs[0]) == 0) {
ok = false
}
if !ok {
return fmt.Errorf("Subscription interest not yet propagated")
}
return nil
})
for _, test := range []struct {
name string
deliverMsg bool
}{
{"just trace", false},
{"deliver msg", true},
} {
t.Run(test.name, func(t *testing.T) {
msg := nats.NewMsg("info.11.22.bar")
msg.Header.Set(MsgTraceDest, traceSub.Subject)
if !test.deliverMsg {
msg.Header.Set(MsgTraceOnly, "true")
}
msg.Data = []byte("hello")
err := nc.PublishMsg(msg)
require_NoError(t, err)
if test.deliverMsg {
appMsg := natsNexMsg(t, sub1, time.Second)
require_Equal[string](t, appMsg.Subject, "B.info.22.11.bar")
appMsg = natsNexMsg(t, sub2, time.Second)
require_Equal[string](t, appMsg.Subject, "C.info.11.22.bar")
}
// Check that no (more) messages are received.
for _, sub := range []*nats.Subscription{sub1, sub2} {
if msg, err := sub.NextMsg(100 * time.Millisecond); msg != nil || err != nats.ErrTimeout {
t.Fatalf("Did not expect application message, got msg=%v err=%v", msg, err)
}
}
check := func() {
traceMsg := natsNexMsg(t, traceSub, time.Second)
var e MsgTraceEvent
json.Unmarshal(traceMsg.Data, &e)
ingress := e.Ingress()
require_True(t, ingress != nil)
switch ingress.Kind {
case CLIENT:
require_Equal[string](t, e.Server.Name, hub.Name())
require_Equal[string](t, ingress.Name, "Tracer")
require_Equal[string](t, ingress.Account, "HUB")
require_Equal[string](t, ingress.Subject, "info.11.22.bar")
require_True(t, e.SubjectMapping() == nil)
require_True(t, e.ServiceImports() == nil)
require_True(t, e.StreamExports() == nil)
egress := e.Egresses()
require_Equal[int](t, len(egress), 1)
eg := egress[0]
require_True(t, eg.Kind == LEAF)
require_Equal[string](t, eg.Name, leaf.Name())
require_Equal[string](t, eg.Account, _EMPTY_)
require_Equal[string](t, eg.Subscription, _EMPTY_)
require_Equal[string](t, eg.Queue, _EMPTY_)
case LEAF:
require_Equal[string](t, e.Server.Name, leaf.Name())
require_Equal[string](t, ingress.Name, hub.Name())
require_Equal[string](t, ingress.Account, "A")
require_Equal[string](t, ingress.Subject, "info.11.22.bar")
require_True(t, e.SubjectMapping() == nil)
require_True(t, e.ServiceImports() == nil)
stexps := e.StreamExports()
require_True(t, stexps != nil)
require_Equal[int](t, len(stexps), 2)
for _, se := range stexps {
switch se.Account {
case "B":
require_Equal[string](t, se.To, "B.info.22.11.bar")
case "C":
require_Equal[string](t, se.To, "C.info.11.22.bar")
default:
t.Fatalf("Unexpected stream export: %+v", se)
}
}
egress := e.Egresses()
require_Equal[int](t, len(egress), 2)
for _, eg := range egress {
require_True(t, eg.Kind == CLIENT)
switch eg.Account {
case "B":
require_Equal[string](t, eg.Name, "sub1")
require_Equal[string](t, eg.Subscription, "info.*.*.>")
require_Equal[string](t, eg.Queue, _EMPTY_)
case "C":
require_Equal[string](t, eg.Name, "sub2")
require_Equal[string](t, eg.Subscription, "info.*.*.>")
require_Equal[string](t, eg.Queue, "my_queue")
default:
t.Fatalf("Unexpected egress: %+v", eg)
}
}
default:
t.Fatalf("Unexpected ingress: %+v", ingress)
}
}
// We expect 2 events
check()
check()
// Make sure we are not receiving more traces
if tm, err := traceSub.NextMsg(250 * time.Millisecond); err == nil {
t.Fatalf("Should not have received trace message: %s", tm.Data)
}
})
}
}
func TestMsgTraceJetStream(t *testing.T) {
opts := DefaultTestOptions
opts.Port = -1
opts.JetStream = true
opts.JetStreamMaxMemory = 270
opts.StoreDir = t.TempDir()
s := RunServer(&opts)
defer s.Shutdown()
nc, js := jsClientConnect(t, s)
defer nc.Close()
cfg := &nats.StreamConfig{
Name: "TEST",
Storage: nats.MemoryStorage,
Subjects: []string{"foo"},
Replicas: 1,
AllowRollup: true,
SubjectTransform: &nats.SubjectTransformConfig{
Source: "foo",
Destination: "bar",
},
}
_, err := js.AddStream(cfg)
require_NoError(t, err)
nct := natsConnect(t, s.ClientURL(), nats.Name("Tracer"))
defer nct.Close()
traceSub := natsSubSync(t, nct, "my.trace.subj")
natsFlush(t, nct)
msg := nats.NewMsg("foo")
msg.Header.Set(JSMsgId, "MyId")
msg.Data = make([]byte, 50)
_, err = js.PublishMsg(msg)
require_NoError(t, err)
checkStream := func(t *testing.T, expected int) {
t.Helper()
checkFor(t, time.Second, 15*time.Millisecond, func() error {
si, err := js.StreamInfo("TEST")
if err != nil {
return err
}
if n := si.State.Msgs; int(n) != expected {
return fmt.Errorf("Expected %d messages, got %v", expected, n)
}
return nil
})
}
checkStream(t, 1)
for _, test := range []struct {
name string
deliverMsg bool
}{
{"just trace", false},
{"deliver msg", true},
} {
t.Run(test.name, func(t *testing.T) {
msg := nats.NewMsg("foo")
msg.Header.Set(MsgTraceDest, traceSub.Subject)
if !test.deliverMsg {
msg.Header.Set(MsgTraceOnly, "true")
}
msg.Data = make([]byte, 50)
err = nct.PublishMsg(msg)
require_NoError(t, err)
// Wait a bit and check if message should be in the stream or not.
time.Sleep(50 * time.Millisecond)
if test.deliverMsg {
checkStream(t, 2)
} else {
checkStream(t, 1)
}
traceMsg := natsNexMsg(t, traceSub, time.Second)
var e MsgTraceEvent
json.Unmarshal(traceMsg.Data, &e)
require_Equal[string](t, e.Server.Name, s.Name())
ingress := e.Ingress()
require_True(t, ingress != nil)
require_True(t, ingress.Kind == CLIENT)
require_Equal[string](t, ingress.Name, "Tracer")
require_Equal[int](t, len(e.Egresses()), 0)
js := e.JetStream()
require_True(t, js != nil)
require_True(t, js.Timestamp != time.Time{})
require_Equal[string](t, js.Stream, "TEST")
require_Equal[string](t, js.Subject, "bar")
require_False(t, js.NoInterest)
require_Equal[string](t, js.Error, _EMPTY_)
})
}
jst, err := nct.JetStream()
require_NoError(t, err)
mset, err := s.globalAccount().lookupStream("TEST")
require_NoError(t, err)
// Now we will not ask for delivery and use headers that will fail checks
// and make sure that message is not added, that the stream's clfs is not
// increased, and that the JS trace shows the error.
newMsg := func() *nats.Msg {
msg = nats.NewMsg("foo")
msg.Header.Set(MsgTraceDest, traceSub.Subject)
msg.Header.Set(MsgTraceOnly, "true")
msg.Data = []byte("hello")
return msg
}
msgCount := 2
for _, test := range []struct {
name string
headerName string
headerVal string
expectedErr string
special int
}{
{"unexpected stream name", JSExpectedStream, "WRONG", "expected stream does not match", 0},
{"duplicate id", JSMsgId, "MyId", "duplicate", 0},
{"last seq by subject mismatch", JSExpectedLastSubjSeq, "10", "last sequence by subject mismatch", 0},
{"last seq mismatch", JSExpectedLastSeq, "10", "last sequence mismatch", 0},
{"last msgid mismatch", JSExpectedLastMsgId, "MyId3", "last msgid mismatch", 0},
{"invalid rollup command", JSMsgRollup, "wrong", "rollup value invalid: \"wrong\"", 0},
{"rollup not permitted", JSMsgRollup, JSMsgRollupSubject, "rollup not permitted", 1},
{"max msg size", _EMPTY_, _EMPTY_, ErrMaxPayload.Error(), 2},
{"normal message ok", _EMPTY_, _EMPTY_, _EMPTY_, 3},
{"insufficient resources", _EMPTY_, _EMPTY_, NewJSInsufficientResourcesError().Error(), 0},
{"stream sealed", _EMPTY_, _EMPTY_, NewJSStreamSealedError().Error(), 4},
} {
t.Run(test.name, func(t *testing.T) {
msg = newMsg()
if test.headerName != _EMPTY_ {
msg.Header.Set(test.headerName, test.headerVal)
}
switch test.special {
case 1:
// Update stream to prevent rollups, and set a max size.
cfg.AllowRollup = false
cfg.MaxMsgSize = 100
_, err = js.UpdateStream(cfg)
require_NoError(t, err)
case 2:
msg.Data = make([]byte, 200)
case 3:
pa, err := jst.Publish("foo", make([]byte, 100))
require_NoError(t, err)
msgCount++
checkStream(t, msgCount)
require_Equal[uint64](t, pa.Sequence, 3)
return
case 4:
cfg.Sealed = true
_, err = js.UpdateStream(cfg)
require_NoError(t, err)
default:
}
jst.PublishMsg(msg)
// Message count should not have increased and stay at 2.
checkStream(t, msgCount)
// Check that clfs does not increase
mset.mu.RLock()
clfs := mset.getCLFS()
mset.mu.RUnlock()
if clfs != 0 {
t.Fatalf("Stream's clfs was expected to be 0, is %d", clfs)
}
traceMsg := natsNexMsg(t, traceSub, time.Second)
var e MsgTraceEvent
json.Unmarshal(traceMsg.Data, &e)
require_Equal[string](t, e.Server.Name, s.Name())
ingress := e.Ingress()
require_True(t, ingress != nil)
require_True(t, ingress.Kind == CLIENT)
require_Equal[string](t, ingress.Name, "Tracer")
require_Equal[int](t, len(e.Egresses()), 0)
js := e.JetStream()
require_True(t, js != nil)
require_Equal[string](t, js.Stream, "TEST")
require_Equal[string](t, js.Subject, _EMPTY_)
require_False(t, js.NoInterest)
if et := js.Error; !strings.Contains(et, test.expectedErr) {
t.Fatalf("Expected JS error to contain %q, got %q", test.expectedErr, et)
}
})
}
// Create a stream with interest retention policy
_, err = js.AddStream(&nats.StreamConfig{
Name: "NO_INTEREST",
Subjects: []string{"baz"},
Retention: nats.InterestPolicy,
})
require_NoError(t, err)
msg = nats.NewMsg("baz")
msg.Header.Set(MsgTraceDest, traceSub.Subject)
msg.Header.Set(MsgTraceOnly, "true")
msg.Data = []byte("hello")
err = nct.PublishMsg(msg)
require_NoError(t, err)
traceMsg := natsNexMsg(t, traceSub, time.Second)
var e MsgTraceEvent
json.Unmarshal(traceMsg.Data, &e)
require_Equal[string](t, e.Server.Name, s.Name())
ingress := e.Ingress()
require_True(t, ingress != nil)
require_True(t, ingress.Kind == CLIENT)
require_Equal[string](t, ingress.Name, "Tracer")
require_Equal[int](t, len(e.Egresses()), 0)
ejs := e.JetStream()
require_True(t, js != nil)
require_Equal[string](t, ejs.Stream, "NO_INTEREST")
require_Equal[string](t, ejs.Subject, "baz")
require_True(t, ejs.NoInterest)
require_Equal[string](t, ejs.Error, _EMPTY_)
// Create a new stream to just check that when consumer a JS message,
// the trace header has been disabled.
cfg = &nats.StreamConfig{
Name: "TEST_TRACE_DISABLED",
Subjects: []string{"disabled"},
Retention: nats.LimitsPolicy,
}
_, err = js.AddStream(cfg)
require_NoError(t, err)
sub, err := js.PullSubscribe("disabled", "dur")
require_NoError(t, err)
msg = nats.NewMsg("disabled")
msg.Header.Set(MsgTraceDest, traceSub.Subject)
msg.Data = []byte("hello")
err = nct.PublishMsg(msg)
require_NoError(t, err)
// Check trace msg is still OK
traceMsg = natsNexMsg(t, traceSub, time.Second)
e = MsgTraceEvent{}
json.Unmarshal(traceMsg.Data, &e)
require_Equal[string](t, e.Server.Name, s.Name())
ingress = e.Ingress()
require_True(t, ingress != nil)
require_True(t, ingress.Kind == CLIENT)
require_Equal[string](t, ingress.Name, "Tracer")
require_Equal[int](t, len(e.Egresses()), 0)
ejs = e.JetStream()
require_True(t, js != nil)
require_Equal[string](t, ejs.Stream, "TEST_TRACE_DISABLED")
require_Equal[string](t, ejs.Subject, "disabled")
require_False(t, ejs.NoInterest)
require_Equal[string](t, ejs.Error, _EMPTY_)
// Now consume the message.
jmsgs, err := sub.Fetch(1)
require_NoError(t, err)
require_Len(t, len(jmsgs), 1)
jmsg := jmsgs[0]
require_True(t, len(jmsg.Header) > 0)
require_Equal(t, string(jmsg.Data), "hello")
// When consuming, the message tracing should have been disabled,
// so we should have the MsgTraceDest header set to MsgTraceDestDisabled
require_Equal(t, jmsg.Header.Get(MsgTraceDest), MsgTraceDestDisabled)
// Verify that no trace message was generated.
_, err = traceSub.NextMsg(100 * time.Millisecond)
require_Error(t, err, nats.ErrTimeout)
}
func TestMsgTraceJetStreamWithSuperCluster(t *testing.T) {
sc := createJetStreamSuperCluster(t, 3, 2)
defer sc.shutdown()
traceDest := "my.trace.subj"
// Hack to set the trace destination for the global account in order
// to make sure that the traceParentHdr header is disabled when a message
// is stored in JetStream, which will prevent emitting a trace
// when such message is retrieved and traverses a route.
// Without the account destination set, the trace would not be
// triggered, but that does not mean that we would have been
// doing the right thing of disabling the header.
for _, cl := range sc.clusters {
for _, s := range cl.servers {
acc, err := s.LookupAccount(globalAccountName)
require_NoError(t, err)
acc.setTraceDest(traceDest)
}
}
c1 := sc.clusters[0]
c2 := sc.clusters[1]
nc, js := jsClientConnect(t, c1.randomServer())
defer nc.Close()
checkStream := func(t *testing.T, stream string, expected int) {
t.Helper()
checkFor(t, 5*time.Second, 15*time.Millisecond, func() error {
si, err := js.StreamInfo(stream)
if err != nil {
return err
}
if n := si.State.Msgs; int(n) != expected {
return fmt.Errorf("Expected %d messages for stream %q, got %v", expected, stream, n)
}
return nil
})
}
payload := make([]byte, 50)
for mainIter, mainTest := range []struct {
name string
stream string
}{
{"from stream leader", "TEST1"},
{"from non stream leader", "TEST2"},
{"from other cluster", "TEST3"},
} {
t.Run(mainTest.name, func(t *testing.T) {
cfg := &nats.StreamConfig{
Name: mainTest.stream,
Replicas: 3,
AllowRollup: true,
}
_, err := js.AddStream(cfg)
require_NoError(t, err)
sc.waitOnStreamLeader(globalAccountName, mainTest.stream)
// The streams are created from c1 cluster.
slSrv := c1.streamLeader(globalAccountName, mainTest.stream)
// Store some messages
for i := 0; i < 5; i++ {
_, err = js.Publish(mainTest.stream, payload)
require_NoError(t, err)
}
// We will connect the app that sends the trace message to a server
// that is either the stream leader, a random server in c1, or
// a server in c2 (to go through a GW).
var s *Server
switch mainIter {
case 0:
s = slSrv
case 1:
s = c1.randomNonStreamLeader(globalAccountName, mainTest.stream)
case 2:
s = c2.randomServer()
}
nct := natsConnect(t, s.ClientURL(), nats.Name("Tracer"))
defer nct.Close()
traceSub := natsSubSync(t, nct, traceDest)
natsFlush(t, nct)
for _, test := range []struct {
name string
deliverMsg bool
}{
{"just trace", false},
{"deliver msg", true},
} {
t.Run(test.name, func(t *testing.T) {
msg := nats.NewMsg(mainTest.stream)
msg.Header.Set(MsgTraceDest, traceSub.Subject)
if !test.deliverMsg {
msg.Header.Set(MsgTraceOnly, "true")
}
// We add the traceParentHdr header to make sure that it
// is deactivated in addition to the Nats-Trace-Dest
// header too when needed.
msg.Header.Set(traceParentHdr, "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01")
msg.Header.Set(JSMsgId, "MyId")
msg.Data = payload
err = nct.PublishMsg(msg)
require_NoError(t, err)
if test.deliverMsg {
checkStream(t, mainTest.stream, 6)
} else {
checkStream(t, mainTest.stream, 5)
}
var (
clientOK bool
gatewayOK bool
routeOK bool
)
check := func() {
traceMsg := natsNexMsg(t, traceSub, time.Second)
var e MsgTraceEvent
json.Unmarshal(traceMsg.Data, &e)
checkJS := func() {
t.Helper()
js := e.JetStream()
require_True(t, js != nil)
require_Equal[string](t, js.Stream, mainTest.stream)
require_Equal[string](t, js.Subject, mainTest.stream)
require_False(t, js.NoInterest)
require_Equal[string](t, js.Error, _EMPTY_)
}
ingress := e.Ingress()
require_True(t, ingress != nil)
switch mainIter {
case 0:
require_Equal[string](t, e.Server.Name, s.Name())
require_True(t, ingress.Kind == CLIENT)
require_Equal[string](t, ingress.Name, "Tracer")
require_Equal[int](t, len(e.Egresses()), 0)
checkJS()
case 1:
switch ingress.Kind {
case CLIENT:
require_Equal[string](t, e.Server.Name, s.Name())
require_Equal[string](t, ingress.Name, "Tracer")
egress := e.Egresses()
require_Equal[int](t, len(egress), 1)
ci := egress[0]
require_True(t, ci.Kind == ROUTER)
require_Equal[string](t, ci.Name, slSrv.Name())
case ROUTER:
require_Equal[string](t, e.Server.Name, slSrv.Name())
require_Equal[int](t, len(e.Egresses()), 0)
checkJS()
default:
t.Fatalf("Unexpected ingress: %+v", ingress)
}
case 2:
switch ingress.Kind {
case CLIENT:
require_Equal[string](t, e.Server.Name, s.Name())
require_Equal[string](t, ingress.Name, "Tracer")
egress := e.Egresses()
require_Equal[int](t, len(egress), 1)
ci := egress[0]
require_True(t, ci.Kind == GATEWAY)
// It could have gone to any server in the C1 cluster.
// If it is not the stream leader, it should be
// routed to it.
clientOK = true
case GATEWAY:
require_Equal[string](t, ingress.Name, s.Name())
// If the server that emitted this event is the
// stream leader, then we should have the stream,
// otherwise, it should be routed.
if e.Server.Name == slSrv.Name() {
require_Equal[int](t, len(e.Egresses()), 0)
checkJS()
// Set this so that we know that we don't expect
// to have the route receive it.
routeOK = true
} else {
egress := e.Egresses()
require_Equal[int](t, len(egress), 1)
ci := egress[0]
require_True(t, ci.Kind == ROUTER)
require_Equal[string](t, ci.Name, slSrv.Name())
}
gatewayOK = true
case ROUTER:
require_Equal[string](t, e.Server.Name, slSrv.Name())
require_Equal[int](t, len(e.Egresses()), 0)
checkJS()
routeOK = true
default:
t.Fatalf("Unexpected ingress: %+v", ingress)
}
}
}
check()
if mainIter > 0 {
// There will be at least 2 events
check()
// For the last test, there may be a 3rd.
if mainIter == 2 && !(clientOK && gatewayOK && routeOK) {
check()
}
}
// Make sure we are not receiving more traces
if tm, err := traceSub.NextMsg(250 * time.Millisecond); err == nil {
t.Fatalf("Should not have received trace message: %s", tm.Data)
}
})
}
jst, err := nct.JetStream()
require_NoError(t, err)
newMsg := func() *nats.Msg {
msg := nats.NewMsg(mainTest.stream)
msg.Header.Set(MsgTraceDest, traceSub.Subject)
msg.Header.Set(MsgTraceOnly, "true")
msg.Header.Set(traceParentHdr, "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01")
msg.Data = []byte("hello")
return msg
}
msgCount := 6
for _, subtest := range []struct {
name string
headerName string
headerVal string
expectedErr string
special int
}{
{"unexpected stream name", JSExpectedStream, "WRONG", "expected stream does not match", 0},
{"duplicate id", JSMsgId, "MyId", "duplicate", 0},
{"last seq by subject mismatch", JSExpectedLastSubjSeq, "3", "last sequence by subject mismatch", 0},
{"last seq mismatch", JSExpectedLastSeq, "10", "last sequence mismatch", 0},
{"last msgid mismatch", JSExpectedLastMsgId, "MyId3", "last msgid mismatch", 0},
{"invalid rollup command", JSMsgRollup, "wrong", "rollup value invalid: \"wrong\"", 0},
{"rollup not permitted", JSMsgRollup, JSMsgRollupSubject, "rollup not permitted", 1},
{"max msg size", _EMPTY_, _EMPTY_, ErrMaxPayload.Error(), 2},
{"new message ok", _EMPTY_, _EMPTY_, _EMPTY_, 3},
{"stream sealed", _EMPTY_, _EMPTY_, NewJSStreamSealedError().Error(), 4},
} {
t.Run(subtest.name, func(t *testing.T) {
msg := newMsg()
if subtest.headerName != _EMPTY_ {
msg.Header.Set(subtest.headerName, subtest.headerVal)
}
switch subtest.special {
case 1:
// Update stream to prevent rollups, and set a max size.
cfg.AllowRollup = false
cfg.MaxMsgSize = 100
_, err = js.UpdateStream(cfg)
require_NoError(t, err)
case 2:
msg.Data = make([]byte, 200)
case 3:
pa, err := jst.Publish(mainTest.stream, []byte("hello"))
require_NoError(t, err)
msgCount++
checkStream(t, mainTest.stream, msgCount)
require_Equal[uint64](t, pa.Sequence, 7)
return
case 4:
cfg.Sealed = true
_, err = js.UpdateStream(cfg)
require_NoError(t, err)
default:
}
jst.PublishMsg(msg)
checkStream(t, mainTest.stream, msgCount)
var (
clientOK bool
gatewayOK bool
routeOK bool
)
checkJSTrace := func() {
traceMsg := natsNexMsg(t, traceSub, time.Second)
var e MsgTraceEvent
json.Unmarshal(traceMsg.Data, &e)
checkJS := func() {
t.Helper()
js := e.JetStream()
require_True(t, e.JetStream() != nil)
require_Equal[string](t, js.Stream, mainTest.stream)
require_Equal[string](t, js.Subject, _EMPTY_)
require_False(t, js.NoInterest)
if et := js.Error; !strings.Contains(et, subtest.expectedErr) {
t.Fatalf("Expected JS error to contain %q, got %q", subtest.expectedErr, et)
}
}
ingress := e.Ingress()
require_True(t, ingress != nil)
// We will focus only on the trace message that
// includes the JetStream event.
switch mainIter {
case 0:
require_Equal[string](t, e.Server.Name, s.Name())
require_True(t, ingress.Kind == CLIENT)
require_Equal[string](t, ingress.Name, "Tracer")
require_Equal[int](t, len(e.Egresses()), 0)
checkJS()
case 1:
if ingress.Kind == ROUTER {
require_Equal[string](t, e.Server.Name, slSrv.Name())
require_Equal[int](t, len(e.Egresses()), 0)
require_True(t, e.JetStream() != nil)
checkJS()
}
case 2:
switch ingress.Kind {
case CLIENT:
clientOK = true
case GATEWAY:
require_Equal[string](t, ingress.Name, s.Name())
// If the server that emitted this event is the
// stream leader, then we should have the stream,
// otherwise, it should be routed.
if e.Server.Name == slSrv.Name() {
require_Equal[int](t, len(e.Egresses()), 0)
checkJS()
// We don't expect the route event
routeOK = true
}
gatewayOK = true
case ROUTER:
require_Equal[string](t, e.Server.Name, slSrv.Name())
require_Equal[int](t, len(e.Egresses()), 0)
checkJS()
routeOK = true
}
}
}
checkJSTrace()
if mainIter > 0 {
// There will be at least 2 events
checkJSTrace()
// For the last test, there may be a 3rd.
if mainIter == 2 && !(clientOK && gatewayOK && routeOK) {
checkJSTrace()
}
}
})
}
})
}
// Now cause a step-down, and verify count is as expected.
for _, stream := range []string{"TEST1", "TEST2", "TEST3"} {
_, err := nc.Request(fmt.Sprintf(JSApiStreamLeaderStepDownT, stream), nil, time.Second)
require_NoError(t, err)
sc.waitOnStreamLeader(globalAccountName, stream)
checkStream(t, stream, 7)
}
s := c1.randomNonStreamLeader(globalAccountName, "TEST1")
// Try to get a message that will come from a route and make sure that
// this does not trigger a trace message.
nct := natsConnect(t, s.ClientURL(), nats.Name("Tracer"))
defer nct.Close()
traceSub := natsSubSync(t, nct, traceDest)
natsFlush(t, nct)
jct, err := nct.JetStream()
require_NoError(t, err)
sub, err := jct.SubscribeSync("TEST1")
require_NoError(t, err)
for i := range 7 {
jmsg, err := sub.NextMsg(time.Second)
require_NoError(t, err)
if i < 5 {
require_Len(t, len(jmsg.Header), 0)
require_True(t, bytes.Equal(jmsg.Data, payload))
continue
}
if i == 5 {
require_True(t, jmsg.Header != nil)
require_Equal(t, jmsg.Header.Get(JSMsgId), "MyId")
require_Equal(t, jmsg.Header.Get(traceParentHdr), "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01")
require_Equal(t, jmsg.Header.Get(MsgTraceDest), MsgTraceDestDisabled)
require_True(t, bytes.Equal(jmsg.Data, payload))
continue
}
require_Len(t, len(jmsg.Header), 0)
require_Equal(t, string(jmsg.Data), "hello")
}
msg, err := traceSub.NextMsg(250 * time.Millisecond)
if err != nats.ErrTimeout {
if msg != nil {
t.Fatalf("Expected timeout, got msg headers=%+v data=%s", msg.Header, msg.Data)
}
t.Fatalf("Expected timeout, got err=%v", err)
}
}
func TestMsgTraceWithCompression(t *testing.T) {
o := DefaultOptions()
s := RunServer(o)
defer s.Shutdown()
nc := natsConnect(t, s.ClientURL())
defer nc.Close()
traceSub := natsSubSync(t, nc, "my.trace.subj")
natsFlush(t, nc)
for _, test := range []struct {
compressAlgo string
expectedHdr string
unsupported bool
}{
{"gzip", "gzip", false},
{"snappy", "snappy", false},
{"s2", "snappy", false},
{"bad one", "identity", true},
} {
t.Run(test.compressAlgo, func(t *testing.T) {
msg := nats.NewMsg("foo")
msg.Header.Set(MsgTraceDest, traceSub.Subject)
msg.Header.Set(acceptEncodingHeader, test.compressAlgo)
msg.Data = []byte("hello!")
err := nc.PublishMsg(msg)
require_NoError(t, err)
traceMsg := natsNexMsg(t, traceSub, time.Second)
data := traceMsg.Data
eh := traceMsg.Header.Get(contentEncodingHeader)
require_Equal[string](t, eh, test.expectedHdr)
if test.unsupported {
// We should be able to unmarshal directly
} else {
switch test.expectedHdr {
case "gzip":
zr, err := gzip.NewReader(bytes.NewReader(data))
require_NoError(t, err)
data, err = io.ReadAll(zr)
if err != nil && err != io.ErrUnexpectedEOF {
t.Fatalf("Unexpected error: %v", err)
}
err = zr.Close()
require_NoError(t, err)
case "snappy":
sr := s2.NewReader(bytes.NewReader(data))
data, err = io.ReadAll(sr)
if err != nil && err != io.ErrUnexpectedEOF {
t.Fatalf("Unexpected error: %v", err)
}
}
}
var e MsgTraceEvent
err = json.Unmarshal(data, &e)
require_NoError(t, err)
ingress := e.Ingress()
require_True(t, ingress != nil)
require_Equal[string](t, e.Server.Name, s.Name())
require_Equal[string](t, ingress.Subject, "foo")
})
}
}
func TestMsgTraceHops(t *testing.T) {
// Will have a test with following toplogy
//
// =================== ===================
// = C1 cluster = = C2 cluster =
// =================== <--- Gateway ---> ===================
// = C1-S1 <-> C1-S2 = = C2-S1 =
// =================== ===================
// ^ ^ ^
// | Leafnode | | Leafnode
// | | |
// =================== ===================
// = C3 cluster = = C4 cluster =
// =================== ===================
// = C3-S1 <-> C3-S2 = = C4-S1 =
// =================== ===================
// ^
// | Leafnode
// |-------|
// ===================
// = C5 cluster =
// ===================
// = C5-S1 <-> C5-S2 =
// ===================
//
// And a subscription on "foo" attached to all servers, and the subscription
// on the trace subject attached to c1-s1 (and where the trace message will
// be sent from).
//
commonTmpl := `
port: -1
server_name: "%s-%s"
accounts {
A { users: [{user:"a", pass: "pwd"}] }
$SYS { users: [{user:"admin", pass: "s3cr3t!"}] }
}
system_account: "$SYS"
cluster {
port: -1
name: "%s"
%s
}
`
genCommon := func(cname, sname string, routePort int) string {
var routes string
if routePort > 0 {
routes = fmt.Sprintf(`routes: ["nats://127.0.0.1:%d"]`, routePort)
}
return fmt.Sprintf(commonTmpl, cname, sname, cname, routes)
}
c1s1conf := createConfFile(t, []byte(fmt.Sprintf(`
%s
gateway {
port: -1
name: "C1"
}
leafnodes {
port: -1
}
`, genCommon("C1", "S1", 0))))
c1s1, oc1s1 := RunServerWithConfig(c1s1conf)
defer c1s1.Shutdown()
c1s2conf := createConfFile(t, []byte(fmt.Sprintf(`
%s
gateway {
port: -1
name: "C1"
}
leafnodes {
port: -1
}
`, genCommon("C1", "S2", oc1s1.Cluster.Port))))
c1s2, oc1s2 := RunServerWithConfig(c1s2conf)
defer c1s2.Shutdown()
checkClusterFormed(t, c1s1, c1s2)
c2s1conf := createConfFile(t, []byte(fmt.Sprintf(`
%s
gateway {
port: -1
name: "C2"
gateways [
{
name: "C1"
url: "nats://a:[email protected]:%d"
}
]
}
leafnodes {
port: -1
}
`, genCommon("C2", "S1", 0), oc1s1.Gateway.Port)))
c2s1, oc2s1 := RunServerWithConfig(c2s1conf)
defer c2s1.Shutdown()
c4s1conf := createConfFile(t, []byte(fmt.Sprintf(`
%s
leafnodes {
remotes [{url: "nats://a:[email protected]:%d", account: "A"}]
}
`, genCommon("C4", "S1", 0), oc2s1.LeafNode.Port)))
c4s1, _ := RunServerWithConfig(c4s1conf)
defer c4s1.Shutdown()
for _, s := range []*Server{c1s1, c1s2, c2s1} {
waitForOutboundGateways(t, s, 1, time.Second)
}
waitForInboundGateways(t, c2s1, 2, time.Second)
c3s1conf := createConfFile(t, []byte(fmt.Sprintf(`
%s
leafnodes {
port: -1
remotes [{url: "nats://a:[email protected]:%d", account: "A"}]
}
`, genCommon("C3", "S1", 0), oc1s1.LeafNode.Port)))
c3s1, oc3s1 := RunServerWithConfig(c3s1conf)
defer c3s1.Shutdown()
c3s2conf := createConfFile(t, []byte(fmt.Sprintf(`
%s
leafnodes {
port: -1
remotes [{url: "nats://a:[email protected]:%d", account: "A"}]
}
system_account: "$SYS"
`, genCommon("C3", "S2", oc3s1.Cluster.Port), oc1s2.LeafNode.Port)))
c3s2, oc3s2 := RunServerWithConfig(c3s2conf)
defer c3s2.Shutdown()
checkClusterFormed(t, c3s1, c3s2)
checkLeafNodeConnected(t, c1s1)
checkLeafNodeConnected(t, c1s2)
checkLeafNodeConnected(t, c3s1)
checkLeafNodeConnected(t, c3s2)
c5s1conf := createConfFile(t, []byte(fmt.Sprintf(`
%s
leafnodes {
remotes [{url: "nats://a:[email protected]:%d", account: "A"}]
}
`, genCommon("C5", "S1", 0), oc3s2.LeafNode.Port)))
c5s1, oc5s1 := RunServerWithConfig(c5s1conf)
defer c5s1.Shutdown()
c5s2conf := createConfFile(t, []byte(fmt.Sprintf(`
%s
leafnodes {
remotes [{url: "nats://a:[email protected]:%d", account: "A"}]
}
`, genCommon("C5", "S2", oc5s1.Cluster.Port), oc3s2.LeafNode.Port)))
c5s2, _ := RunServerWithConfig(c5s2conf)
defer c5s2.Shutdown()
checkLeafNodeConnected(t, c5s1)
checkLeafNodeConnected(t, c5s2)
checkLeafNodeConnectedCount(t, c3s2, 3)
nct := natsConnect(t, c1s1.ClientURL(), nats.UserInfo("a", "pwd"), nats.Name("Tracer"))
defer nct.Close()
traceSub := natsSubSync(t, nct, "my.trace.subj")
natsFlush(t, nct)
allServers := []*Server{c1s1, c1s2, c2s1, c3s1, c3s2, c4s1, c5s1, c5s2}
// Check that the subscription interest on the trace subject reaches all servers.
for _, s := range allServers {
if s == c2s1 {
// Gateway needs to be checked differently.
checkGWInterestOnlyModeInterestOn(t, c2s1, "C1", "A", traceSub.Subject)
continue
}
checkSubInterest(t, s, "A", traceSub.Subject, time.Second)
}
var subs []*nats.Subscription
// Now create a subscription on "foo" on all servers (do in reverse order).
for i := len(allServers) - 1; i >= 0; i-- {
s := allServers[i]
nc := natsConnect(t, s.ClientURL(), nats.UserInfo("a", "pwd"), nats.Name(fmt.Sprintf("sub%d", i+1)))
defer nc.Close()
subs = append(subs, natsSubSync(t, nc, "foo"))
natsFlush(t, nc)
}
// Check sub interest on "foo" on all servers.
for _, s := range allServers {
checkSubInterest(t, s, "A", "foo", time.Second)
}
// Now send a trace message from c1s1
msg := nats.NewMsg("foo")
msg.Header.Set(MsgTraceDest, traceSub.Subject)
msg.Data = []byte("hello!")
err := nct.PublishMsg(msg)
require_NoError(t, err)
// Check that all subscriptions received the message
for i, sub := range subs {
appMsg, err := sub.NextMsg(time.Second)
if err != nil {
t.Fatalf("Error getting app message for server %q", allServers[i])
}
require_Equal[string](t, string(appMsg.Data), "hello!")
// Check that no (more) messages are received.
if msg, err := sub.NextMsg(100 * time.Millisecond); err != nats.ErrTimeout {
t.Fatalf("Did not expect application message, got %s", msg.Data)
}
}
events := make(map[string]*MsgTraceEvent, 8)
// We expect 8 events
for i := 0; i < 8; i++ {
traceMsg := natsNexMsg(t, traceSub, time.Second)
var e MsgTraceEvent
json.Unmarshal(traceMsg.Data, &e)
var hop string
if hopVals := e.Request.Header[MsgTraceHop]; len(hopVals) > 0 {
hop = hopVals[0]
}
events[hop] = &e
}
// Make sure we are not receiving more traces
if tm, err := traceSub.NextMsg(250 * time.Millisecond); err == nil {
t.Fatalf("Should not have received trace message: %s", tm.Data)
}
checkIngress := func(e *MsgTraceEvent, kind int, name, hop string) *MsgTraceIngress {
t.Helper()
ingress := e.Ingress()
require_True(t, ingress != nil)
require_True(t, ingress.Kind == kind)
require_Equal[string](t, ingress.Account, "A")
require_Equal[string](t, ingress.Subject, "foo")
require_Equal[string](t, ingress.Name, name)
var hhop string
if hopVals := e.Request.Header[MsgTraceHop]; len(hopVals) > 0 {
hhop = hopVals[0]
}
require_Equal[string](t, hhop, hop)
return ingress
}
checkEgressClient := func(eg *MsgTraceEgress, name string) {
t.Helper()
require_True(t, eg.Kind == CLIENT)
require_Equal[string](t, eg.Name, name)
require_Equal[string](t, eg.Hop, _EMPTY_)
require_Equal[string](t, eg.Subscription, "foo")
require_Equal[string](t, eg.Queue, _EMPTY_)
}
// First, we should have an event without a "hop" header, that is the
// ingress from the client.
e, ok := events[_EMPTY_]
require_True(t, ok)
checkIngress(e, CLIENT, "Tracer", _EMPTY_)
require_Equal[int](t, e.Hops, 3)
egress := e.Egresses()
require_Equal[int](t, len(egress), 4)
var (
leafC3S1Hop string
leafC3S2Hop string
leafC4S1Hop string
leafC5S1Hop string
leafC5S2Hop string
routeC1S2Hop string
gwC2S1Hop string
)
for _, eg := range egress {
switch eg.Kind {
case CLIENT:
checkEgressClient(eg, "sub1")
case ROUTER:
require_Equal[string](t, eg.Name, c1s2.Name())
routeC1S2Hop = eg.Hop
case LEAF:
require_Equal[string](t, eg.Name, c3s1.Name())
leafC3S1Hop = eg.Hop
case GATEWAY:
require_Equal[string](t, eg.Name, c2s1.Name())
gwC2S1Hop = eg.Hop
default:
t.Fatalf("Unexpected egress: %+v", eg)
}
}
// All "hop" ids should be not empty and different from each other
require_True(t, leafC3S1Hop != _EMPTY_ && routeC1S2Hop != _EMPTY_ && gwC2S1Hop != _EMPTY_)
require_True(t, leafC3S1Hop != routeC1S2Hop && leafC3S1Hop != gwC2S1Hop && routeC1S2Hop != gwC2S1Hop)
// Now check the routed server in C1 (c1s2)
e, ok = events[routeC1S2Hop]
require_True(t, ok)
checkIngress(e, ROUTER, c1s1.Name(), routeC1S2Hop)
require_Equal[int](t, e.Hops, 1)
egress = e.Egresses()
require_Equal[int](t, len(egress), 2)
for _, eg := range egress {
switch eg.Kind {
case CLIENT:
checkEgressClient(eg, "sub2")
case LEAF:
require_Equal[string](t, eg.Name, c3s2.Name())
require_Equal[string](t, eg.Hop, routeC1S2Hop+".1")
leafC3S2Hop = eg.Hop
default:
t.Fatalf("Unexpected egress: %+v", eg)
}
}
require_True(t, leafC3S2Hop != _EMPTY_)
// Let's check the gateway server
e, ok = events[gwC2S1Hop]
require_True(t, ok)
checkIngress(e, GATEWAY, c1s1.Name(), gwC2S1Hop)
require_Equal[int](t, e.Hops, 1)
egress = e.Egresses()
require_Equal[int](t, len(egress), 2)
for _, eg := range egress {
switch eg.Kind {
case CLIENT:
checkEgressClient(eg, "sub3")
case LEAF:
require_Equal[string](t, eg.Name, c4s1.Name())
require_Equal[string](t, eg.Hop, gwC2S1Hop+".1")
leafC4S1Hop = eg.Hop
default:
t.Fatalf("Unexpected egress: %+v", eg)
}
}
require_True(t, leafC4S1Hop != _EMPTY_)
// Let's check the C3 cluster, starting at C3-S1
e, ok = events[leafC3S1Hop]
require_True(t, ok)
checkIngress(e, LEAF, c1s1.Name(), leafC3S1Hop)
require_Equal[int](t, e.Hops, 0)
egress = e.Egresses()
require_Equal[int](t, len(egress), 1)
checkEgressClient(egress[0], "sub4")
// Now C3-S2
e, ok = events[leafC3S2Hop]
require_True(t, ok)
checkIngress(e, LEAF, c1s2.Name(), leafC3S2Hop)
require_Equal[int](t, e.Hops, 2)
egress = e.Egresses()
require_Equal[int](t, len(egress), 3)
for _, eg := range egress {
switch eg.Kind {
case CLIENT:
checkEgressClient(eg, "sub5")
case LEAF:
require_True(t, eg.Name == c5s1.Name() || eg.Name == c5s2.Name())
require_True(t, eg.Hop == leafC3S2Hop+".1" || eg.Hop == leafC3S2Hop+".2")
if eg.Name == c5s1.Name() {
leafC5S1Hop = eg.Hop
} else {
leafC5S2Hop = eg.Hop
}
default:
t.Fatalf("Unexpected egress: %+v", eg)
}
}
// The leafC5SxHop must be different and not empty
require_True(t, leafC5S1Hop != _EMPTY_ && leafC5S1Hop != leafC5S2Hop && leafC5S2Hop != _EMPTY_)
// Check the C4 cluster
e, ok = events[leafC4S1Hop]
require_True(t, ok)
checkIngress(e, LEAF, c2s1.Name(), leafC4S1Hop)
require_Equal[int](t, e.Hops, 0)
egress = e.Egresses()
require_Equal[int](t, len(egress), 1)
checkEgressClient(egress[0], "sub6")
// Finally, the C5 cluster, starting with C5-S1
e, ok = events[leafC5S1Hop]
require_True(t, ok)
checkIngress(e, LEAF, c3s2.Name(), leafC5S1Hop)
require_Equal[int](t, e.Hops, 0)
egress = e.Egresses()
require_Equal[int](t, len(egress), 1)
checkEgressClient(egress[0], "sub7")
// Then C5-S2
e, ok = events[leafC5S2Hop]
require_True(t, ok)
checkIngress(e, LEAF, c3s2.Name(), leafC5S2Hop)
require_Equal[int](t, e.Hops, 0)
egress = e.Egresses()
require_Equal[int](t, len(egress), 1)
checkEgressClient(egress[0], "sub8")
}
func TestMsgTraceTriggeredByExternalHeader(t *testing.T) {
tmpl := `
port: -1
server_name: "%s"
accounts {
A {
users: [{user:A, password: pwd}]
trace_dest: "acc.trace.dest"
}
B {
users: [{user:B, password: pwd}]
%s
}
}
cluster {
name: "local"
port: -1
%s
}
`
conf1 := createConfFile(t, []byte(fmt.Sprintf(tmpl, "A", _EMPTY_, _EMPTY_)))
s1, o1 := RunServerWithConfig(conf1)
defer s1.Shutdown()
conf2 := createConfFile(t, []byte(fmt.Sprintf(tmpl, "B", _EMPTY_, fmt.Sprintf(`routes: ["nats://127.0.0.1:%d"]`, o1.Cluster.Port))))
s2, _ := RunServerWithConfig(conf2)
defer s2.Shutdown()
checkClusterFormed(t, s1, s2)
nc2 := natsConnect(t, s2.ClientURL(), nats.UserInfo("A", "pwd"))
defer nc2.Close()
appSub := natsSubSync(t, nc2, "foo")
natsFlush(t, nc2)
checkSubInterest(t, s1, "A", "foo", time.Second)
nc1 := natsConnect(t, s1.ClientURL(), nats.UserInfo("A", "pwd"))
defer nc1.Close()
traceSub := natsSubSync(t, nc1, "trace.dest")
accTraceSub := natsSubSync(t, nc1, "acc.trace.dest")
natsFlush(t, nc1)
checkSubInterest(t, s1, "A", traceSub.Subject, time.Second)
checkSubInterest(t, s1, "A", accTraceSub.Subject, time.Second)
var msgCount int
for _, test := range []struct {
name string
setHeaders func(h nats.Header)
traceTriggered bool
traceOnly bool
expectedAccSub bool
}{
// Tests with external header only (no Nats-Trace-Dest). In this case, the
// trace is triggered based on sampling (last token is `-01`). The presence
// of Nats-Trace-Only has no effect and message should always be delivered
// to the application.
{"only external header sampling",
func(h nats.Header) {
h.Set(traceParentHdr, "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01")
},
true,
false,
true},
{"only external header with different case and sampling",
func(h nats.Header) {
h.Set("TraceParent", "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01")
},
true,
false,
true},
{"only external header sampling but not simply 01",
func(h nats.Header) {
h.Set(traceParentHdr, "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-25")
},
true,
false,
true},
{"only external header no sampling",
func(h nats.Header) {
h.Set(traceParentHdr, "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-00")
},
false,
false,
false},
{"external header sampling and trace only",
func(h nats.Header) {
h.Set(traceParentHdr, "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01")
h.Set(MsgTraceOnly, "true")
},
true,
false,
true},
{"external header no sampling and trace only",
func(h nats.Header) {
h.Set(traceParentHdr, "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-00")
h.Set(MsgTraceOnly, "true")
},
false,
false,
false},
// Tests where Nats-Trace-Dest is present, so ignore external header and
// always deliver to the Nats-Trace-Dest, not the account.
{"trace dest and external header sampling",
func(h nats.Header) {
h.Set(MsgTraceDest, traceSub.Subject)
h.Set(traceParentHdr, "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01")
},
true,
false,
false},
{"trace dest and external header no sampling",
func(h nats.Header) {
h.Set(MsgTraceDest, traceSub.Subject)
h.Set(traceParentHdr, "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-00")
},
true,
false,
false},
{"trace dest with trace only and external header sampling",
func(h nats.Header) {
h.Set(MsgTraceDest, traceSub.Subject)
h.Set(MsgTraceOnly, "true")
h.Set(traceParentHdr, "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01")
},
true,
true,
false},
{"trace dest with trace only and external header no sampling",
func(h nats.Header) {
h.Set(MsgTraceDest, traceSub.Subject)
h.Set(MsgTraceOnly, "true")
h.Set(traceParentHdr, "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-00")
},
true,
true,
false},
} {
t.Run(test.name, func(t *testing.T) {
msg := nats.NewMsg("foo")
test.setHeaders(msg.Header)
msgCount++
msgPayload := fmt.Sprintf("msg%d", msgCount)
msg.Data = []byte(msgPayload)
err := nc1.PublishMsg(msg)
require_NoError(t, err)
if !test.traceOnly {
appMsg := natsNexMsg(t, appSub, time.Second)
require_Equal[string](t, string(appMsg.Data), msgPayload)
}
// Make sure we don't receive more (or not if trace only)
if appMsg, err := appSub.NextMsg(100 * time.Millisecond); err != nats.ErrTimeout {
t.Fatalf("Expected no app message, got %q", appMsg.Data)
}
checkTrace := func(sub *nats.Subscription) {
// We should receive 2 traces, 1 per server.
for i := 0; i < 2; i++ {
tm := natsNexMsg(t, sub, time.Second)
var e MsgTraceEvent
err := json.Unmarshal(tm.Data, &e)
require_NoError(t, err)
}
}
if test.traceTriggered {
if test.expectedAccSub {
checkTrace(accTraceSub)
} else {
checkTrace(traceSub)
}
}
// Make sure no trace is received in the other trace sub
// or no trace received at all.
for _, sub := range []*nats.Subscription{accTraceSub, traceSub} {
if tm, err := sub.NextMsg(100 * time.Millisecond); err != nats.ErrTimeout {
t.Fatalf("Expected no trace for the trace sub on %q, got %q", sub.Subject, tm.Data)
}
}
})
}
nc1.Close()
nc2.Close()
// Now replace connections and subs for account "B"
nc2 = natsConnect(t, s2.ClientURL(), nats.UserInfo("B", "pwd"))
defer nc2.Close()
appSub = natsSubSync(t, nc2, "foo")
natsFlush(t, nc2)
checkSubInterest(t, s1, "B", "foo", time.Second)
nc1 = natsConnect(t, s1.ClientURL(), nats.UserInfo("B", "pwd"))
defer nc1.Close()
traceSub = natsSubSync(t, nc1, "trace.dest")
accTraceSub = natsSubSync(t, nc1, "acc.trace.dest")
natsFlush(t, nc1)
checkSubInterest(t, s1, "B", traceSub.Subject, time.Second)
checkSubInterest(t, s1, "B", accTraceSub.Subject, time.Second)
for _, test := range []struct {
name string
reload bool
}{
{"external header but no account destination", true},
{"external header with account destination added through config reload", false},
} {
t.Run(test.name, func(t *testing.T) {
msg := nats.NewMsg("foo")
msg.Header.Set(traceParentHdr, "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01")
msg.Data = []byte("hello")
err := nc1.PublishMsg(msg)
require_NoError(t, err)
// Application should receive the message
appMsg := natsNexMsg(t, appSub, time.Second)
require_Equal[string](t, string(appMsg.Data), "hello")
// Only once...
if appMsg, err := appSub.NextMsg(100 * time.Millisecond); err != nats.ErrTimeout {
t.Fatalf("Expected no app message, got %q", appMsg.Data)
}
if !test.reload {
// We should receive the traces (1 per server) on the account destination
for i := 0; i < 2; i++ {
tm := natsNexMsg(t, accTraceSub, time.Second)
var e MsgTraceEvent
err := json.Unmarshal(tm.Data, &e)
require_NoError(t, err)
}
}
// No (or no more) trace message should be received.
for _, sub := range []*nats.Subscription{accTraceSub, traceSub} {
if tm, err := sub.NextMsg(100 * time.Millisecond); err != nats.ErrTimeout {
t.Fatalf("Expected no trace for the trace sub on %q, got %q", sub.Subject, tm.Data)
}
}
// Do the config reload and we will repeat the test and now
// we should receive the trace message into the account
// destination trace.
if test.reload {
reloadUpdateConfig(t, s1, conf1, fmt.Sprintf(tmpl, "A", `trace_dest: "acc.trace.dest"`, _EMPTY_))
reloadUpdateConfig(t, s2, conf2, fmt.Sprintf(tmpl, "B", `trace_dest: "acc.trace.dest"`, fmt.Sprintf(`routes: ["nats://127.0.0.1:%d"]`, o1.Cluster.Port)))
}
})
}
}
func TestMsgTraceAccountTraceDestJWTUpdate(t *testing.T) {
// create system account
sysKp, _ := nkeys.CreateAccount()
sysPub, _ := sysKp.PublicKey()
sysCreds := newUser(t, sysKp)
// create account A
akp, _ := nkeys.CreateAccount()
aPub, _ := akp.PublicKey()
claim := jwt.NewAccountClaims(aPub)
aJwt, err := claim.Encode(oKp)
require_NoError(t, err)
dir := t.TempDir()
conf := createConfFile(t, []byte(fmt.Sprintf(`
listen: -1
operator: %s
resolver: {
type: full
dir: '%s'
}
system_account: %s
`, ojwt, dir, sysPub)))
s, _ := RunServerWithConfig(conf)
defer s.Shutdown()
updateJwt(t, s.ClientURL(), sysCreds, aJwt, 1)
nc := natsConnect(t, s.ClientURL(), createUserCreds(t, nil, akp))
defer nc.Close()
sub := natsSubSync(t, nc, "acc.trace.dest")
natsFlush(t, nc)
for i, test := range []struct {
name string
traceTriggered bool
}{
{"no acc dest", false},
{"adding trace dest", true},
{"removing trace dest", false},
} {
t.Run(test.name, func(t *testing.T) {
msg := nats.NewMsg("foo")
msg.Header.Set(traceParentHdr, "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01")
msg.Data = []byte("hello")
err = nc.PublishMsg(msg)
require_NoError(t, err)
if test.traceTriggered {
tm := natsNexMsg(t, sub, time.Second)
var e MsgTraceEvent
err = json.Unmarshal(tm.Data, &e)
require_NoError(t, err)
// Simple check
require_Equal[string](t, e.Server.Name, s.Name())
}
// No (more) trace message expected.
tm, err := sub.NextMsg(250 * time.Millisecond)
if err != nats.ErrTimeout {
t.Fatalf("Expected no trace message, got %s", tm.Data)
}
if i < 2 {
if i == 0 {
claim.Trace = &jwt.MsgTrace{Destination: "acc.trace.dest"}
} else {
claim.Trace = nil
}
aJwt, err = claim.Encode(oKp)
require_NoError(t, err)
updateJwt(t, s.ClientURL(), sysCreds, aJwt, 1)
}
})
}
}
func TestMsgTraceServiceJWTUpdate(t *testing.T) {
// create system account
sysKp, _ := nkeys.CreateAccount()
sysPub, _ := sysKp.PublicKey()
sysCreds := newUser(t, sysKp)
// create account A
akp, _ := nkeys.CreateAccount()
aPub, _ := akp.PublicKey()
aClaim := jwt.NewAccountClaims(aPub)
serviceExport := &jwt.Export{Subject: "req", Type: jwt.Service}
aClaim.Exports.Add(serviceExport)
aJwt, err := aClaim.Encode(oKp)
require_NoError(t, err)
// create account B
bkp, _ := nkeys.CreateAccount()
bPub, _ := bkp.PublicKey()
bClaim := jwt.NewAccountClaims(bPub)
serviceImport := &jwt.Import{Account: aPub, Subject: "req", Type: jwt.Service}
bClaim.Imports.Add(serviceImport)
bJwt, err := bClaim.Encode(oKp)
require_NoError(t, err)
dir := t.TempDir()
conf := createConfFile(t, []byte(fmt.Sprintf(`
listen: -1
operator: %s
resolver: {
type: full
dir: '%s'
}
system_account: %s
`, ojwt, dir, sysPub)))
s, _ := RunServerWithConfig(conf)
defer s.Shutdown()
updateJwt(t, s.ClientURL(), sysCreds, aJwt, 1)
updateJwt(t, s.ClientURL(), sysCreds, bJwt, 1)
ncA := natsConnect(t, s.ClientURL(), createUserCreds(t, nil, akp), nats.Name("Service"))
defer ncA.Close()
natsSub(t, ncA, "req", func(m *nats.Msg) {
m.Respond([]byte("resp"))
})
natsFlush(t, ncA)
ncB := natsConnect(t, s.ClientURL(), createUserCreds(t, nil, bkp))
defer ncB.Close()
sub := natsSubSync(t, ncB, "trace.dest")
natsFlush(t, ncB)
for i, test := range []struct {
name string
allowTrace bool
}{
{"trace not allowed", false},
{"trace allowed", true},
{"trace not allowed again", false},
} {
t.Run(test.name, func(t *testing.T) {
msg := nats.NewMsg("req")
msg.Header.Set(MsgTraceDest, sub.Subject)
msg.Data = []byte("request")
reply, err := ncB.RequestMsg(msg, time.Second)
require_NoError(t, err)
require_Equal[string](t, string(reply.Data), "resp")
tm := natsNexMsg(t, sub, time.Second)
var e MsgTraceEvent
err = json.Unmarshal(tm.Data, &e)
require_NoError(t, err)
require_Equal[string](t, e.Server.Name, s.Name())
require_Equal[string](t, e.Ingress().Account, bPub)
sis := e.ServiceImports()
require_Equal[int](t, len(sis), 1)
si := sis[0]
require_Equal[string](t, si.Account, aPub)
egresses := e.Egresses()
if !test.allowTrace {
require_Equal[int](t, len(egresses), 0)
} else {
require_Equal[int](t, len(egresses), 1)
eg := egresses[0]
require_Equal[string](t, eg.Name, "Service")
require_Equal[string](t, eg.Account, aPub)
require_Equal[string](t, eg.Subscription, "req")
}
// No (more) trace message expected.
tm, err = sub.NextMsg(250 * time.Millisecond)
if err != nats.ErrTimeout {
t.Fatalf("Expected no trace message, got %s", tm.Data)
}
if i < 2 {
// Set AllowTrace to true at the first iteration, then
// false at the second.
aClaim.Exports[0].AllowTrace = (i == 0)
aJwt, err = aClaim.Encode(oKp)
require_NoError(t, err)
updateJwt(t, s.ClientURL(), sysCreds, aJwt, 1)
}
})
}
}
func TestMsgTraceStreamJWTUpdate(t *testing.T) {
// create system account
sysKp, _ := nkeys.CreateAccount()
sysPub, _ := sysKp.PublicKey()
sysCreds := newUser(t, sysKp)
// create account A
akp, _ := nkeys.CreateAccount()
aPub, _ := akp.PublicKey()
aClaim := jwt.NewAccountClaims(aPub)
streamExport := &jwt.Export{Subject: "info", Type: jwt.Stream}
aClaim.Exports.Add(streamExport)
aJwt, err := aClaim.Encode(oKp)
require_NoError(t, err)
// create account B
bkp, _ := nkeys.CreateAccount()
bPub, _ := bkp.PublicKey()
bClaim := jwt.NewAccountClaims(bPub)
streamImport := &jwt.Import{Account: aPub, Subject: "info", To: "b", Type: jwt.Stream}
bClaim.Imports.Add(streamImport)
bJwt, err := bClaim.Encode(oKp)
require_NoError(t, err)
dir := t.TempDir()
conf := createConfFile(t, []byte(fmt.Sprintf(`
listen: -1
operator: %s
resolver: {
type: full
dir: '%s'
}
system_account: %s
`, ojwt, dir, sysPub)))
s, _ := RunServerWithConfig(conf)
defer s.Shutdown()
updateJwt(t, s.ClientURL(), sysCreds, aJwt, 1)
updateJwt(t, s.ClientURL(), sysCreds, bJwt, 1)
ncA := natsConnect(t, s.ClientURL(), createUserCreds(t, nil, akp))
defer ncA.Close()
traceSub := natsSubSync(t, ncA, "trace.dest")
natsFlush(t, ncA)
ncB := natsConnect(t, s.ClientURL(), createUserCreds(t, nil, bkp), nats.Name("BInfo"))
defer ncB.Close()
appSub := natsSubSync(t, ncB, "b.info")
natsFlush(t, ncB)
for i, test := range []struct {
name string
allowTrace bool
}{
{"trace not allowed", false},
{"trace allowed", true},
{"trace not allowed again", false},
} {
t.Run(test.name, func(t *testing.T) {
msg := nats.NewMsg("info")
msg.Header.Set(MsgTraceDest, traceSub.Subject)
msg.Data = []byte("some info")
err = ncA.PublishMsg(msg)
require_NoError(t, err)
appMsg := natsNexMsg(t, appSub, time.Second)
require_Equal[string](t, string(appMsg.Data), "some info")
tm := natsNexMsg(t, traceSub, time.Second)
var e MsgTraceEvent
err = json.Unmarshal(tm.Data, &e)
require_NoError(t, err)
require_Equal[string](t, e.Server.Name, s.Name())
ses := e.StreamExports()
require_Equal[int](t, len(ses), 1)
se := ses[0]
require_Equal[string](t, se.Account, bPub)
require_Equal[string](t, se.To, "b.info")
egresses := e.Egresses()
if !test.allowTrace {
require_Equal[int](t, len(egresses), 0)
} else {
require_Equal[int](t, len(egresses), 1)
eg := egresses[0]
require_Equal[string](t, eg.Name, "BInfo")
require_Equal[string](t, eg.Account, bPub)
require_Equal[string](t, eg.Subscription, "info")
}
// No (more) trace message expected.
tm, err = traceSub.NextMsg(250 * time.Millisecond)
if err != nats.ErrTimeout {
t.Fatalf("Expected no trace message, got %s", tm.Data)
}
if i < 2 {
// Set AllowTrace to true at the first iteration, then
// false at the second.
bClaim.Imports[0].AllowTrace = (i == 0)
bJwt, err = bClaim.Encode(oKp)
require_NoError(t, err)
updateJwt(t, s.ClientURL(), sysCreds, bJwt, 1)
}
})
}
}
func TestMsgTraceParseAccountDestWithSampling(t *testing.T) {
tmpl := `
port: -1
accounts {
A {
users: [{user: a, password: pwd}]
%s
}
}
`
for _, test := range []struct {
name string
samplingStr string
want int
}{
{"trace sampling no dest", `msg_trace: {sampling: 50}`, 0},
{"trace dest only", `msg_trace: {dest: foo}`, 100},
{"trace dest with number only", `msg_trace: {dest: foo, sampling: 20}`, 20},
{"trace dest with percentage", `msg_trace: {dest: foo, sampling: 50%}`, 50},
} {
t.Run(test.name, func(t *testing.T) {
conf := createConfFile(t, []byte(fmt.Sprintf(tmpl, test.samplingStr)))
o := LoadConfig(conf)
_, sampling := o.Accounts[0].getTraceDestAndSampling()
require_Equal[int](t, test.want, sampling)
})
}
}
func TestMsgTraceAccountDestWithSampling(t *testing.T) {
tmpl := `
port: -1
server_name: %s
accounts {
A {
users: [{user: a, password:pwd}]
msg_trace: {dest: "acc.dest"%s}
}
}
cluster {
port: -1
%s
}
`
conf1 := createConfFile(t, []byte(fmt.Sprintf(tmpl, "A", _EMPTY_, _EMPTY_)))
s1, o1 := RunServerWithConfig(conf1)
defer s1.Shutdown()
routes := fmt.Sprintf("routes: [\"nats://127.0.0.1:%d\"]", o1.Cluster.Port)
conf2 := createConfFile(t, []byte(fmt.Sprintf(tmpl, "B", _EMPTY_, routes)))
s2, _ := RunServerWithConfig(conf2)
defer s2.Shutdown()
checkClusterFormed(t, s1, s2)
nc2 := natsConnect(t, s2.ClientURL(), nats.UserInfo("a", "pwd"))
defer nc2.Close()
natsSub(t, nc2, "foo", func(_ *nats.Msg) {})
natsFlush(t, nc2)
nc1 := natsConnect(t, s1.ClientURL(), nats.UserInfo("a", "pwd"))
defer nc1.Close()
sub := natsSubSync(t, nc1, "acc.dest")
natsFlush(t, nc1)
checkSubInterest(t, s1, "A", "foo", time.Second)
checkSubInterest(t, s2, "A", "acc.dest", time.Second)
for iter, test := range []struct {
name string
samplingStr string
sampling int
}{
// Sampling is considered 100% if not specified or <=0 or >= 100.
// To disable sampling, the account destination should not be specified.
{"no sampling specified", _EMPTY_, 100},
{"sampling specified", ", sampling: \"25%\"", 25},
{"no sampling again", _EMPTY_, 100},
} {
t.Run(test.name, func(t *testing.T) {
if iter > 0 {
reloadUpdateConfig(t, s1, conf1, fmt.Sprintf(tmpl, "A", test.samplingStr, _EMPTY_))
reloadUpdateConfig(t, s2, conf2, fmt.Sprintf(tmpl, "B", test.samplingStr, routes))
}
msg := nats.NewMsg("foo")
msg.Header.Set(traceParentHdr, "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01")
total := 400
for i := 0; i < total; i++ {
err := nc1.PublishMsg(msg)
require_NoError(t, err)
}
// Wait a bit to make sure that we received all traces that should
// have been received.
time.Sleep(500 * time.Millisecond)
n, _, err := sub.Pending()
require_NoError(t, err)
fromClient := 0
fromRoute := 0
for i := 0; i < n; i++ {
msg = natsNexMsg(t, sub, time.Second)
var e MsgTraceEvent
err = json.Unmarshal(msg.Data, &e)
require_NoError(t, err)
ingress := e.Ingress()
require_True(t, ingress != nil)
switch ingress.Kind {
case CLIENT:
fromClient++
case ROUTER:
fromRoute++
default:
t.Fatalf("Unexpected ingress: %+v", ingress)
}
}
// There should be as many messages coming from the origin server
// and the routed server. This checks that if sampling is not 100%
// then when a message is routed, the header is properly deactivated.
require_Equal[int](t, fromClient, fromRoute)
// Now check that if sampling was 100%, we have the total number
// of published messages.
if test.sampling == 100 {
require_Equal[int](t, fromClient, total)
} else {
// Otherwise, we should have no more (but let's be conservative)
// than the sampling number.
require_LessThan[int](t, fromClient, int(float64(test.sampling*total/100)*1.35))
}
})
}
}
func TestMsgTraceAccDestWithSamplingJWTUpdate(t *testing.T) {
// create system account
sysKp, _ := nkeys.CreateAccount()
sysPub, _ := sysKp.PublicKey()
sysCreds := newUser(t, sysKp)
// create account A
akp, _ := nkeys.CreateAccount()
aPub, _ := akp.PublicKey()
claim := jwt.NewAccountClaims(aPub)
claim.Trace = &jwt.MsgTrace{Destination: "acc.trace.dest"}
aJwt, err := claim.Encode(oKp)
require_NoError(t, err)
dir := t.TempDir()
conf := createConfFile(t, []byte(fmt.Sprintf(`
listen: -1
operator: %s
resolver: {
type: full
dir: '%s'
}
system_account: %s
`, ojwt, dir, sysPub)))
s, _ := RunServerWithConfig(conf)
defer s.Shutdown()
updateJwt(t, s.ClientURL(), sysCreds, aJwt, 1)
nc := natsConnect(t, s.ClientURL(), createUserCreds(t, nil, akp))
defer nc.Close()
sub := natsSubSync(t, nc, "acc.trace.dest")
natsFlush(t, nc)
for iter, test := range []struct {
name string
sampling int
}{
{"no sampling specified", 100},
{"sampling", 25},
{"set back sampling to 0", 100},
} {
t.Run(test.name, func(t *testing.T) {
if iter > 0 {
claim.Trace = &jwt.MsgTrace{Destination: "acc.trace.dest", Sampling: test.sampling}
aJwt, err = claim.Encode(oKp)
require_NoError(t, err)
updateJwt(t, s.ClientURL(), sysCreds, aJwt, 1)
}
msg := nats.NewMsg("foo")
msg.Header.Set(traceParentHdr, "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01")
msg.Data = []byte("hello")
total := 400
for i := 0; i < total; i++ {
err := nc.PublishMsg(msg)
require_NoError(t, err)
}
// Wait a bit to make sure that we received all traces that should
// have been received.
time.Sleep(500 * time.Millisecond)
n, _, err := sub.Pending()
require_NoError(t, err)
for i := 0; i < n; i++ {
msg = natsNexMsg(t, sub, time.Second)
var e MsgTraceEvent
err = json.Unmarshal(msg.Data, &e)
require_NoError(t, err)
}
// Now check that if sampling was 100%, we have the total number
// of published messages.
if test.sampling == 100 {
require_Equal[int](t, n, total)
} else {
// Otherwise, we should have no more (but let's be conservative)
// than the sampling number.
require_LessThan[int](t, n, int(float64(test.sampling*total/100)*1.35))
}
})
}
}
Reference in New Issue View Git Blame Copy Permalink