Repository: nats-io/nats-server
Pull request #1854 (2021-01-27 23:11:14)

[added] cluster/gateway/leafnode  tls required/verify/timeout to varz
Signed-off-by: Matthias Hanel <mh@synadia.com>

per request from @davidkemper


Patch:
diff --git a/server/const.go b/server/const.go
index c89e4cc4..a4944eea 100644
--- a/server/const.go
+++ b/server/const.go
@@ -40,7 +40,7 @@ var (
 
 const (
 	// VERSION is the current version for the server.
-	VERSION = "2.2.0-beta.56"
+	VERSION = "2.2.0-beta.58"
 
 	// PROTO is the currently supported protocol.
 	// 0 was the original
diff --git a/server/consumer.go b/server/consumer.go
index 7533234e..4b3c5557 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -2316,6 +2316,13 @@ func (mset *Stream) DeleteConsumer(o *Consumer) error {
 	return o.Delete()
 }
 
+func (o *Consumer) Stream() string {
+	o.mu.RLock()
+	mset := o.mset
+	o.mu.RUnlock()
+	return mset.Name()
+}
+
 // Active indicates if this consumer is still active.
 func (o *Consumer) Active() bool {
 	o.mu.Lock()
diff --git a/server/errors.go b/server/errors.go
index 22cf76cc..b82d5183 100644
--- a/server/errors.go
+++ b/server/errors.go
@@ -185,6 +185,9 @@ var (
 	// ErrJetStreamStreamAlreadyUsed is returned when a stream name has already been taken.
 	ErrJetStreamStreamAlreadyUsed = errors.New("stream name already in use")
 
+	// ErrJetStreamConsumerAlreadyUsed is returned when a consumer name has already been taken.
+	ErrJetStreamConsumerAlreadyUsed = errors.New("consumer name already in use")
+
 	// ErrJetStreamNotEnabledForAccount is returned JetStream is not enabled for this account.
 	ErrJetStreamNotEnabledForAccount = errors.New("jetstream not enabled for account")
 
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index d00aa90f..ae9ea579 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -162,33 +162,45 @@ const (
 	// JSAdvisoryConsumerMsgTerminatedPre is a notification published when a message has been terminated.
 	JSAdvisoryConsumerMsgTerminatedPre = "$JS.EVENT.ADVISORY.CONSUMER.MSG_TERMINATED"
 
-	// JSAdvisoryStreamCreatedPre notification that a stream was created
+	// JSAdvisoryStreamCreatedPre notification that a stream was created.
 	JSAdvisoryStreamCreatedPre = "$JS.EVENT.ADVISORY.STREAM.CREATED"
 
-	// JSAdvisoryStreamDeletedPre notification that a stream was deleted
+	// JSAdvisoryStreamDeletedPre notification that a stream was deleted.
 	JSAdvisoryStreamDeletedPre = "$JS.EVENT.ADVISORY.STREAM.DELETED"
 
-	// JSAdvisoryStreamUpdatedPre notification that a stream was updated
+	// JSAdvisoryStreamUpdatedPre notification that a stream was updated.
 	JSAdvisoryStreamUpdatedPre = "$JS.EVENT.ADVISORY.STREAM.UPDATED"
 
-	// JSAdvisoryConsumerCreatedPre notification that a template created
+	// JSAdvisoryConsumerCreatedPre notification that a template created.
 	JSAdvisoryConsumerCreatedPre = "$JS.EVENT.ADVISORY.CONSUMER.CREATED"
 
-	// JSAdvisoryConsumerDeletedPre notification that a template deleted
+	// JSAdvisoryConsumerDeletedPre notification that a template deleted.
 	JSAdvisoryConsumerDeletedPre = "$JS.EVENT.ADVISORY.CONSUMER.DELETED"
 
-	// JSAdvisoryStreamSnapshotCreatePre notification that a snapshot was created
+	// JSAdvisoryStreamSnapshotCreatePre notification that a snapshot was created.
 	JSAdvisoryStreamSnapshotCreatePre = "$JS.EVENT.ADVISORY.STREAM.SNAPSHOT_CREATE"
 
-	// JSAdvisoryStreamSnapshotCompletePre notification that a snapshot was completed
+	// JSAdvisoryStreamSnapshotCompletePre notification that a snapshot was completed.
 	JSAdvisoryStreamSnapshotCompletePre = "$JS.EVENT.ADVISORY.STREAM.SNAPSHOT_COMPLETE"
 
-	// JSAdvisoryStreamRestoreCreatePre notification that a restore was start
+	// JSAdvisoryStreamRestoreCreatePre notification that a restore was start.
 	JSAdvisoryStreamRestoreCreatePre = "$JS.EVENT.ADVISORY.STREAM.RESTORE_CREATE"
 
-	// JSAdvisoryStreamRestoreCompletePre notification that a restore was completed
+	// JSAdvisoryStreamRestoreCompletePre notification that a restore was completed.
 	JSAdvisoryStreamRestoreCompletePre = "$JS.EVENT.ADVISORY.STREAM.RESTORE_COMPLETE"
 
+	// JSAdvisoryStreamLeaderElectPre notification that a replicated stream has elected a leader.
+	JSAdvisoryStreamLeaderElectedPre = "$JS.EVENT.ADVISORY.STREAM.LEADER_ELECTED"
+
+	// JSAdvisoryStreamQuorumLostPre notification that a stream and its consumers are stalled.
+	JSAdvisoryStreamQuorumLostPre = "$JS.EVENT.ADVISORY.STREAM.QUORUM_LOST"
+
+	// JSAdvisoryConsumerLeaderElectPre notification that a replicated consumer has elected a leader.
+	JSAdvisoryConsumerLeaderElectedPre = "$JS.EVENT.ADVISORY.CONSUMER.LEADER_ELECTED"
+
+	// JSAdvisoryConsumerQuorumLostPre notification that a consumer is stalled.
+	JSAdvisoryConsumerQuorumLostPre = "$JS.EVENT.ADVISORY.CONSUMER.QUORUM_LOST"
+
 	// JSAuditAdvisory is a notification about JetStream API access.
 	// FIXME - Add in details about who..
 	JSAuditAdvisory = "$JS.EVENT.ADVISORY.API"
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index e2f01c7f..d1be8650 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -897,7 +897,7 @@ func (rg *raftGroup) setPreferred() {
 }
 
 // createRaftGroup is called to spin up this raft group if needed.
-func (js *jetStream) createRaftGroup(rg *raftGroup) {
+func (js *jetStream) createRaftGroup(rg *raftGroup) error {
 	js.mu.Lock()
 	defer js.mu.Unlock()
 
@@ -906,31 +906,31 @@ func (js *jetStream) createRaftGroup(rg *raftGroup) {
 	// If this is a single peer raft group or we are not a member return.
 	if len(rg.Peers) <= 1 || !rg.isMember(cc.meta.ID()) {
 		// Nothing to do here.
-		return
+		return nil
 	}
 
 	// We already have this assigned.
 	if node := s.lookupRaftNode(rg.Name); node != nil {
 		s.Debugf("JetStream cluster already has raft group %q assigned", rg.Name)
-		return
+		return nil
 	}
 
 	s.Debugf("JetStream cluster creating raft group:%+v", rg)
 
 	sysAcc := s.SystemAccount()
 	if sysAcc == nil {
-		s.Debugf("JetStream cluster detected shutdown processing raft group:%+v", rg)
-		return
+		s.Debugf("JetStream cluster detected shutdown processing raft group: %+v", rg)
+		return errors.New("shutting down")
 	}
 
 	stateDir := path.Join(js.config.StoreDir, sysAcc.Name, defaultStoreDirName, rg.Name)
 	fs, bootstrap, err := newFileStore(
 		FileStoreConfig{StoreDir: stateDir, BlockSize: 32 * 1024 * 1024},
-		StreamConfig{Name: rg.Name, Storage: rg.Storage},
+		StreamConfig{Name: rg.Name, Storage: FileStorage},
 	)
 	if err != nil {
 		s.Errorf("Error creating filestore: %v", err)
-		return
+		return err
 	}
 
 	cfg := &RaftConfig{Name: rg.Name, Store: stateDir, Log: fs}
@@ -941,7 +941,7 @@ func (js *jetStream) createRaftGroup(rg *raftGroup) {
 	n, err := s.startRaftNode(cfg)
 	if err != nil {
 		s.Debugf("Error creating raft group: %v", err)
-		return
+		return err
 	}
 	rg.node = n
 
@@ -949,6 +949,8 @@ func (js *jetStream) createRaftGroup(rg *raftGroup) {
 	if n.ID() == rg.Preferred {
 		n.Campaign()
 	}
+
+	return nil
 }
 
 func (mset *Stream) raftNode() RaftNode {
@@ -1239,21 +1241,65 @@ func (js *jetStream) applyStreamEntries(mset *Stream, ce *CommittedEntry) (bool,
 	return didSnap, nil
 }
 
+// Returns the PeerInfo for all replicas of a raft node. This is different than node.Peers()
+// and is used for external facing advisories.
+func (s *Server) replicas(node RaftNode) []*PeerInfo {
+	now := time.Now()
+	var replicas []*PeerInfo
+	for _, rp := range node.Peers() {
+		pi := &PeerInfo{Name: s.serverNameForNode(rp.ID), Current: rp.Current, Active: now.Sub(rp.Last)}
+		replicas = append(replicas, pi)
+	}
+	return replicas
+}
+
 func (js *jetStream) processStreamLeaderChange(mset *Stream, sa *streamAssignment, isLeader bool) {
+	js.mu.Lock()
+	s, account, err := js.srv, sa.Client.Account, sa.err
+	client, reply := sa.Client, sa.Reply
+	hasResponded := sa.responded
+	sa.responded = true
+	js.mu.Unlock()
+
+	stream := mset.Name()
+
 	if isLeader {
-		js.srv.Noticef("JetStream cluster new stream leader for '%s > %s'", sa.Client.Account, mset.Name())
+		s.Noticef("JetStream cluster new stream leader for '%s > %s'", sa.Client.Account, stream)
+		if node := mset.raftNode(); node != nil {
+			s.publishAdvisory(mset.account(), JSAdvisoryStreamLeaderElectedPre+"."+stream, &JSStreamLeaderElectedAdvisory{
+				TypedEvent: TypedEvent{
+					Type: JSStreamLeaderElectedAdvisoryType,
+					ID:   nuid.Next(),
+					Time: time.Now().UTC(),
+				},
+				Stream:   stream,
+				Leader:   s.serverNameForNode(node.GroupLeader()),
+				Replicas: s.replicas(node),
+			})
+		}
+	} else {
+		// We are stepping down. Make sure if we are doing so because we have lost quorum that
+		// we send the appropriate advisories.
+		if node := mset.raftNode(); node != nil && !node.Quorum() {
+			s.Warnf("JetStream cluster stream '%s > %s' has lost quorum, stalled.", sa.Client.Account, stream)
+			s.publishAdvisory(mset.account(), JSAdvisoryStreamQuorumLostPre+"."+stream, &JSStreamQuorumLostAdvisory{
+				TypedEvent: TypedEvent{
+					Type: JSStreamQuorumLostAdvisoryType,
+					ID:   nuid.Next(),
+					Time: time.Now().UTC(),
+				},
+				Stream:   stream,
+				Replicas: s.replicas(node),
+			})
+		}
 	}
 
+	// Tell stream to switch leader status.
 	mset.setLeader(isLeader)
 
-	js.mu.Lock()
-	if !isLeader || sa.responded {
-		js.mu.Unlock()
+	if !isLeader || hasResponded {
 		return
 	}
-	s, account, err := js.srv, sa.Client.Account, sa.err
-	client, reply := sa.Client, sa.Reply
-	js.mu.Unlock()
 
 	acc, _ := s.LookupAccount(account)
 	if acc == nil {
@@ -1325,32 +1371,23 @@ func (js *jetStream) processStreamAssignment(sa *streamAssignment) {
 
 	// Check if this is for us..
 	if isMember {
-		js.processClusterCreateStream(sa)
+		js.processClusterCreateStream(acc, sa)
 	}
 }
 
 // processClusterCreateStream is called when we have a stream assignment that
 // has been committed and this server is a member of the peer group.
-func (js *jetStream) processClusterCreateStream(sa *streamAssignment) {
+func (js *jetStream) processClusterCreateStream(acc *Account, sa *streamAssignment) {
 	if sa == nil {
 		return
 	}
 
 	js.mu.RLock()
-	s := js.srv
-	acc, err := s.LookupAccount(sa.Client.Account)
-	if err != nil {
-		s.Warnf("JetStream cluster failed to lookup account %q: %v", sa.Client.Account, err)
-		js.mu.RUnlock()
-		return
-	}
-	rg := sa.Group
+	s, rg := js.srv, sa.Group
 	js.mu.RUnlock()
 
 	// Process the raft group and make sure it's running if needed.
-	js.createRaftGroup(rg)
-
-	var mset *Stream
+	err := js.createRaftGroup(rg)
 
 	// If we are restoring, create the stream if we are R>1 and not the preferred who handles the
 	// receipt of the snapshot itself.
@@ -1360,8 +1397,11 @@ func (js *jetStream) processClusterCreateStream(sa *streamAssignment) {
 			shouldCreate = false
 		}
 	}
+
+	var mset *Stream
+
 	// Process here if not restoring or not the leader.
-	if shouldCreate {
+	if shouldCreate && err == nil {
 		// Go ahead and create or update the stream.
 		mset, err = acc.LookupStream(sa.Config.Name)
 		if err == nil && mset != nil {
@@ -1977,27 +2017,60 @@ func decodeDeliveredUpdate(buf []byte) (dseq, sseq, dc uint64, ts int64, err err
 }
 
 func (js *jetStream) processConsumerLeaderChange(o *Consumer, ca *consumerAssignment, isLeader bool) {
-	if isLeader {
-		js.srv.Noticef("JetStream cluster new consumer leader for '%s > %s > %s'", ca.Client.Account, ca.Stream, ca.Name)
-	}
-
-	o.setLeader(isLeader)
-
 	js.mu.Lock()
-	if !isLeader || ca.responded {
-		ca.responded = true
-		js.mu.Unlock()
-		return
-	}
 	s, account, err := js.srv, ca.Client.Account, ca.err
 	client, reply := ca.Client, ca.Reply
+	hasResponded := ca.responded
+	ca.responded = true
 	js.mu.Unlock()
 
+	stream := o.Stream()
+	consumer := o.Name()
 	acc, _ := s.LookupAccount(account)
 	if acc == nil {
 		return
 	}
 
+	if isLeader {
+		s.Noticef("JetStream cluster new consumer leader for '%s > %s > %s'", ca.Client.Account, stream, consumer)
+		if node := o.raftNode(); node != nil {
+			s.publishAdvisory(acc, JSAdvisoryConsumerLeaderElectedPre+"."+stream+"."+consumer, &JSConsumerLeaderElectedAdvisory{
+				TypedEvent: TypedEvent{
+					Type: JSConsumerLeaderElectedAdvisoryType,
+					ID:   nuid.Next(),
+					Time: time.Now().UTC(),
+				},
+				Stream:   stream,
+				Consumer: consumer,
+				Leader:   s.serverNameForNode(node.GroupLeader()),
+				Replicas: s.replicas(node),
+			})
+		}
+	} else {
+		// We are stepping down. Make sure if we are doing so because we have lost quorum that
+		// we send the appropriate advisories.
+		if node := o.raftNode(); node != nil && !node.Quorum() {
+			s.Warnf("JetStream cluster consumer '%s > %s >%s' has lost quorum, stalled.", ca.Client.Account, stream, consumer)
+			s.publishAdvisory(acc, JSAdvisoryConsumerQuorumLostPre+"."+stream+"."+consumer, &JSConsumerQuorumLostAdvisory{
+				TypedEvent: TypedEvent{
+					Type: JSConsumerQuorumLostAdvisoryType,
+					ID:   nuid.Next(),
+					Time: time.Now().UTC(),
+				},
+				Stream:   stream,
+				Consumer: consumer,
+				Replicas: s.replicas(node),
+			})
+		}
+	}
+
+	// Tell consumer to switch leader status.
+	o.setLeader(isLeader)
+
+	if !isLeader || hasResponded {
+		return
+	}
+
 	var resp = JSApiConsumerCreateResponse{ApiResponse: ApiResponse{Type: JSApiConsumerCreateResponseType}}
 	if err != nil {
 		resp.Error = jsError(err)
@@ -2682,7 +2755,7 @@ func (s *Server) jsClusteredConsumerRequest(ci *ClientInfo, subject, reply strin
 	} else {
 		oname = cfg.Durable
 		if sa.consumers[oname] != nil {
-			resp.Error = jsError(ErrJetStreamStreamAlreadyUsed)
+			resp.Error = jsError(ErrJetStreamConsumerAlreadyUsed)
 			s.sendAPIResponse(ci, acc, subject, reply, string(rmsg), s.jsonResponse(&resp))
 			return
 		}
@@ -2847,7 +2920,6 @@ func (mset *Stream) snapshot() []byte {
 
 // processClusteredMsg will propose the inbound message to the underlying raft group.
 func (mset *Stream) processClusteredInboundMsg(subject, reply string, hdr, msg []byte) error {
-
 	// For possible error response.
 	var response []byte
 
diff --git a/server/jetstream_events.go b/server/jetstream_events.go
index 00ec6c5a..19108f06 100644
--- a/server/jetstream_events.go
+++ b/server/jetstream_events.go
@@ -145,3 +145,51 @@ type JSRestoreCompleteAdvisory struct {
 
 // JSRestoreCompleteAdvisoryType is the schema type for JSSnapshotCreateAdvisory
 const JSRestoreCompleteAdvisoryType = "io.nats.jetstream.advisory.v1.restore_complete"
+
+// Clustering specific.
+
+// JSStreamLeaderElectedAdvisoryType is sent when the system elects a leader for a stream.
+const JSStreamLeaderElectedAdvisoryType = "io.nats.jetstream.advisory.v1.stream_leader_elected"
+
+// JSStreamQuorumLostAdvisory indicates that a stream has lost quorum and is stalled.
+type JSStreamLeaderElectedAdvisory struct {
+	TypedEvent
+	Stream   string      `json:"stream"`
+	Leader   string      `json:"leader"`
+	Replicas []*PeerInfo `json:"replicas"`
+}
+
+// JSStreamQuorumAdvisoryType is sent when the system detects a clustered stream and
+// its consumers are stalled and unable to make progress.
+const JSStreamQuorumLostAdvisoryType = "io.nats.jetstream.advisory.v1.stream_quorum_lost"
+
+// JSStreamQuorumLostAdvisory indicates that a stream has lost quorum and is stalled.
+type JSStreamQuorumLostAdvisory struct {
+	TypedEvent
+	Stream   string      `json:"stream"`
+	Replicas []*PeerInfo `json:"replicas"`
+}
+
+// JSConsumerLeaderElectedAdvisoryType is sent when the system elects a leader for a consumer.
+const JSConsumerLeaderElectedAdvisoryType = "io.nats.jetstream.advisory.v1.consumer_leader_elected"
+
+// JSStreamQuorumLostAdvisory indicates that a stream has lost quorum and is stalled.
+type JSConsumerLeaderElectedAdvisory struct {
+	TypedEvent
+	Stream   string      `json:"stream"`
+	Consumer string      `json:"consumer"`
+	Leader   string      `json:"leader"`
+	Replicas []*PeerInfo `json:"replicas"`
+}
+
+// JSConsumerQuorumAdvisoryType is sent when the system detects a clustered consumer and
+// is stalled and unable to make progress.
+const JSConsumerQuorumLostAdvisoryType = "io.nats.jetstream.advisory.v1.consumer_quorum_lost"
+
+// JSConsumerQuorumLostAdvisory indicates that a consumer has lost quorum and is stalled.
+type JSConsumerQuorumLostAdvisory struct {
+	TypedEvent
+	Stream   string      `json:"stream"`
+	Consumer string      `json:"consumer"`
+	Replicas []*PeerInfo `json:"replicas"`
+}
diff --git a/server/monitor.go b/server/monitor.go
index 6319f6fd..06c3ce09 100644
--- a/server/monitor.go
+++ b/server/monitor.go
@@ -1062,6 +1062,9 @@ type ClusterOptsVarz struct {
 	Port        int      `json:"cluster_port,omitempty"`
 	AuthTimeout float64  `json:"auth_timeout,omitempty"`
 	URLs        []string `json:"urls,omitempty"`
+	TLSTimeout  float64  `json:"tls_timeout,omitempty"`
+	TLSRequired bool     `json:"tls_required,omitempty"`
+	TLSVerify   bool     `json:"tls_verify,omitempty"`
 }
 
 // GatewayOptsVarz contains monitoring gateway information
@@ -1071,6 +1074,8 @@ type GatewayOptsVarz struct {
 	Port           int                     `json:"port,omitempty"`
 	AuthTimeout    float64                 `json:"auth_timeout,omitempty"`
 	TLSTimeout     float64                 `json:"tls_timeout,omitempty"`
+	TLSRequired    bool                    `json:"tls_required,omitempty"`
+	TLSVerify      bool                    `json:"tls_verify,omitempty"`
 	Advertise      string                  `json:"advertise,omitempty"`
 	ConnectRetries int                     `json:"connect_retries,omitempty"`
 	Gateways       []RemoteGatewayOptsVarz `json:"gateways,omitempty"`
@@ -1090,6 +1095,8 @@ type LeafNodeOptsVarz struct {
 	Port        int                  `json:"port,omitempty"`
 	AuthTimeout float64              `json:"auth_timeout,omitempty"`
 	TLSTimeout  float64              `json:"tls_timeout,omitempty"`
+	TLSRequired bool                 `json:"tls_required,omitempty"`
+	TLSVerify   bool                 `json:"tls_verify,omitempty"`
 	Remotes     []RemoteLeafOptsVarz `json:"remotes,omitempty"`
 }
 
@@ -1195,6 +1202,10 @@ func (s *Server) createVarz(pcpu float64, rss int64) *Varz {
 	c := &opts.Cluster
 	gw := &opts.Gateway
 	ln := &opts.LeafNode
+	clustTlsReq := c.TLSConfig != nil
+	gatewayTlsReq := gw.TLSConfig != nil
+	leafTlsReq := ln.TLSConfig != nil
+	leafTlsVerify := leafTlsReq && ln.TLSConfig.ClientAuth == tls.RequireAndVerifyClientCert
 	varz := &Varz{
 		ID:           info.ID,
 		Version:      info.Version,
@@ -1214,6 +1225,9 @@ func (s *Server) createVarz(pcpu float64, rss int64) *Varz {
 			Host:        c.Host,
 			Port:        c.Port,
 			AuthTimeout: c.AuthTimeout,
+			TLSTimeout:  c.TLSTimeout,
+			TLSRequired: clustTlsReq,
+			TLSVerify:   clustTlsReq,
 		},
 		Gateway: GatewayOptsVarz{
 			Name:           gw.Name,
@@ -1221,6 +1235,8 @@ func (s *Server) createVarz(pcpu float64, rss int64) *Varz {
 			Port:           gw.Port,
 			AuthTimeout:    gw.AuthTimeout,
 			TLSTimeout:     gw.TLSTimeout,
+			TLSRequired:    gatewayTlsReq,
+			TLSVerify:      gatewayTlsReq,
 			Advertise:      gw.Advertise,
 			ConnectRetries: gw.ConnectRetries,
 			Gateways:       []RemoteGatewayOptsVarz{},
@@ -1231,6 +1247,8 @@ func (s *Server) createVarz(pcpu float64, rss int64) *Varz {
 			Port:        ln.Port,
 			AuthTimeout: ln.AuthTimeout,
 			TLSTimeout:  ln.TLSTimeout,
+			TLSRequired: leafTlsReq,
+			TLSVerify:   leafTlsVerify,
 			Remotes:     []RemoteLeafOptsVarz{},
 		},
 		Start:    s.start,
diff --git a/server/raft.go b/server/raft.go
index 76c41733..ad16cdcd 100644
--- a/server/raft.go
+++ b/server/raft.go
@@ -37,6 +37,7 @@ type RaftNode interface {
 	State() RaftState
 	Size() (entries, bytes uint64)
 	Leader() bool
+	Quorum() bool
 	Current() bool
 	GroupLeader() string
 	StepDown() error
@@ -754,9 +755,6 @@ func (n *raft) Group() string {
 func (n *raft) Peers() []*Peer {
 	n.RLock()
 	defer n.RUnlock()
-	if n.state != Leader {
-		return nil
-	}
 
 	var peers []*Peer
 	for id, ps := range n.peers {
@@ -1196,9 +1194,30 @@ func (n *raft) runAsLeader() {
 	}
 }
 
+// Quorum reports the quorum status. Will be called on former leaders.
+func (n *raft) Quorum() bool {
+	n.RLock()
+	defer n.RUnlock()
+
+	now, nc := time.Now().UnixNano(), 1
+	for _, peer := range n.peers {
+		if now-peer.ts < int64(lostQuorumInterval) {
+			nc++
+			if nc >= n.qn {
+				return true
+			}
+		}
+	}
+	return false
+}
+
 func (n *raft) lostQuorum() bool {
 	n.RLock()
 	defer n.RUnlock()
+	return n.lostQuorumLocked()
+}
+
+func (n *raft) lostQuorumLocked() bool {
 	now, nc := time.Now().UnixNano(), 1
 	for _, peer := range n.peers {
 		if now-peer.ts < int64(lostQuorumInterval) {
@@ -2190,6 +2209,9 @@ func (n *raft) switchToCandidate() {
 	defer n.Unlock()
 	if n.state != Candidate {
 		n.notice("Switching to candidate")
+	} else if n.lostQuorumLocked() {
+		// We signal to the upper layers such that can alert on quorum lost.
+		n.updateLeadChange(false)
 	}
 	// Increment the term.
 	n.term++


Test patch:
diff --git a/server/monitor_test.go b/server/monitor_test.go
index 3955b7b9..69d8a5d1 100644
--- a/server/monitor_test.go
+++ b/server/monitor_test.go
@@ -15,6 +15,7 @@ package server
 
 import (
 	"bytes"
+	"crypto/tls"
 	"encoding/json"
 	"fmt"
 	"io/ioutil"
@@ -2093,6 +2094,62 @@ func TestConnzTLSInHandshake(t *testing.T) {
 	}
 }
 
+func TestConnzTLSCfg(t *testing.T) {
+	resetPreviousHTTPConnections()
+
+	tc := &TLSConfigOpts{}
+	tc.CertFile = "configs/certs/server.pem"
+	tc.KeyFile = "configs/certs/key.pem"
+
+	var err error
+	opts := DefaultMonitorOptions()
+	opts.NoSystemAccount = true
+	opts.TLSTimeout = 1.5 // 1.5 seconds
+	opts.TLSConfig, err = GenTLSConfig(tc)
+	require_NoError(t, err)
+	opts.TLSConfig.ClientAuth = tls.RequireAndVerifyClientCert
+	opts.Gateway.TLSConfig, err = GenTLSConfig(tc)
+	require_NoError(t, err)
+	opts.Gateway.TLSTimeout = 1.5
+	opts.LeafNode.TLSConfig, err = GenTLSConfig(tc)
+	require_NoError(t, err)
+	opts.LeafNode.TLSConfig.ClientAuth = tls.RequireAndVerifyClientCert
+	opts.LeafNode.TLSTimeout = 1.5
+	opts.Cluster.TLSConfig, err = GenTLSConfig(tc)
+	require_NoError(t, err)
+	opts.Cluster.TLSTimeout = 1.5
+
+	s := RunServer(opts)
+	defer s.Shutdown()
+
+	check := func(verify, required bool, timeout float64) {
+		t.Helper()
+		if !verify {
+			t.Fatalf("Expected tls_verify to be true")
+		}
+		if !required {
+			t.Fatalf("Expected tls_required to be true")
+		}
+		if timeout != 1.5 {
+			t.Fatalf("Expected tls_timeout to be 1.5")
+		}
+	}
+
+	start := time.Now()
+	endpoint := fmt.Sprintf("http://%s:%d/varz", opts.HTTPHost, s.MonitorAddr().Port)
+	for mode := 0; mode < 2; mode++ {
+		varz := pollVarz(t, s, mode, endpoint, nil)
+		duration := time.Since(start)
+		if duration >= 1500*time.Millisecond {
+			t.Fatalf("Looks like varz blocked on handshake, took %v", duration)
+		}
+		check(varz.TLSVerify, varz.TLSRequired, varz.TLSTimeout)
+		check(varz.Cluster.TLSVerify, varz.Cluster.TLSRequired, varz.Cluster.TLSTimeout)
+		check(varz.Gateway.TLSVerify, varz.Gateway.TLSRequired, varz.Gateway.TLSTimeout)
+		check(varz.LeafNode.TLSVerify, varz.LeafNode.TLSRequired, varz.LeafNode.TLSTimeout)
+	}
+}
+
 func TestServerIDs(t *testing.T) {
 	s := runMonitorServer()
 	defer s.Shutdown()
@@ -2425,6 +2482,9 @@ func TestMonitorCluster(t *testing.T) {
 		opts.Cluster.Port,
 		opts.Cluster.AuthTimeout,
 		[]string{"127.0.0.1:1234"},
+		opts.Cluster.TLSTimeout,
+		opts.Cluster.TLSConfig != nil,
+		opts.Cluster.TLSConfig != nil,
 	}
 
 	varzURL := fmt.Sprintf("http://127.0.0.1:%d/varz", s.MonitorAddr().Port)
@@ -2440,7 +2500,7 @@ func TestMonitorCluster(t *testing.T) {
 
 		// Having this here to make sure that if fields are added in ClusterOptsVarz,
 		// we make sure to update this test (compiler will report an error if we don't)
-		_ = ClusterOptsVarz{"", "", 0, 0, nil}
+		_ = ClusterOptsVarz{"", "", 0, 0, nil, 2, false, false}
 
 		// Alter the fields to make sure that we have a proper deep copy
 		// of what may be stored in the server. Anything we change here
@@ -2590,6 +2650,8 @@ func TestMonitorGateway(t *testing.T) {
 		opts.Gateway.Port,
 		opts.Gateway.AuthTimeout,
 		opts.Gateway.TLSTimeout,
+		opts.Gateway.TLSConfig != nil,
+		opts.Gateway.TLSConfig != nil,
 		opts.Gateway.Advertise,
 		opts.Gateway.ConnectRetries,
 		[]RemoteGatewayOptsVarz{{"B", 1, nil}},
@@ -2631,7 +2693,7 @@ func TestMonitorGateway(t *testing.T) {
 
 		// Having this here to make sure that if fields are added in GatewayOptsVarz,
 		// we make sure to update this test (compiler will report an error if we don't)
-		_ = GatewayOptsVarz{"", "", 0, 0, 0, "", 0, []RemoteGatewayOptsVarz{{"", 0, nil}}, false}
+		_ = GatewayOptsVarz{"", "", 0, 0, 0, false, false, "", 0, []RemoteGatewayOptsVarz{{"", 0, nil}}, false}
 
 		// Alter the fields to make sure that we have a proper deep copy
 		// of what may be stored in the server. Anything we change here
@@ -2756,6 +2818,8 @@ func TestMonitorLeafNode(t *testing.T) {
 		opts.LeafNode.Port,
 		opts.LeafNode.AuthTimeout,
 		opts.LeafNode.TLSTimeout,
+		opts.LeafNode.TLSConfig != nil,
+		opts.LeafNode.TLSConfig != nil,
 		[]RemoteLeafOptsVarz{
 			{
 				"acc", 1, []string{"localhost:1234"},
@@ -2777,7 +2841,7 @@ func TestMonitorLeafNode(t *testing.T) {
 
 		// Having this here to make sure that if fields are added in ClusterOptsVarz,
 		// we make sure to update this test (compiler will report an error if we don't)
-		_ = LeafNodeOptsVarz{"", 0, 0, 0, []RemoteLeafOptsVarz{{"", 0, nil}}}
+		_ = LeafNodeOptsVarz{"", 0, 0, 0, false, false, []RemoteLeafOptsVarz{{"", 0, nil}}}
 
 		// Alter the fields to make sure that we have a proper deep copy
 		// of what may be stored in the server. Anything we change here
diff --git a/test/jetstream_cluster_test.go b/test/jetstream_cluster_test.go
index 36b42e31..98287ab7 100644
--- a/test/jetstream_cluster_test.go
+++ b/test/jetstream_cluster_test.go
@@ -169,11 +169,12 @@ func TestJetStreamClusterSingleReplicaStreams(t *testing.T) {
 		t.Fatalf("ConsumerInfo is not correct %+v", ci)
 	}
 
-	// Now make sure that if we kill and restart the server that is the stream and consumer leader they return.
+	// Now make sure that if we kill and restart the server that this stream and consumer return.
 	sl := c.streamLeader("$G", "TEST")
 	sl.Shutdown()
 	c.restartServer(sl)
 	c.waitOnNewStreamLeader("$G", "TEST")
+	time.Sleep(500 * time.Millisecond)
 
 	si, err = js.StreamInfo("TEST")
 	if err != nil {
@@ -184,7 +185,8 @@ func TestJetStreamClusterSingleReplicaStreams(t *testing.T) {
 	}
 	// Now durable consumer.
 	c.waitOnNewConsumerLeader("$G", "TEST", "dlc")
-	time.Sleep(200 * time.Millisecond)
+	time.Sleep(500 * time.Millisecond)
+
 	if _, err = js.ConsumerInfo("TEST", "dlc"); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -247,6 +249,56 @@ func TestJetStreamClusterMultiReplicaStreams(t *testing.T) {
 	}
 }
 
+func TestJetStreamClusterMemoryStore(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3M", 3)
+	defer c.shutdown()
+
+	// Client based API
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	// FIXME(dlc) - This should be default.
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo", "bar"},
+		Replicas: 3,
+		Storage:  nats.MemoryStorage,
+	})
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	// Send in 100 messages.
+	msg, toSend := []byte("Hello MemoryStore"), 100
+	for i := 0; i < toSend; i++ {
+		if _, err = js.Publish("foo", msg); err != nil {
+			t.Fatalf("Unexpected publish error: %v", err)
+		}
+	}
+	// Now grab info for this stream.
+	si, err := js.StreamInfo("TEST")
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if si == nil || si.Config.Name != "TEST" {
+		t.Fatalf("StreamInfo is not correct %+v", si)
+	}
+	if si.Cluster == nil || len(si.Cluster.Replicas) != 2 {
+		t.Fatalf("Cluster info is incorrect: %+v", si.Cluster)
+	}
+	// Check active state as well, shows that the owner answered.
+	if si.State.Msgs != uint64(toSend) {
+		t.Fatalf("Expected %d msgs, got bad state: %+v", toSend, si.State)
+	}
+	// Do a normal sub.
+	sub, err := js.SubscribeSync("foo")
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
+	checkSubsPending(t, sub, toSend)
+}
+
 func TestJetStreamClusterCompaction(t *testing.T) {
 	// This test takes a long time to observe compactions.
 	// Once moved to server we can adjust and re-enable.
@@ -2233,10 +2285,29 @@ func TestJetStreamClusterNoQuorumStepdown(t *testing.T) {
 	nc, js := jsClientConnect(t, s)
 	defer nc.Close()
 
+	// Setup subscription for leader elected.
+	lesub, err := nc.SubscribeSync(server.JSAdvisoryStreamLeaderElectedPre + ".*")
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
 	if _, err := js.AddStream(&nats.StreamConfig{Name: "NO-Q", Replicas: 2}); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
 
+	// Make sure we received our leader elected advisory.
+	leadv, _ := lesub.NextMsg(0)
+	if leadv == nil {
+		t.Fatalf("Expected to receive a leader elected advisory")
+	}
+	var le server.JSStreamLeaderElectedAdvisory
+	if err := json.Unmarshal(leadv.Data, &le); err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if ln := c.streamLeader("$G", "NO-Q").Name(); le.Leader != ln {
+		t.Fatalf("Expected to have leader %q in elect advisory, got %q", ln, le.Leader)
+	}
+
 	payload := []byte("Hello JSC")
 	for i := 0; i < 10; i++ {
 		if _, err := js.Publish("NO-Q", payload); err != nil {
@@ -2244,6 +2315,12 @@ func TestJetStreamClusterNoQuorumStepdown(t *testing.T) {
 		}
 	}
 
+	// Setup subscription for leader elected.
+	clesub, err := nc.SubscribeSync(server.JSAdvisoryConsumerLeaderElectedPre + ".*.*")
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+
 	sub, err := js.SubscribeSync("NO-Q")
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -2253,10 +2330,27 @@ func TestJetStreamClusterNoQuorumStepdown(t *testing.T) {
 		t.Fatalf("Unexpected error: %v", err)
 	}
 
+	// Make sure we received our consumer leader elected advisory.
+	leadv, _ = clesub.NextMsg(0)
+	if leadv == nil {
+		t.Fatalf("Expected to receive a consumer leader elected advisory")
+	}
+
+	// Setup subscriptions for lost quorum advisory.
+	ssub, err := nc.SubscribeSync(server.JSAdvisoryStreamQuorumLostPre + ".*")
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	csub, err := nc.SubscribeSync(server.JSAdvisoryConsumerQuorumLostPre + ".*.*")
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	nc.Flush()
+
 	// Shutdown the non-leader.
 	c.randomNonStreamLeader("$G", "NO-Q").Shutdown()
 
-	// This should eventually have us stepdown as leader since we would have lost quorum.
+	// This should eventually have us stepdown as leader since we would have lost quorum with R=2.
 	checkFor(t, 2*time.Second, 100*time.Millisecond, func() error {
 		if sl := c.streamLeader("$G", "NO-Q"); sl == nil {
 			return nil
@@ -2287,6 +2381,31 @@ func TestJetStreamClusterNoQuorumStepdown(t *testing.T) {
 		t.Fatalf("Expected an 'unavailable' error, got %v", err)
 	}
 
+	// Make sure we received our lost quorum advisories.
+	adv, _ := ssub.NextMsg(0)
+	if adv == nil {
+		t.Fatalf("Expected to receive a stream quorum lost advisory")
+	}
+	var lqa server.JSStreamQuorumLostAdvisory
+	if err := json.Unmarshal(adv.Data, &lqa); err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if len(lqa.Replicas) != 2 {
+		t.Fatalf("Expected reports for both replicas, only got %d", len(lqa.Replicas))
+	}
+	// Consumer too.
+	adv, _ = csub.NextMsg(time.Second)
+	if adv == nil {
+		t.Fatalf("Expected to receive a consumer quorum lost advisory")
+	}
+	var clqa server.JSConsumerQuorumLostAdvisory
+	if err := json.Unmarshal(adv.Data, &clqa); err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if len(clqa.Replicas) != 2 {
+		t.Fatalf("Expected reports for both replicas, only got %d", len(clqa.Replicas))
+	}
+
 	// Now let's take out the other non meta-leader server.
 	// We should get same error for general API calls.
 	c.randomNonLeader().Shutdown()
@@ -2545,7 +2664,7 @@ func (c *cluster) waitOnNewConsumerLeader(account, stream, consumer string) {
 	expires := time.Now().Add(10 * time.Second)
 	for time.Now().Before(expires) {
 		if leader := c.consumerLeader(account, stream, consumer); leader != nil {
-			time.Sleep(50 * time.Millisecond)
+			time.Sleep(100 * time.Millisecond)
 			return
 		}
 		time.Sleep(100 * time.Millisecond)
@@ -2568,7 +2687,7 @@ func (c *cluster) waitOnNewStreamLeader(account, stream string) {
 	expires := time.Now().Add(10 * time.Second)
 	for time.Now().Before(expires) {
 		if leader := c.streamLeader(account, stream); leader != nil {
-			time.Sleep(50 * time.Millisecond)
+			time.Sleep(100 * time.Millisecond)
 			return
 		}
 		time.Sleep(100 * time.Millisecond)
@@ -2601,7 +2720,7 @@ func (c *cluster) waitOnStreamCurrent(s *server.Server, account, stream string)
 	expires := time.Now().Add(10 * time.Second)
 	for time.Now().Before(expires) {
 		if s.JetStreamIsStreamCurrent(account, stream) {
-			time.Sleep(50 * time.Millisecond)
+			time.Sleep(100 * time.Millisecond)
 			return
 		}
 		time.Sleep(100 * time.Millisecond)
@@ -2614,7 +2733,7 @@ func (c *cluster) waitOnServerCurrent(s *server.Server) {
 	expires := time.Now().Add(5 * time.Second)
 	for time.Now().Before(expires) {
 		if s.JetStreamIsCurrent() {
-			time.Sleep(50 * time.Millisecond)
+			time.Sleep(100 * time.Millisecond)
 			return
 		}
 		time.Sleep(100 * time.Millisecond)
@@ -2684,6 +2803,7 @@ func (c *cluster) waitOnClusterReady() {
 	// Now make sure we have all peers.
 	for leader != nil && time.Now().Before(expires) {
 		if len(leader.JetStreamClusterPeers()) == len(c.servers) {
+			time.Sleep(50 * time.Millisecond)
 			return
 		}
 		time.Sleep(50 * time.Millisecond)