Skip to content

healthcheck: update healthy tablets correctly when a stream returns an error or times out #7654

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

Merged
deepthi merged 3 commits into from
Mar 13, 2021
Merged

healthcheck: update healthy tablets correctly when a stream returns an error or times out #7654

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
a1b7a1f
healthcheck: update healthy tablets correctly when a stream returns a…
deepthi Mar 10, 2021
7eb7228
healthcheck: remove PRIMARY from healthy list when there's an error o…
deepthi Mar 11, 2021
f3a432e
healthcheck: reorganize code to be more readable
deepthi Mar 12, 2021
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
60 changes: 32 additions & 28 deletions go/vt/discovery/healthcheck.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -404,17 +404,17 @@ func (hc *HealthCheckImpl) deleteTablet(tablet *topodata.Tablet) {
}
}

func (hc *HealthCheckImpl) updateHealth(th *TabletHealth, shr *query.StreamHealthResponse, currentTarget *query.Target, trivialNonMasterUpdate bool, isMasterUpdate bool, isMasterChange bool) {
func (hc *HealthCheckImpl) updateHealth(th *TabletHealth, currentTarget *query.Target, trivialUpdate bool, isPrimaryUp bool) {

@sougou sougou Mar 11, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Rename this to isPrimaryUpdate? Otherwise, I was reading it as "is primary up".

@deepthi deepthi Mar 11, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

It is supposed to be "is primary up". The old parameters (isMasterChange and isMasterUpdate) were actually unnecessary because all the information they pass in is available within the scope of this func. However, we do need to know whether the primary should be marked unhealthy (if there is an error/timeout on the healthcheck connection), so I introduced this parameter.

// hc.healthByAlias is authoritative, it should be updated
hc.mu.Lock()
defer hc.mu.Unlock()

tabletAlias := tabletAliasString(topoproto.TabletAliasString(shr.TabletAlias))

hcErrorCounters.Add([]string{shr.Target.Keyspace, shr.Target.Shard, topoproto.TabletTypeLString(shr.Target.TabletType)}, 0)
targetKey := hc.keyFromTarget(shr.Target)
targetChanged := currentTarget.TabletType != shr.Target.TabletType || currentTarget.Keyspace != shr.Target.Keyspace || currentTarget.Shard != shr.Target.Shard
tabletAlias := tabletAliasString(topoproto.TabletAliasString(th.Tablet.Alias))
targetKey := hc.keyFromTarget(th.Target)
targetChanged := currentTarget.TabletType != th.Target.TabletType || currentTarget.Keyspace != th.Target.Keyspace || currentTarget.Shard != th.Target.Shard
if targetChanged {
// Error counter has to be set here in case we get a new tablet type for the first time in a stream response
hcErrorCounters.Add([]string{th.Target.Keyspace, th.Target.Shard, topoproto.TabletTypeLString(th.Target.TabletType)}, 0)
// keyspace and shard are not expected to change, but just in case ...
// move this tabletHealthCheck to the correct map
oldTargetKey := hc.keyFromTarget(currentTarget)
Expand All @@ -427,44 +427,48 @@ func (hc *HealthCheckImpl) updateHealth(th *TabletHealth, shr *query.StreamHealt
// add it to the map by target
hc.healthData[targetKey][tabletAlias] = th

if isMasterUpdate {
if len(hc.healthy[targetKey]) == 0 {
hc.healthy[targetKey] = append(hc.healthy[targetKey], th)
} else {
// We already have one up server, see if we
// need to replace it.
if shr.TabletExternallyReparentedTimestamp < hc.healthy[targetKey][0].MasterTermStartTime {
log.Warningf("not marking healthy master %s as Up for %s because its MasterTermStartTime is smaller than the highest known timestamp from previous MASTERs %s: %d < %d ",
topoproto.TabletAliasString(shr.TabletAlias),
topoproto.KeyspaceShardString(shr.Target.Keyspace, shr.Target.Shard),
topoproto.TabletAliasString(hc.healthy[targetKey][0].Tablet.Alias),
shr.TabletExternallyReparentedTimestamp,
hc.healthy[targetKey][0].MasterTermStartTime)
if th.Target.TabletType == topodata.TabletType_MASTER {

@sougou sougou Mar 11, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

I would presume that isPrimaryUp would be true only if the tablet type was MASTER. Is there a case where this is not the case?

@deepthi deepthi Mar 11, 2021 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

That is a good point. I suppose this is more of a consistency check - to make sure that we still behave correctly if a caller passes in isPrimaryUp true for a non-master tablet type.

if isPrimaryUp {
if len(hc.healthy[targetKey]) == 0 {
hc.healthy[targetKey] = append(hc.healthy[targetKey], th)
} else {
// Just replace it.
hc.healthy[targetKey][0] = th
// We already have one up server, see if we
// need to replace it.
if th.MasterTermStartTime < hc.healthy[targetKey][0].MasterTermStartTime {
log.Warningf("not marking healthy master %s as Up for %s because its MasterTermStartTime is smaller than the highest known timestamp from previous MASTERs %s: %d < %d ",
topoproto.TabletAliasString(th.Tablet.Alias),
topoproto.KeyspaceShardString(th.Target.Keyspace, th.Target.Shard),
topoproto.TabletAliasString(hc.healthy[targetKey][0].Tablet.Alias),
th.MasterTermStartTime,
hc.healthy[targetKey][0].MasterTermStartTime)
} else {
// Just replace it.
hc.healthy[targetKey][0] = th
}
}
} else {

@sougou sougou Mar 11, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

If this section was the reason you added the MASTER check, it may read better if you explicitly checked for it here. Something like th.Target.TabletType == topodata.TabletType_MASTER && !isPrimaryUp. But I'm not sure if that means the same thing.

@deepthi deepthi Mar 11, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

The MASTER check was already there but in the form of a boolean (isMasterUpdate).
I can break this into two separate if blocks instead of an if .. if .. else if that is easier to read/understand. What you are suggesting will be equivalent to how it is written today.

@sougou sougou Mar 11, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

The concern was the deep nesting which made it non-obvious. A cascading if (or switch) may read better. Try it. If it doesn't improve it, we can keep this as is.

@deepthi deepthi Mar 12, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Ok, I think the switch made it better.

// No healthy master tablet
hc.healthy[targetKey] = []*TabletHealth{}
}
}
if !trivialNonMasterUpdate {
if !trivialUpdate {
// We re-sort the healthy tablet list whenever we get a health update for tablets we can route to.
// Tablets from other cells for non-master targets should not trigger a re-sort;
// they should also be excluded from healthy list.
if shr.Target.TabletType != topodata.TabletType_MASTER && hc.isIncluded(shr.Target.TabletType, shr.TabletAlias) {
if th.Target.TabletType != topodata.TabletType_MASTER && hc.isIncluded(th.Target.TabletType, th.Tablet.Alias) {
hc.recomputeHealthy(targetKey)
}
if targetChanged && currentTarget.TabletType != topodata.TabletType_MASTER && hc.isIncluded(shr.Target.TabletType, shr.TabletAlias) { // also recompute old target's healthy list
if targetChanged && currentTarget.TabletType != topodata.TabletType_MASTER && hc.isIncluded(th.Target.TabletType, th.Tablet.Alias) { // also recompute old target's healthy list
oldTargetKey := hc.keyFromTarget(currentTarget)
hc.recomputeHealthy(oldTargetKey)
}
}
if isMasterChange {
log.Errorf("Adding 1 to MasterPromoted counter for tablet: %v, shr.Tablet: %v, shr.TabletType: %v", currentTarget, topoproto.TabletAliasString(shr.TabletAlias), shr.Target.TabletType)
hcMasterPromotedCounters.Add([]string{shr.Target.Keyspace, shr.Target.Shard}, 1)
if currentTarget.TabletType != topodata.TabletType_MASTER && th.Target.TabletType == topodata.TabletType_MASTER {
log.Errorf("Adding 1 to MasterPromoted counter for target: %v, tablet: %v, tabletType: %v", currentTarget, topoproto.TabletAliasString(th.Tablet.Alias), th.Target.TabletType)
hcMasterPromotedCounters.Add([]string{th.Target.Keyspace, th.Target.Shard}, 1)
}
// broadcast to subscribers
hc.broadcast(th)

}

func (hc *HealthCheckImpl) recomputeHealthy(key keyspaceShardTabletType) {
Expand Down
71 changes: 71 additions & 0 deletions go/vt/discovery/healthcheck_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,6 +65,8 @@ func init() {
}

func TestHealthCheck(t *testing.T) {
// reset error counters
hcErrorCounters.ResetAll()
ts := memorytopo.NewServer("cell")
hc := createTestHc(ts)
// close healthcheck
Expand Down Expand Up @@ -257,6 +259,69 @@ func TestHealthCheckStreamError(t *testing.T) {
result = <-resultChan
//TODO: figure out how to compare objects that contain errors using utils.MustMatch
assert.True(t, want.DeepEqual(result), "Wrong TabletHealth data\n Expected: %v\n Actual: %v", want, result)
// tablet should be removed from healthy list
a := hc.GetHealthyTabletStats(&querypb.Target{Keyspace: "k", Shard: "s", TabletType: topodatapb.TabletType_REPLICA})
assert.Empty(t, a, "wrong result, expected empty list")
}

// TestHealthCheckErrorOnPrimary is the same as TestHealthCheckStreamError except for tablet type
func TestHealthCheckErrorOnPrimary(t *testing.T) {
ts := memorytopo.NewServer("cell")
hc := createTestHc(ts)
defer hc.Close()

tablet := createTestTablet(0, "cell", "a")
input := make(chan *querypb.StreamHealthResponse)
resultChan := hc.Subscribe()
fc := createFakeConn(tablet, input)
fc.errCh = make(chan error)
hc.AddTablet(tablet)

// Immediately after AddTablet() there will be the first notification.
want := &TabletHealth{
Tablet: tablet,
Target: &querypb.Target{Keyspace: "k", Shard: "s"},
Serving: false,
MasterTermStartTime: 0,
}
result := <-resultChan
mustMatch(t, want, result, "Wrong TabletHealth data")

// one tablet after receiving a StreamHealthResponse
shr := &querypb.StreamHealthResponse{
TabletAlias: tablet.Alias,
Target: &querypb.Target{Keyspace: "k", Shard: "s", TabletType: topodatapb.TabletType_MASTER},
Serving: true,
TabletExternallyReparentedTimestamp: 10,
RealtimeStats: &querypb.RealtimeStats{SecondsBehindMaster: 1, CpuUsage: 0.2},
}
want = &TabletHealth{
Tablet: tablet,
Target: &querypb.Target{Keyspace: "k", Shard: "s", TabletType: topodatapb.TabletType_MASTER},
Serving: true,
Stats: &querypb.RealtimeStats{SecondsBehindMaster: 1, CpuUsage: 0.2},
MasterTermStartTime: 10,
}
input <- shr
result = <-resultChan
mustMatch(t, want, result, "Wrong TabletHealth data")

// Stream error
fc.errCh <- fmt.Errorf("some stream error")
want = &TabletHealth{
Tablet: tablet,
Target: &querypb.Target{Keyspace: "k", Shard: "s", TabletType: topodatapb.TabletType_MASTER},
Serving: false,
Stats: &querypb.RealtimeStats{SecondsBehindMaster: 1, CpuUsage: 0.2},
MasterTermStartTime: 10,
LastError: fmt.Errorf("some stream error"),
}
result = <-resultChan
//TODO: figure out how to compare objects that contain errors using utils.MustMatch
assert.True(t, want.DeepEqual(result), "Wrong TabletHealth data\n Expected: %v\n Actual: %v", want, result)
// tablet should be removed from healthy list
a := hc.GetHealthyTabletStats(&querypb.Target{Keyspace: "k", Shard: "s", TabletType: topodatapb.TabletType_MASTER})
assert.Empty(t, a, "wrong result, expected empty list")
}

func TestHealthCheckVerifiesTabletAlias(t *testing.T) {
Expand Down Expand Up @@ -363,6 +428,8 @@ func TestHealthCheckCloseWaitsForGoRoutines(t *testing.T) {
}

func TestHealthCheckTimeout(t *testing.T) {
// reset counters
hcErrorCounters.ResetAll()
ts := memorytopo.NewServer("cell")
hc := createTestHc(ts)
hc.healthCheckTimeout = 500 * time.Millisecond
Expand Down Expand Up @@ -410,6 +477,10 @@ func TestHealthCheckTimeout(t *testing.T) {
assert.Nil(t, checkErrorCounter("k", "s", topodatapb.TabletType_REPLICA, 1))
assert.True(t, fc.isCanceled(), "StreamHealth should be canceled after timeout, but is not")

// tablet should be removed from healthy list
a := hc.GetHealthyTabletStats(&querypb.Target{Keyspace: "k", Shard: "s", TabletType: topodatapb.TabletType_REPLICA})
assert.Empty(t, a, "wrong result, expected empty list")

// repeat the wait. It will timeout one more time trying to get the connection.
fc.resetCanceledFlag()
time.Sleep(hc.healthCheckTimeout)
Expand Down
19 changes: 12 additions & 7 deletions go/vt/discovery/tablet_health_check.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -186,10 +186,8 @@ func (thc *tabletHealthCheck) processResponse(hc *HealthCheckImpl, shr *query.St

currentTarget := thc.Target
// check whether this is a trivial update so as to update healthy map
trivialNonMasterUpdate := thc.LastError == nil && thc.Serving && shr.RealtimeStats.HealthError == "" && shr.Serving &&
trivialUpdate := thc.LastError == nil && thc.Serving && shr.RealtimeStats.HealthError == "" && shr.Serving &&
currentTarget.TabletType != topodata.TabletType_MASTER && currentTarget.TabletType == shr.Target.TabletType && thc.isTrivialReplagChange(shr.RealtimeStats)
isMasterUpdate := shr.Target.TabletType == topodata.TabletType_MASTER
isMasterChange := thc.Target.TabletType != topodata.TabletType_MASTER && shr.Target.TabletType == topodata.TabletType_MASTER
thc.lastResponseTimestamp = time.Now()
thc.Target = shr.Target
thc.MasterTermStartTime = shr.TabletExternallyReparentedTimestamp
Expand All @@ -202,7 +200,7 @@ func (thc *tabletHealthCheck) processResponse(hc *HealthCheckImpl, shr *query.St
thc.setServingState(serving, reason)

// notify downstream for master change
hc.updateHealth(thc.SimpleCopy(), shr, currentTarget, trivialNonMasterUpdate, isMasterUpdate, isMasterChange)
hc.updateHealth(thc.SimpleCopy(), currentTarget, trivialUpdate, true)
return nil
}

Expand Down Expand Up @@ -241,6 +239,9 @@ func (thc *tabletHealthCheck) checkConn(hc *HealthCheckImpl) {
hc.connsWG.Done()
}()

// Initialize error counter
hcErrorCounters.Add([]string{thc.Target.Keyspace, thc.Target.Shard, topoproto.TabletTypeLString(thc.Target.TabletType)}, 0)

retryDelay := hc.retryDelay
for {
streamCtx, streamCancel := context.WithCancel(thc.ctx)
Expand Down Expand Up @@ -287,12 +288,14 @@ func (thc *tabletHealthCheck) checkConn(hc *HealthCheckImpl) {
streamCancel()

if err != nil {
hcErrorCounters.Add([]string{thc.Target.Keyspace, thc.Target.Shard, topoproto.TabletTypeLString(thc.Target.TabletType)}, 1)
if strings.Contains(err.Error(), "health stats mismatch") {
hc.deleteTablet(thc.Tablet)
return
}
res := thc.SimpleCopy()
hc.broadcast(res)
// trivialUpdate = false because this is an error
// isPrimaryUp = false because we did not get a healthy response
hc.updateHealth(thc.SimpleCopy(), thc.Target, false, false)
}
// If there was a timeout send an error. We do this after stream has returned.
// This will ensure that this update prevails over any previous message that
Expand All @@ -301,7 +304,9 @@ func (thc *tabletHealthCheck) checkConn(hc *HealthCheckImpl) {
thc.LastError = fmt.Errorf("healthcheck timed out (latest %v)", thc.lastResponseTimestamp)
thc.setServingState(false, thc.LastError.Error())
hcErrorCounters.Add([]string{thc.Target.Keyspace, thc.Target.Shard, topoproto.TabletTypeLString(thc.Target.TabletType)}, 1)
hc.broadcast(thc.SimpleCopy())
// trivialUpdate = false because this is an error
// isPrimaryUp = false because we did not get a healthy response within the timeout
hc.updateHealth(thc.SimpleCopy(), thc.Target, false, false)
}

// Streaming RPC failed e.g. because vttablet was restarted or took too long.
Expand Down