Skip to content

VReplication: Improve Error/Status Reporting #12052

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
mattlord merged 11 commits into from
Jan 12, 2023
Merged

VReplication: Improve Error/Status Reporting #12052

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
a7c5080
Correct progress reporting when vstreams have not started
mattlord Jan 6, 2023
ec67dcc
Additional error handling/progress improvements
mattlord Jan 6, 2023
7e1d6f8
Add unit test
mattlord Jan 7, 2023
7e820dd
Include error message in Show/Progress stream states
mattlord Jan 7, 2023
da59866
Allow the test to pass in the CI while it's a WiP
mattlord Jan 7, 2023
fc0895c
Get unit tests working
mattlord Jan 8, 2023
c487d1c
Minor changes after self review
mattlord Jan 8, 2023
be50843
Add additional unrecoverable errors
mattlord Jan 11, 2023
d808374
Fix unit tests
mattlord Jan 11, 2023
3a9d882
Fix minor typo in dry run output
mattlord Jan 11, 2023
ddf0c8a
Address review comments
mattlord Jan 12, 2023
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
20 changes: 10 additions & 10 deletions go/mysql/constants.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -368,6 +368,7 @@ const (
ERServerShutdown = 1053

// not found
ERDbDropExists = 1008
Copy link
Copy Markdown
Member Author

@mattlord mattlord Jan 11, 2023 •
edited
Loading

Choose a reason for hiding this comment

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

Changes in this file were just me ordering the subgroups by error code as it makes it easier to walk through all of them as I did to see what ones should be in the vreplication.unRecoverableError() list.

ERCantFindFile = 1017
ERFormNotFound = 1029
ERKeyNotFound = 1032
Expand All @@ -379,7 +380,6 @@ const (
ERNoSuchTable = 1146
ERNonExistingTableGrant = 1147
ERKeyDoesNotExist = 1176
ERDbDropExists = 1008

// permissions
ERDBAccessDenied = 1044
Expand Down Expand Up @@ -407,6 +407,7 @@ const (
ERRowIsReferenced = 1217
ERCantUpdateWithReadLock = 1223
ERNoDefault = 1230
ERMasterFatalReadingBinlog = 1236
EROperandColumns = 1241
ERSubqueryNo1Row = 1242
ERWarnDataOutOfRange = 1264
Expand All @@ -415,19 +416,18 @@ const (
EROptionPreventsStatement = 1290
ERDuplicatedValueInType = 1291
ERSPDoesNotExist = 1305
ERNoDefaultForField = 1364
ErSPNotVarArg = 1414
ERRowIsReferenced2 = 1451
ErNoReferencedRow2 = 1452
ErSPNotVarArg = 1414
ERInnodbReadOnly = 1874
ERMasterFatalReadingBinlog = 1236
ERNoDefaultForField = 1364

// already exists
ERDbCreateExists = 1007
ERTableExists = 1050
ERDupEntry = 1062
ERFileExists = 1086
ERUDFExists = 1125
ERDbCreateExists = 1007

// aborted
ERGotSignal = 1078
Expand Down Expand Up @@ -513,7 +513,11 @@ const (
ERWrongFKDef = 1239
ERKeyRefDoNotMatchTableRef = 1240
ERCyclicReference = 1245
ERIllegalReference = 1247
ERDerivedMustHaveAlias = 1248
ERTableNameNotAllowedHere = 1250
ERCollationCharsetMismatch = 1253
ERWarnDataTruncated = 1265
ERCantAggregate2Collations = 1267
ERCantAggregate3Collations = 1270
ERCantAggregateNCollations = 1271
Expand All @@ -527,16 +531,13 @@ const (
ERInvalidOnUpdate = 1294
ERUnknownTimeZone = 1298
ERInvalidCharacterString = 1300
ERIllegalReference = 1247
ERDerivedMustHaveAlias = 1248
ERTableNameNotAllowedHere = 1250
ERQueryInterrupted = 1317
ERTruncatedWrongValueForField = 1366
ERIllegalValueForType = 1367
ERDataTooLong = 1406
ErrWrongValueForType = 1411
ERWarnDataTruncated = 1265
ERForbidSchemaChange = 1450
ERWrongValue = 1525
ERDataOutOfRange = 1690
ERInvalidJSONText = 3140
ERInvalidJSONTextInParams = 3141
Expand All @@ -545,7 +546,6 @@ const (
ERInvalidCastToJSON = 3147
ERJSONValueTooBig = 3150
ERJSONDocumentTooDeep = 3157
ERWrongValue = 1525

// max execution time exceeded
ERQueryTimeout = 3024
Expand Down
4 changes: 2 additions & 2 deletions go/test/endtoend/vreplication/vreplication_test_env.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@ var dryRunResultsSwitchWritesCustomerShard = []string{
"Lock keyspace customer",
"Stop writes on keyspace product, tables [Lead,Lead-1,customer,db_order_test]:",
"/ Keyspace product, Shard 0 at Position",
"Wait for VReplication on stopped streams to catchup for upto 30s",
"Wait for VReplication on stopped streams to catchup for up to 30s",
"Create reverse replication workflow p2c_reverse",
"Create journal entries on source databases",
"Enable writes on keyspace customer tables [Lead,Lead-1,customer,db_order_test]",
Expand Down Expand Up @@ -60,7 +60,7 @@ var dryRunResultsSwitchWritesM2m3 = []string{
"Stop writes on keyspace merchant-type, tables [/.*]:",
"/ Keyspace merchant-type, Shard -80 at Position",
"/ Keyspace merchant-type, Shard 80- at Position",
"Wait for VReplication on stopped streams to catchup for upto 30s",
"Wait for VReplication on stopped streams to catchup for up to 30s",
"Create reverse replication workflow m2m3_reverse",
"Create journal entries on source databases",
"Enable writes on keyspace merchant-type tables [/.*]",
Expand Down
40 changes: 23 additions & 17 deletions go/vt/vtctl/vtctl.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2180,22 +2180,28 @@ func commandVRWorkflow(ctx context.Context, wr *wrangler.Wrangler, subFlags *pfl
if err != nil {
return err
}
s += fmt.Sprintf("Following vreplication streams are running for workflow %s.%s:\n\n", target, workflowName)
s += fmt.Sprintf("The following vreplication streams exist for workflow %s.%s:\n\n", target, workflowName)
for ksShard := range res.ShardStatuses {
statuses := res.ShardStatuses[ksShard].PrimaryReplicationStatuses
for _, st := range statuses {
now := time.Now().Nanosecond()
msg := ""
updateLag := int64(now) - st.TimeUpdated
if updateLag > 0*1e9 {
msg += " Vstream may not be running."
}
txLag := int64(now) - st.TransactionTimestamp
msg += fmt.Sprintf(" VStream Lag: %ds.", txLag/1e9)
if st.TransactionTimestamp > 0 { // if no events occur after copy phase, TransactionTimeStamp can be 0
msg += fmt.Sprintf(" Tx time: %s.", time.Unix(st.TransactionTimestamp, 0).Format(time.ANSIC))
if st.State == "Error" {
msg += fmt.Sprintf(": %s.", st.Message)
} else if st.Pos == "" {
msg += ". VStream has not started."
} else {
now := time.Now().Nanosecond()
updateLag := int64(now) - st.TimeUpdated
if updateLag > 0*1e9 {
msg += ". VStream may not be running"
}
txLag := int64(now) - st.TransactionTimestamp
msg += fmt.Sprintf(". VStream Lag: %ds.", txLag/1e9)
if st.TransactionTimestamp > 0 { // if no events occur after copy phase, TransactionTimeStamp can be 0
msg += fmt.Sprintf(" Tx time: %s.", time.Unix(st.TransactionTimestamp, 0).Format(time.ANSIC))
}
}
s += fmt.Sprintf("id=%d on %s: Status: %s.%s\n", st.ID, ksShard, st.State, msg)
s += fmt.Sprintf("id=%d on %s: Status: %s%s\n", st.ID, ksShard, st.State, msg)
}
}
wr.Logger().Printf("\n%s\n", s)
Expand Down Expand Up @@ -2319,9 +2325,7 @@ func commandVRWorkflow(ctx context.Context, wr *wrangler.Wrangler, subFlags *pfl
if err != nil {
return err
}
if copyProgress == nil {
wr.Logger().Printf("\nCopy Completed.\n")
} else {
if copyProgress != nil {
wr.Logger().Printf("\nCopy Progress (approx):\n")
var tables []string
for table := range *copyProgress {
Expand All @@ -2346,7 +2350,6 @@ func commandVRWorkflow(ctx context.Context, wr *wrangler.Wrangler, subFlags *pfl
wr.Logger().Printf("\n%s\n", s)
}
return printDetails()

}

if *dryRun {
Expand Down Expand Up @@ -2390,7 +2393,6 @@ func commandVRWorkflow(ctx context.Context, wr *wrangler.Wrangler, subFlags *pfl
for {
select {
case <-ctx.Done():
errCh <- fmt.Errorf("workflow did not start within %s", (*timeout).String())
return
case <-ticker.C:
totalStreams, startedStreams, workflowErrors, err := wf.GetStreamCount()
Expand Down Expand Up @@ -2419,9 +2421,13 @@ func commandVRWorkflow(ctx context.Context, wr *wrangler.Wrangler, subFlags *pfl
return nil
}
wr.Logger().Printf("%d%% ... ", 100*progress.started/progress.total)
case <-timedCtx.Done():
wr.Logger().Printf("\nThe workflow did not start within %s. The workflow may simply be slow to start or there may be an issue.\n",
(*timeout).String())
wr.Logger().Printf("Check the status using the 'Workflow %s show' client command for details.\n", ksWorkflow)
Copy link
Copy Markdown
Contributor

@shlomi-noach shlomi-noach Jan 12, 2023

Choose a reason for hiding this comment

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

I wonder if it possible to check the status at this point and print the output? Save the user an extra step?

Copy link
Copy Markdown
Member Author

@mattlord mattlord Jan 12, 2023

Choose a reason for hiding this comment

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

IMO the output is already a little noisy and at this point (at least sub-second before) it had not yet started. I also like that we're informing/reminding them of the way to monitor the status and see the workflow details.

return fmt.Errorf("timed out waiting for workflow to start")
case err := <-errCh:
wr.Logger().Error(err)
cancelTimedCtx()
return err
case wfErrs := <-wfErrCh:
wr.Logger().Printf("Found problems with the streams created for this workflow:\n")
Expand Down
Loading