Repository: nats-io/nats-server
Pull request #3023 (2022-04-11 14:56:20)

Stream Alternates
This introduces stream alternates as an optional part of StreamInfo.

Alternates provide a priority list of mirrors and the source in relation to where the request originated.
So if a stream S has 2 mirrors, and we receive a user request for the stream info for S, it will include a sorted list of itself, since it could be the closest, and all mirrors, allowing clients to select read replicas if so desired.

Signed-off-by: Derek Collison <derek@nats.io>

/cc @nats-io/core


ripienaar: Looks pretty good, you say optional part of stream info - but its not something i can request or not its always there, you mean optional as in it's not there post create for example right? Just to be sure we didnt miss some option somewhere

It's nice at initial start but do you anticipate for example a KV bucket would request info regularly to reconfigure itself for new mirros should they become available? 
derekcollison: Won't merge until other PRs land and @kozlovic can do his magic to not have all the redundant commits.

Patch:
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index ae00a413..c73da01d 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -1388,7 +1388,12 @@ func (s *Server) jsStreamCreateRequest(sub *subscription, c *client, _ *Account,
 		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
 		return
 	}
-	resp.StreamInfo = &StreamInfo{Created: mset.createdTime(), State: mset.state(), Config: mset.config()}
+	resp.StreamInfo = &StreamInfo{
+		Created: mset.createdTime(),
+		State:   mset.state(),
+		Config:  mset.config(),
+		Domain:  s.getOpts().JetStreamDomain,
+	}
 	resp.DidCreate = true
 	s.sendAPIResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(resp))
 }
@@ -1488,9 +1493,14 @@ func (s *Server) jsStreamUpdateRequest(sub *subscription, c *client, _ *Account,
 		return
 	}
 
-	js, _ := s.getJetStreamCluster()
-
-	resp.StreamInfo = &StreamInfo{Created: mset.createdTime(), State: mset.state(), Config: mset.config(), Cluster: js.clusterInfo(mset.raftGroup())}
+	resp.StreamInfo = &StreamInfo{
+		Created: mset.createdTime(),
+		State:   mset.state(),
+		Config:  mset.config(),
+		Domain:  s.getOpts().JetStreamDomain,
+		Mirror:  mset.mirrorInfo(),
+		Sources: mset.sourcesInfo(),
+	}
 	s.sendAPIResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(resp))
 }
 
@@ -1702,13 +1712,15 @@ func (s *Server) jsStreamListRequest(sub *subscription, c *client, _ *Account, s
 	}
 
 	for _, mset := range msets[offset:] {
+		config := mset.config()
 		resp.Streams = append(resp.Streams, &StreamInfo{
 			Created: mset.createdTime(),
 			State:   mset.state(),
-			Config:  mset.config(),
+			Config:  config,
+			Domain:  s.getOpts().JetStreamDomain,
 			Mirror:  mset.mirrorInfo(),
-			Sources: mset.sourcesInfo()},
-		)
+			Sources: mset.sourcesInfo(),
+		})
 		if len(resp.Streams) >= JSApiListLimit {
 			break
 		}
@@ -1845,11 +1857,14 @@ func (s *Server) jsStreamInfoRequest(sub *subscription, c *client, a *Account, s
 	js, _ := s.getJetStreamCluster()
 
 	resp.StreamInfo = &StreamInfo{
-		Created: mset.createdTime(),
-		State:   mset.stateWithDetail(details),
-		Config:  config,
-		Domain:  s.getOpts().JetStreamDomain,
-		Cluster: js.clusterInfo(mset.raftGroup()),
+		Created:    mset.createdTime(),
+		State:      mset.stateWithDetail(details),
+		Config:     config,
+		Domain:     s.getOpts().JetStreamDomain,
+		Cluster:    js.clusterInfo(mset.raftGroup()),
+		Mirror:     mset.mirrorInfo(),
+		Sources:    mset.sourcesInfo(),
+		Alternates: js.streamAlternates(ci, config.Name),
 	}
 	if clusterWideConsCount > 0 {
 		resp.StreamInfo.State.Consumers = clusterWideConsCount
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index e881263c..612ae6fd 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -6067,6 +6067,59 @@ func (mset *stream) checkClusterInfo(ci *ClusterInfo) {
 	}
 }
 
+// Return a list of alternates, ranked by preference order to the request, of stream mirrors.
+// This allows clients to select or get more information about read replicas that could be a
+// better option to connect to versus the original source.
+func (js *jetStream) streamAlternates(ci *ClientInfo, stream string) []StreamAlternate {
+	if js == nil {
+		return nil
+	}
+
+	js.mu.RLock()
+	defer js.mu.RUnlock()
+
+	s, cc := js.srv, js.cluster
+	// Track our domain.
+	domain := s.getOpts().JetStreamDomain
+
+	// No clustering just return nil.
+	if cc == nil {
+		return nil
+	}
+	acc, _ := s.LookupAccount(ci.serviceAccount())
+	if acc == nil {
+		return nil
+	}
+
+	// Collect our ordering first for clusters.
+	weights := make(map[string]int)
+	all := []string{ci.Cluster}
+	all = append(all, ci.Alternates...)
+
+	for i := 0; i < len(all); i++ {
+		weights[all[i]] = len(all) - i
+	}
+
+	var alts []StreamAlternate
+	for _, sa := range cc.streams[acc.Name] {
+		// Add in ourselves and any mirrors.
+		if sa.Config.Name == stream || (sa.Config.Mirror != nil && sa.Config.Mirror.Name == stream) {
+			alts = append(alts, StreamAlternate{Name: sa.Config.Name, Domain: domain, Cluster: sa.Group.Cluster})
+		}
+	}
+	// If just us don't fill in.
+	if len(alts) == 1 {
+		return nil
+	}
+
+	// Sort based on our weights that originate from the request itself.
+	sort.Slice(alts, func(i, j int) bool {
+		return weights[alts[i].Cluster] > weights[alts[j].Cluster]
+	})
+
+	return alts
+}
+
 func (mset *stream) handleClusterStreamInfoRequest(sub *subscription, c *client, _ *Account, subject, reply string, _ []byte) {
 	mset.mu.RLock()
 	sysc, js, sa, config := mset.sysc, mset.srv.js, mset.sa, mset.cfg
diff --git a/server/stream.go b/server/stream.go
index 54ba8605..b0482884 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -96,13 +96,20 @@ type PubAck struct {
 
 // StreamInfo shows config and current state for this stream.
 type StreamInfo struct {
-	Config  StreamConfig        `json:"config"`
-	Created time.Time           `json:"created"`
-	State   StreamState         `json:"state"`
-	Domain  string              `json:"domain,omitempty"`
-	Cluster *ClusterInfo        `json:"cluster,omitempty"`
-	Mirror  *StreamSourceInfo   `json:"mirror,omitempty"`
-	Sources []*StreamSourceInfo `json:"sources,omitempty"`
+	Config     StreamConfig        `json:"config"`
+	Created    time.Time           `json:"created"`
+	State      StreamState         `json:"state"`
+	Domain     string              `json:"domain,omitempty"`
+	Cluster    *ClusterInfo        `json:"cluster,omitempty"`
+	Mirror     *StreamSourceInfo   `json:"mirror,omitempty"`
+	Sources    []*StreamSourceInfo `json:"sources,omitempty"`
+	Alternates []StreamAlternate   `json:"alternates,omitempty"`
+}
+
+type StreamAlternate struct {
+	Name    string `json:"name"`
+	Domain  string `json:"domain,omitempty"`
+	Cluster string `json:"cluster"`
 }
 
 // ClusterInfo shows information about the underlying set of servers


Test patch:
diff --git a/server/jetstream_cluster_test.go b/server/jetstream_cluster_test.go
index 6f96b560..d871fcc9 100644
--- a/server/jetstream_cluster_test.go
+++ b/server/jetstream_cluster_test.go
@@ -12384,6 +12384,66 @@ func TestJetStreamImportConsumerStreamSubjectRemap(t *testing.T) {
 	})
 }
 
+func TestJetStreamClusterStreamAlternates(t *testing.T) {
+	sc := createJetStreamTaggedSuperCluster(t)
+	defer sc.shutdown()
+
+	nc, js := jsClientConnect(t, sc.randomServer())
+	defer nc.Close()
+
+	// C1
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:      "SOURCE",
+		Subjects:  []string{"foo", "bar", "baz"},
+		Replicas:  3,
+		Placement: &nats.Placement{Tags: []string{"cloud:aws", "country:us"}},
+	})
+	require_NoError(t, err)
+
+	// C2
+	_, err = js.AddStream(&nats.StreamConfig{
+		Name:      "MIRROR-1",
+		Replicas:  1,
+		Mirror:    &nats.StreamSource{Name: "SOURCE"},
+		Placement: &nats.Placement{Tags: []string{"cloud:gcp", "country:uk"}},
+	})
+	require_NoError(t, err)
+
+	// C3
+	_, err = js.AddStream(&nats.StreamConfig{
+		Name:      "MIRROR-2",
+		Replicas:  2,
+		Mirror:    &nats.StreamSource{Name: "SOURCE"},
+		Placement: &nats.Placement{Tags: []string{"cloud:az", "country:jp"}},
+	})
+	require_NoError(t, err)
+
+	// No client support yet, so do by hand.
+	getStreamInfo := func(nc *nats.Conn, expected string) {
+		t.Helper()
+		resp, err := nc.Request(fmt.Sprintf(JSApiStreamInfoT, "SOURCE"), nil, time.Second)
+		require_NoError(t, err)
+		var si StreamInfo
+		err = json.Unmarshal(resp.Data, &si)
+		require_NoError(t, err)
+		require_True(t, len(si.Alternates) == 3)
+		require_True(t, si.Alternates[0].Cluster == expected)
+		seen := make(map[string]struct{})
+		for _, alt := range si.Alternates {
+			seen[alt.Cluster] = struct{}{}
+		}
+		require_True(t, len(seen) == 3)
+	}
+
+	// Connect to different clusters to check ordering.
+	nc, _ = jsClientConnect(t, sc.clusterForName("C1").randomServer())
+	getStreamInfo(nc, "C1")
+	nc, _ = jsClientConnect(t, sc.clusterForName("C2").randomServer())
+	getStreamInfo(nc, "C2")
+	nc, _ = jsClientConnect(t, sc.clusterForName("C3").randomServer())
+	getStreamInfo(nc, "C3")
+}
+
 // Support functions
 
 // Used to setup superclusters for tests.