Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[IMPROVED] Share higher fidelity client info with JetStream #5019

Merged
merged 2 commits into from
Jan 31, 2024
Merged

[IMPROVED] Share higher fidelity client info with JetStream #5019

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
c339c8d
Share higher fidelity client info with JetStream
derekcollison Jan 31, 2024
8e0e768
Need this protection here as well since calling directly for JetStrea…
derekcollison Jan 31, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
4 changes: 4 additions & 0 deletions server/accounts.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1877,6 +1877,10 @@ func (a *Account) addServiceImport(dest *Account, from, to string, claim *jwt.Im
rt := Singleton
var lat *serviceLatency

if dest == nil {
return nil, ErrMissingAccount
}

dest.mu.RLock()
se := dest.getServiceExport(to)
if se != nil {
Expand Down
9 changes: 7 additions & 2 deletions server/jetstream.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
// Copyright 2019-2023 The NATS Authors
// Copyright 2019-2024 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
Expand Down Expand Up @@ -653,9 +653,14 @@ func (a *Account) enableAllJetStreamServiceImportsAndMappings() error {
}

if !a.serviceImportExists(jsAllAPI) {
if err := a.AddServiceImport(s.SystemAccount(), jsAllAPI, _EMPTY_); err != nil {
// Capture si so we can turn on implicit sharing with JetStream layer.
si, err := a.addServiceImport(s.SystemAccount(), jsAllAPI, _EMPTY_, nil)
if err != nil {
return fmt.Errorf("Error setting up jetstream service imports for account: %v", err)
}
a.mu.Lock()
si.share = true
a.mu.Unlock()
}

// Check if we have a Domain specified.
Expand Down
2 changes: 1 addition & 1 deletion server/jetstream_events.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@ import (
"time"
)

func (s *Server) publishAdvisory(acc *Account, subject string, adv interface{}) {
func (s *Server) publishAdvisory(acc *Account, subject string, adv any) {
if acc == nil {
acc = s.SystemAccount()
if acc == nil {
Expand Down
10 changes: 8 additions & 2 deletions server/jetstream_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10631,8 +10631,14 @@ func TestJetStreamAccountImportJSAdvisoriesAsService(t *testing.T) {
gotEvents = map[string]int{}
for i := 0; i < 2; i++ {
msg, err := subAgg.NextMsg(time.Second * 2)
if err != nil {
t.Fatalf("Unexpected error: %v", err)
require_NoError(t, err)
var adv JSAPIAudit
require_NoError(t, json.Unmarshal(msg.Data, &adv))
// Make sure we have full fidelity info via implicit share.
if adv.Client != nil {
require_True(t, adv.Client.Host != _EMPTY_)
require_True(t, adv.Client.User != _EMPTY_)
require_True(t, adv.Client.Lang != _EMPTY_)
}
gotEvents[msg.Subject]++
}
Expand Down