Repository: nats-io/nats-server
Pull request #5041 (2024-02-05 22:46:20)

[ADDED] JetStream domain elected advisory for when we have a new meta leader.
Signed-off-by: Derek Collison <derek@nats.io>



Patch:
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index 5c501cd4..03c55ea5 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -277,6 +277,9 @@ const (
 	// JSAdvisoryStreamRestoreCompletePre notification that a restore was completed.
 	JSAdvisoryStreamRestoreCompletePre = "$JS.EVENT.ADVISORY.STREAM.RESTORE_COMPLETE"
 
+	// JSAdvisoryDomainLeaderElectedPre notification that a jetstream domain has elected a leader.
+	JSAdvisoryDomainLeaderElected = "$JS.EVENT.ADVISORY.DOMAIN.LEADER_ELECTED"
+
 	// JSAdvisoryStreamLeaderElectedPre notification that a replicated stream has elected a leader.
 	JSAdvisoryStreamLeaderElectedPre = "$JS.EVENT.ADVISORY.STREAM.LEADER_ELECTED"
 
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index e7756cd1..ee3576fe 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -5333,6 +5333,31 @@ func (js *jetStream) stopUpdatesSub() {
 	}
 }
 
+func (s *Server) sendDomainLeaderElectAdvisory() {
+	js, cc := s.getJetStreamCluster()
+	if js == nil || cc == nil {
+		return
+	}
+
+	js.mu.RLock()
+	node := cc.meta
+	js.mu.RUnlock()
+
+	adv := &JSDomainLeaderElectedAdvisory{
+		TypedEvent: TypedEvent{
+			Type: JSDomainLeaderElectedAdvisoryType,
+			ID:   nuid.Next(),
+			Time: time.Now().UTC(),
+		},
+		Leader:   node.GroupLeader(),
+		Replicas: s.replicas(node),
+		Cluster:  s.cachedClusterName(),
+		Domain:   s.getOpts().JetStreamDomain,
+	}
+
+	s.publishAdvisory(nil, JSAdvisoryDomainLeaderElected, adv)
+}
+
 func (js *jetStream) processLeaderChange(isLeader bool) {
 	if js == nil {
 		return
@@ -5346,6 +5371,7 @@ func (js *jetStream) processLeaderChange(isLeader bool) {
 
 	if isLeader {
 		s.Noticef("Self is new JetStream cluster metadata leader")
+		s.sendDomainLeaderElectAdvisory()
 	} else {
 		var node string
 		if meta := js.getMetaGroup(); meta != nil {
diff --git a/server/jetstream_events.go b/server/jetstream_events.go
index 7ea14ef8..1852811b 100644
--- a/server/jetstream_events.go
+++ b/server/jetstream_events.go
@@ -193,10 +193,22 @@ const JSRestoreCompleteAdvisoryType = "io.nats.jetstream.advisory.v1.restore_com
 
 // Clustering specific.
 
-// JSStreamLeaderElectedAdvisoryType is sent when the system elects a leader for a stream.
+// JSClusterLeaderElectedAdvisoryType is sent when the system elects a new meta leader.
+const JSDomainLeaderElectedAdvisoryType = "io.nats.jetstream.advisory.v1.domain_leader_elected"
+
+// JSClusterLeaderElectedAdvisory indicates that a domain has elected a new leader.
+type JSDomainLeaderElectedAdvisory struct {
+	TypedEvent
+	Leader   string      `json:"leader"`
+	Replicas []*PeerInfo `json:"replicas"`
+	Cluster  string      `json:"cluster"`
+	Domain   string      `json:"domain,omitempty"`
+}
+
+// JSStreamLeaderElectedAdvisoryType is sent when the system elects a new leader for a stream.
 const JSStreamLeaderElectedAdvisoryType = "io.nats.jetstream.advisory.v1.stream_leader_elected"
 
-// JSStreamLeaderElectedAdvisory indicates that a stream has lost quorum and is stalled.
+// JSStreamLeaderElectedAdvisory indicates that a stream has elected a new leader.
 type JSStreamLeaderElectedAdvisory struct {
 	TypedEvent
 	Account  string      `json:"account,omitempty"`
@@ -222,7 +234,7 @@ type JSStreamQuorumLostAdvisory struct {
 // JSConsumerLeaderElectedAdvisoryType is sent when the system elects a leader for a consumer.
 const JSConsumerLeaderElectedAdvisoryType = "io.nats.jetstream.advisory.v1.consumer_leader_elected"
 
-// JSConsumerLeaderElectedAdvisory indicates that a stream has lost quorum and is stalled.
+// JSConsumerLeaderElectedAdvisory indicates that a consumer has elected a new leader.
 type JSConsumerLeaderElectedAdvisory struct {
 	TypedEvent
 	Account  string      `json:"account,omitempty"`


Test patch:
diff --git a/server/jetstream_cluster_3_test.go b/server/jetstream_cluster_3_test.go
index a269523c..64335c73 100644
--- a/server/jetstream_cluster_3_test.go
+++ b/server/jetstream_cluster_3_test.go
@@ -5995,3 +5995,39 @@ func TestJetStreamClusterStreamResetPreacks(t *testing.T) {
 		return nil
 	})
 }
+
+func TestJetStreamClusterDomainAdvisory(t *testing.T) {
+	tmpl := strings.Replace(jsClusterAccountsTempl, "store_dir:", "domain: NGS, store_dir:", 1)
+	c := createJetStreamCluster(t, tmpl, "R3S", _EMPTY_, 3, 18033, true)
+	defer c.shutdown()
+
+	// Connect to system account.
+	nc, _ := jsClientConnect(t, c.randomServer(), nats.UserInfo("admin", "s3cr3t!"))
+	defer nc.Close()
+
+	sub, err := nc.SubscribeSync(JSAdvisoryDomainLeaderElected)
+	require_NoError(t, err)
+
+	// Ask meta leader to move and make sure we get an advisory.
+	nc.Request(JSApiLeaderStepDown, nil, time.Second)
+	c.waitOnLeader()
+
+	checkSubsPending(t, sub, 1)
+
+	m, err := sub.NextMsg(time.Second)
+	require_NoError(t, err)
+
+	var adv JSDomainLeaderElectedAdvisory
+	require_NoError(t, json.Unmarshal(m.Data, &adv))
+
+	ml := c.leader()
+	js, cc := ml.getJetStreamCluster()
+	js.mu.RLock()
+	peer := cc.meta.ID()
+	js.mu.RUnlock()
+
+	require_Equal(t, adv.Leader, peer)
+	require_Equal(t, adv.Domain, "NGS")
+	require_Equal(t, adv.Cluster, "R3S")
+	require_Equal(t, len(adv.Replicas), 3)
+}