gogrlx/nats-server
1
0
Fork 0
mirror of https://github.com/gogrlx/nats-server.git synced 2026-04-02 03:38:42 -07:00
Code Issues Packages Projects Releases Wiki Activity

Fix for datarace in healthcheck

Browse Source 
Signed-off-by: Derek Collison <derek@nats.io>
This commit is contained in:
Derek Collison
2023-04-02 16:30:13 -07:00
parent d5ac4d283a
commit ff3f102cdd
2 changed files with 12 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
server/jetstream_cluster.go
View File

@@ -436,8 +436,11 @@ func (cc *jetStreamCluster) isStreamCurrent(account, stream string) bool {
// isStreamHealthy will determine if the stream is up to date or very close.
// For R1 it will make sure the stream is present on this server.
// Read lock should be held.
func (cc *jetStreamCluster) isStreamHealthy(account, stream string) bool {
func (js *jetStream) isStreamHealthy(account, stream string) bool {
js.mu.RLock()
defer js.mu.RUnlock()
cc := js.cluster
if cc == nil {
// Non-clustered mode
return true
@@ -477,8 +480,11 @@ func (cc *jetStreamCluster) isStreamHealthy(account, stream string) bool {
// isConsumerCurrent will determine if the consumer is up to date.
// For R1 it will make sure the consunmer is present on this server.
// Read lock should be held.
func (cc *jetStreamCluster) isConsumerCurrent(account, stream, consumer string) bool {
func (js *jetStream) isConsumerCurrent(account, stream, consumer string) bool {
js.mu.RLock()
defer js.mu.RUnlock()
cc := js.cluster
if cc == nil {
// Non-clustered mode
return true

4
server/monitor.go
View File

@@ -3129,7 +3129,7 @@ func (s *Server) healthz(opts *HealthzOptions) *HealthStatus {
for stream, sa := range asa {
if sa.Group.isMember(ourID) {
// Make sure we can look up
if !cc.isStreamHealthy(acc, stream) {
if !js.isStreamHealthy(acc, stream) {
health.Status = na
health.Error = fmt.Sprintf("JetStream stream '%s > %s' is not current", acc, stream)
return health
@@ -3137,7 +3137,7 @@ func (s *Server) healthz(opts *HealthzOptions) *HealthStatus {
// Now check consumers.
for consumer, ca := range sa.consumers {
if ca.Group.isMember(ourID) {
if !cc.isConsumerCurrent(acc, stream, consumer) {
if !js.isConsumerCurrent(acc, stream, consumer) {
health.Status = na
health.Error = fmt.Sprintf("JetStream consumer '%s > %s > %s' is not current", acc, stream, consumer)
return health