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

(2.11) ADR-44: JetStream Dynamic Metadata #5857

Merged
merged 17 commits into from
Sep 7, 2024
Merged

(2.11) ADR-44: JetStream Dynamic Metadata #5857

Show file tree
Hide file tree
Changes from 8 commits
Commits
Show all changes
17 commits
Select commit Hold shift + click to select a range
d3ab9f8
(2.11) ADR-44: JetStream Asset Versioning in Metadata
MauriceVanVeen Aug 30, 2024
ce10203
Lint imports
MauriceVanVeen Aug 30, 2024
7fa3b13
Fix for TestJetStreamClusterConsumerActions
MauriceVanVeen Aug 30, 2024
2c3f998
Skip JS versioning tests when running other tests from the server pac…
MauriceVanVeen Aug 30, 2024
1b70d9c
More comprehensive fix when doing equality check on cfg and prevCfg
MauriceVanVeen Aug 30, 2024
e34a60c
Lint
MauriceVanVeen Aug 30, 2024
491aab5
Simplify validation
MauriceVanVeen Sep 3, 2024
768f8c2
(2.11) ADR-44: JetStream Dynamic Metadata
MauriceVanVeen Sep 2, 2024
7598505
Remove extra copy
MauriceVanVeen Sep 5, 2024
0f60da2
Shorten function names
MauriceVanVeen Sep 5, 2024
604cc0e
Shorten function names
MauriceVanVeen Sep 5, 2024
5e33ddb
Merge branch 'adr-44/jetstream-asset-versioning' into adr-44/dynamic-…
MauriceVanVeen Sep 5, 2024
5dd64fe
Update test names
MauriceVanVeen Sep 5, 2024
ca9ec95
Use int for JSApiLevel
MauriceVanVeen Sep 5, 2024
447ac0b
Merge branch 'adr-44/jetstream-asset-versioning' into adr-44/dynamic-…
MauriceVanVeen Sep 5, 2024
5205a05
Use pointers for setDynamicStreamMetadata
MauriceVanVeen Sep 6, 2024
375287c
Merge branch 'main' into adr-44/dynamic-metadata
MauriceVanVeen Sep 7, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions server/jetstream.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -467,6 +467,7 @@ func (s *Server) enableJetStream(cfg JetStreamConfig) error {
s.Noticef(" TPM File: %q, Pcr: %d", opts.JetStreamTpm.KeysFile,
opts.JetStreamTpm.Pcr)
}
s.Noticef(" API Level: %s", JSApiLevel)
s.Noticef("-------------------------------------------")

// Setup our internal subscriptions.
Expand Down
38 changes: 29 additions & 9 deletions server/jetstream_api.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1422,6 +1422,9 @@ func (s *Server) jsStreamCreateRequest(sub *subscription, c *client, _ *Account,
return
}

// Initialize asset version metadata.
setStreamAssetVersionMetadata(&cfg.StreamConfig, nil)

streamName := streamNameFromSubject(subject)
if streamName != cfg.Name {
resp.Error = NewJSStreamMismatchError()
Expand Down Expand Up @@ -1475,7 +1478,7 @@ func (s *Server) jsStreamCreateRequest(sub *subscription, c *client, _ *Account,
resp.StreamInfo = &StreamInfo{
Created: mset.createdTime(),
State: mset.state(),
Config: mset.config(),
Config: includeDynamicStreamAssetVersionMetadata(mset.config()),
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Prefer shorter names ;)

TimeStamp: time.Now().UTC(),
Mirror: mset.mirrorInfo(),
Sources: mset.sourcesInfo(),
Expand Down Expand Up @@ -1557,6 +1560,9 @@ func (s *Server) jsStreamUpdateRequest(sub *subscription, c *client, _ *Account,
return
}

// Update asset version metadata.
setStreamAssetVersionMetadata(&cfg, &mset.cfg)

if err := mset.updatePedantic(&cfg, ncfg.Pedantic); err != nil {
resp.Error = NewJSStreamUpdateError(err, Unless(err))
s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
Expand All @@ -1566,7 +1572,7 @@ func (s *Server) jsStreamUpdateRequest(sub *subscription, c *client, _ *Account,
resp.StreamInfo = &StreamInfo{
Created: mset.createdTime(),
State: mset.state(),
Config: mset.config(),
Config: includeDynamicStreamAssetVersionMetadata(mset.config()),
Domain: s.getOpts().JetStreamDomain,
Mirror: mset.mirrorInfo(),
Sources: mset.sourcesInfo(),
Expand Down Expand Up @@ -1949,7 +1955,7 @@ func (s *Server) jsStreamInfoRequest(sub *subscription, c *client, a *Account, s
resp.StreamInfo = &StreamInfo{
Created: mset.createdTime(),
State: mset.stateWithDetail(details),
Config: config,
Config: includeDynamicStreamAssetVersionMetadata(config),
Domain: s.getOpts().JetStreamDomain,
Cluster: js.clusterInfo(mset.raftGroup()),
Mirror: mset.mirrorInfo(),
Expand Down Expand Up @@ -3588,7 +3594,7 @@ func (s *Server) processStreamRestore(ci *ClientInfo, acc *Account, cfg *StreamC
resp.StreamInfo = &StreamInfo{
Created: mset.createdTime(),
State: mset.state(),
Config: mset.config(),
Config: includeDynamicStreamAssetVersionMetadata(mset.config()),
TimeStamp: time.Now().UTC(),
}
s.Noticef("Completed restore of %s for stream '%s > %s' in %v",
Expand Down Expand Up @@ -4028,12 +4034,17 @@ func (s *Server) jsConsumerCreateRequest(sub *subscription, c *client, a *Accoun
return
}

// If the consumer already exists then don't allow updating the PauseUntil, just set
// it back to whatever the current configured value is.
var oldCfg *ConsumerConfig
if o := stream.lookupConsumer(consumerName); o != nil {
oldCfg = &o.cfg
// If the consumer already exists then don't allow updating the PauseUntil, just set
// it back to whatever the current configured value is.
req.Config.PauseUntil = o.cfg.PauseUntil
}

// Initialize/update asset version metadata.
setConsumerAssetVersionMetadata(&req.Config, oldCfg)

o, err := stream.addConsumerWithAction(&req.Config, req.Action, req.Pedantic)

if err != nil {
Expand All @@ -4046,7 +4057,7 @@ func (s *Server) jsConsumerCreateRequest(sub *subscription, c *client, a *Accoun
s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
return
}
resp.ConsumerInfo = o.initialInfo()
resp.ConsumerInfo = includeDynamicConsumerInfoVersionMetadata(o.initialInfo())
s.sendAPIResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(resp))

if o.cfg.PauseUntil != nil && !o.cfg.PauseUntil.IsZero() && time.Now().Before(*o.cfg.PauseUntil) {
Expand Down Expand Up @@ -4392,7 +4403,7 @@ func (s *Server) jsConsumerInfoRequest(sub *subscription, c *client, _ *Account,
Stream: ca.Stream,
Name: ca.Name,
Created: ca.Created,
Config: ca.Config,
Config: includeDynamicConsumerAssetVersionMetadata(ca.Config),
TimeStamp: time.Now().UTC(),
}
b := s.jsonResponse(resp)
Expand Down Expand Up @@ -4435,7 +4446,7 @@ func (s *Server) jsConsumerInfoRequest(sub *subscription, c *client, _ *Account,
return
}

if resp.ConsumerInfo = obs.info(); resp.ConsumerInfo == nil {
if resp.ConsumerInfo = includeDynamicConsumerInfoVersionMetadata(obs.info()); resp.ConsumerInfo == nil {
// This consumer returned nil which means it's closed. Respond with not found.
resp.Error = NewJSConsumerNotFoundError()
s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
Expand Down Expand Up @@ -4589,6 +4600,11 @@ func (s *Server) jsConsumerPauseRequest(sub *subscription, c *client, _ *Account
} else {
nca.Config.PauseUntil = nil
}

// Update asset version metadata due to updating pause/resume.
// Only PauseUntil is updated above, so reuse config for both.
setConsumerAssetVersionMetadata(nca.Config, nca.Config)

eca := encodeAddConsumerAssignment(&nca)
cc.meta.Propose(eca)

Expand Down Expand Up @@ -4622,6 +4638,10 @@ func (s *Server) jsConsumerPauseRequest(sub *subscription, c *client, _ *Account
ncfg.PauseUntil = nil
}

// Update asset version metadata due to updating pause/resume.
// Only PauseUntil is updated above, so reuse config for both.
setConsumerAssetVersionMetadata(&ncfg, &ncfg)

if err := obs.updateConfig(&ncfg); err != nil {
// The only type of error that should be returned here is from o.store,
// so use a store failed error type.
Expand Down
26 changes: 20 additions & 6 deletions server/jetstream_cluster.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3278,7 +3278,7 @@ func (js *jetStream) processStreamLeaderChange(mset *stream, isLeader bool) {
resp.StreamInfo = &StreamInfo{
Created: mset.createdTime(),
State: mset.state(),
Config: mset.config(),
Config: includeDynamicStreamAssetVersionMetadata(mset.config()),
Cluster: js.clusterInfo(mset.raftGroup()),
Sources: mset.sourcesInfo(),
Mirror: mset.mirrorInfo(),
Expand Down Expand Up @@ -3708,7 +3708,7 @@ func (js *jetStream) processClusterUpdateStream(acc *Account, osa, sa *streamAss
resp.StreamInfo = &StreamInfo{
Created: mset.createdTime(),
State: mset.state(),
Config: mset.config(),
Config: includeDynamicStreamAssetVersionMetadata(mset.config()),
Cluster: js.clusterInfo(mset.raftGroup()),
Mirror: mset.mirrorInfo(),
Sources: mset.sourcesInfo(),
Expand Down Expand Up @@ -3775,7 +3775,7 @@ func (js *jetStream) processClusterCreateStream(acc *Account, sa *streamAssignme
resp.StreamInfo = &StreamInfo{
Created: mset.createdTime(),
State: mset.state(),
Config: mset.config(),
Config: includeDynamicStreamAssetVersionMetadata(mset.config()),
Cluster: js.clusterInfo(mset.raftGroup()),
Sources: mset.sourcesInfo(),
Mirror: mset.mirrorInfo(),
Expand Down Expand Up @@ -4484,7 +4484,7 @@ func (js *jetStream) processClusterCreateConsumer(ca *consumerAssignment, state
client, subject, reply := ca.Client, ca.Subject, ca.Reply
js.mu.Unlock()
var resp = JSApiConsumerCreateResponse{ApiResponse: ApiResponse{Type: JSApiConsumerCreateResponseType}}
resp.ConsumerInfo = o.info()
resp.ConsumerInfo = includeDynamicConsumerInfoVersionMetadata(o.info())
s.sendAPIResponse(client, acc, subject, reply, _EMPTY_, s.jsonResponse(&resp))
return
}
Expand Down Expand Up @@ -4522,7 +4522,7 @@ func (js *jetStream) processClusterCreateConsumer(ca *consumerAssignment, state
js.mu.RUnlock()
if !recovering {
var resp = JSApiConsumerCreateResponse{ApiResponse: ApiResponse{Type: JSApiConsumerCreateResponseType}}
resp.ConsumerInfo = o.info()
resp.ConsumerInfo = includeDynamicConsumerInfoVersionMetadata(o.info())
s.sendAPIResponse(client, acc, subject, reply, _EMPTY_, s.jsonResponse(&resp))
}
}
Expand Down Expand Up @@ -5222,7 +5222,7 @@ func (js *jetStream) processConsumerLeaderChange(o *consumer, isLeader bool) err
resp.Error = NewJSConsumerCreateError(err, Unless(err))
s.sendAPIErrResponse(client, acc, subject, reply, _EMPTY_, s.jsonResponse(&resp))
} else {
resp.ConsumerInfo = o.initialInfo()
resp.ConsumerInfo = includeDynamicConsumerInfoVersionMetadata(o.initialInfo())
s.sendAPIResponse(client, acc, subject, reply, _EMPTY_, s.jsonResponse(&resp))
o.sendCreateAdvisory()
}
Expand Down Expand Up @@ -6244,6 +6244,10 @@ func (s *Server) jsClusteredStreamUpdateRequest(ci *ClientInfo, acc *Account, su
s.sendAPIErrResponse(ci, acc, subject, reply, string(rmsg), s.jsonResponse(&resp))
return
}

// Update asset version metadata.
setStreamAssetVersionMetadata(cfg, osa.Config)

var newCfg *StreamConfig
if jsa := js.accounts[acc.Name]; jsa != nil {
js.mu.Unlock()
Expand Down Expand Up @@ -7282,6 +7286,9 @@ func (s *Server) jsClusteredConsumerRequest(ci *ClientInfo, acc *Account, subjec
// if name was set by the user.
if oname != _EMPTY_ {
if ca = sa.consumers[oname]; ca != nil && !ca.deleted {
// Provided config might miss metadata, copy from existing config.
copyConsumerAssetVersionMetadata(cfg, ca.Config)

if action == ActionCreate && !reflect.DeepEqual(cfg, ca.Config) {
resp.Error = NewJSConsumerAlreadyExistsError()
s.sendAPIErrResponse(ci, acc, subject, reply, string(rmsg), s.jsonResponse(&resp))
Expand All @@ -7296,6 +7303,13 @@ func (s *Server) jsClusteredConsumerRequest(ci *ClientInfo, acc *Account, subjec
}
}

// Initialize/update asset version metadata.
var oldCfg *ConsumerConfig
if ca != nil {
oldCfg = ca.Config
}
setConsumerAssetVersionMetadata(cfg, oldCfg)

// If this is new consumer.
if ca == nil {
if action == ActionUpdate {
Expand Down
Loading