Skip to content
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

adding visibility into current mirror state #657

Closed
wants to merge 3 commits into from
+1,161 −729
Closed

adding visibility into current mirror state #657

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
bd72a5d
adding visibility into current mirror state
heavycrystal Nov 14, 2023
7ceb855
fixed bugs and updated temporal config
heavycrystal Nov 14, 2023
e720a67
minor test fixes
heavycrystal Nov 23, 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
63 changes: 57 additions & 6 deletions flow/cmd/handler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -364,30 +364,81 @@ func (h *FlowRequestHandler) ShutdownFlow(
}, nil
}

func (h *FlowRequestHandler) getWorkflowStatus(ctx context.Context, workflowID string) (*protos.FlowStatus, error) {
res, err := h.temporalClient.QueryWorkflow(ctx, workflowID, "", peerflow.FlowStatusQuery)
if err != nil {
return nil, fmt.Errorf("failed to get state in workflow with ID %s: %w", workflowID, err)
}
var state *protos.FlowStatus
err = res.Get(&state)
if err != nil {
return nil, fmt.Errorf("failed to get state in workflow with ID %s: %w", workflowID, err)
}
return state, nil
}

func (h *FlowRequestHandler) updateWorkflowStatus(ctx context.Context,
workflowID string, state *protos.FlowStatus) error {
_, err := h.temporalClient.UpdateWorkflow(ctx, workflowID, "", peerflow.FlowStatusUpdate, state)
if err != nil {
return fmt.Errorf("failed to update state in workflow with ID %s: %w", workflowID, err)
}
return nil
}

func (h *FlowRequestHandler) FlowStateChange(
ctx context.Context,
req *protos.FlowStateChangeRequest,
) (*protos.FlowStateChangeResponse, error) {
var err error
if req.RequestedFlowState == protos.FlowState_STATE_PAUSED {
workflowID, err := h.getWorkflowID(ctx, req.FlowJobName)
if err != nil {
return nil, err
}
currState, err := h.getWorkflowStatus(ctx, workflowID)
if err != nil {
return nil, err
}
if req.RequestedFlowState == protos.FlowStatus_STATUS_PAUSED &&
*currState == protos.FlowStatus_STATUS_RUNNING {
err = h.updateWorkflowStatus(ctx, workflowID, protos.FlowStatus_STATUS_PAUSING.Enum())
if err != nil {
return nil, err
}
err = h.temporalClient.SignalWorkflow(
ctx,
req.WorkflowId,
workflowID,
"",
shared.CDCFlowSignalName,
shared.PauseSignal,
)
} else if req.RequestedFlowState == protos.FlowState_STATE_RUNNING {
} else if req.RequestedFlowState == protos.FlowStatus_STATUS_RUNNING &&
*currState == protos.FlowStatus_STATUS_PAUSED {
err = h.temporalClient.SignalWorkflow(
ctx,
req.WorkflowId,
workflowID,
"",
shared.CDCFlowSignalName,
shared.NoopSignal,
)
} else if req.RequestedFlowState == protos.FlowStatus_STATUS_TERMINATED &&
(*currState == protos.FlowStatus_STATUS_RUNNING || *currState == protos.FlowStatus_STATUS_PAUSED) {
err = h.updateWorkflowStatus(ctx, workflowID, protos.FlowStatus_STATUS_TERMINATING.Enum())
if err != nil {
return nil, err
}
_, err = h.ShutdownFlow(ctx, &protos.ShutdownRequest{
Copy link
Contributor

@serprex serprex Nov 14, 2023 •
edited
Loading

Choose a reason for hiding this comment

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

Calling ShutdownFlow doesn't seem related to visibility. Is the purpose of this call to reduce latency to start terminating?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

this is to preserve existing behavior that waits till the workflow terminates before returning, while also ensuring we go from STATUS_TERMINATING -> STATUS_TERMINATED

WorkflowId: workflowID,
FlowJobName: req.FlowJobName,
SourcePeer: req.SourcePeer,
DestinationPeer: req.DestinationPeer,
RemoveFlowEntry: false,
})
} else {
return nil, fmt.Errorf("illegal state change requested: %v, current state is: %v",
req.RequestedFlowState, currState)
}
if err != nil {
return nil, fmt.Errorf("unable to signal PeerFlow workflow: %w", err)
return nil, fmt.Errorf("unable to signal CDCFlow workflow: %w", err)
}

return &protos.FlowStateChangeResponse{
Expand Down
23 changes: 23 additions & 0 deletions flow/cmd/mirror_status.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,16 @@ func (h *FlowRequestHandler) MirrorStatus(
ErrorMessage: fmt.Sprintf("unable to query flow: %s", err.Error()),
}, nil
}
workflowID, err := h.getWorkflowID(ctx, req.FlowJobName)
if err != nil {
return nil, err
}
currState, err := h.getWorkflowStatus(ctx, workflowID)
if err != nil {
return &protos.MirrorStatusResponse{
ErrorMessage: fmt.Sprintf("unable to get flow state: %s", err.Error()),
}, nil
}

if cdcFlow {
cdcStatus, err := h.CDCFlowStatus(ctx, req)
Expand All @@ -36,6 +46,7 @@ func (h *FlowRequestHandler) MirrorStatus(
Status: &protos.MirrorStatusResponse_CdcStatus{
CdcStatus: cdcStatus,
},
CurrentFlowState: *currState,
}, nil
} else {
qrepStatus, err := h.QRepFlowStatus(ctx, req)
Expand All @@ -50,6 +61,7 @@ func (h *FlowRequestHandler) MirrorStatus(
Status: &protos.MirrorStatusResponse_QrepStatus{
QrepStatus: qrepStatus,
},
CurrentFlowState: *currState,
}, nil
}
}
Expand Down Expand Up @@ -273,3 +285,14 @@ func (h *FlowRequestHandler) getCloneTableFlowNames(ctx context.Context, flowJob

return flowNames, nil
}

func (h *FlowRequestHandler) getWorkflowID(ctx context.Context, flowJobName string) (string, error) {
q := "SELECT workflow_id FROM flows WHERE name ILIKE $1"
row := h.pool.QueryRow(ctx, q, flowJobName)
var workflowID string
if err := row.Scan(&workflowID); err != nil {
return "", fmt.Errorf("unable to get workflowID for flow job %s: %w", flowJobName, err)
}

return workflowID, nil
}
24 changes: 12 additions & 12 deletions flow/e2e/bigquery/peer_flow_bq_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -260,7 +260,7 @@ func (s *PeerFlowE2ETestSuiteBQ) Test_Complete_Simple_Flow_BQ() {
// in a separate goroutine, wait for PeerFlowStatusQuery to finish setup
// and then insert 10 rows into the source table
go func() {
e2e.SetupCDCFlowStatusQuery(env, connectionGen)
e2e.SetupCDCFlowStateQuery(env, connectionGen)
// insert 10 rows into the source table
for i := 0; i < 10; i++ {
testKey := fmt.Sprintf("test_key_%d", i)
Expand Down Expand Up @@ -330,7 +330,7 @@ func (s *PeerFlowE2ETestSuiteBQ) Test_Toast_BQ() {
// in a separate goroutine, wait for PeerFlowStatusQuery to finish setup
// and execute a transaction touching toast columns
go func() {
e2e.SetupCDCFlowStatusQuery(env, connectionGen)
e2e.SetupCDCFlowStateQuery(env, connectionGen)
/*
Executing a transaction which
1. changes both toast column
Expand Down Expand Up @@ -400,7 +400,7 @@ func (s *PeerFlowE2ETestSuiteBQ) Test_Toast_Nochanges_BQ() {
// in a separate goroutine, wait for PeerFlowStatusQuery to finish setup
// and execute a transaction touching toast columns
go func() {
e2e.SetupCDCFlowStatusQuery(env, connectionGen)
e2e.SetupCDCFlowStateQuery(env, connectionGen)
/* transaction updating no rows */
_, err = s.pool.Exec(context.Background(), fmt.Sprintf(`
BEGIN;
Expand Down Expand Up @@ -463,7 +463,7 @@ func (s *PeerFlowE2ETestSuiteBQ) Test_Toast_Advance_1_BQ() {
// in a separate goroutine, wait for PeerFlowStatusQuery to finish setup
// and execute a transaction touching toast columns
go func() {
e2e.SetupCDCFlowStatusQuery(env, connectionGen)
e2e.SetupCDCFlowStateQuery(env, connectionGen)
//complex transaction with random DMLs on a table with toast columns
_, err = s.pool.Exec(context.Background(), fmt.Sprintf(`
BEGIN;
Expand Down Expand Up @@ -538,7 +538,7 @@ func (s *PeerFlowE2ETestSuiteBQ) Test_Toast_Advance_2_BQ() {
// in a separate goroutine, wait for PeerFlowStatusQuery to finish setup
// and execute a transaction touching toast columns
go func() {
e2e.SetupCDCFlowStatusQuery(env, connectionGen)
e2e.SetupCDCFlowStateQuery(env, connectionGen)
//complex transaction with random DMLs on a table with toast columns
_, err = s.pool.Exec(context.Background(), fmt.Sprintf(`
BEGIN;
Expand Down Expand Up @@ -608,7 +608,7 @@ func (s *PeerFlowE2ETestSuiteBQ) Test_Toast_Advance_3_BQ() {
// in a separate goroutine, wait for PeerFlowStatusQuery to finish setup
// and execute a transaction touching toast columns
go func() {
e2e.SetupCDCFlowStatusQuery(env, connectionGen)
e2e.SetupCDCFlowStateQuery(env, connectionGen)
/*
transaction updating a single row
multiple times with changed/unchanged toast columns
Expand Down Expand Up @@ -678,7 +678,7 @@ func (s *PeerFlowE2ETestSuiteBQ) Test_Types_BQ() {
// in a separate goroutine, wait for PeerFlowStatusQuery to finish setup
// and execute a transaction touching toast columns
go func() {
e2e.SetupCDCFlowStatusQuery(env, connectionGen)
e2e.SetupCDCFlowStateQuery(env, connectionGen)
/* test inserting various types*/
_, err = s.pool.Exec(context.Background(), fmt.Sprintf(`
INSERT INTO %s SELECT 2,2,b'1',b'101',
Expand Down Expand Up @@ -755,7 +755,7 @@ func (s *PeerFlowE2ETestSuiteBQ) Test_Multi_Table_BQ() {
// in a separate goroutine, wait for PeerFlowStatusQuery to finish setup
// and execute a transaction touching toast columns
go func() {
e2e.SetupCDCFlowStatusQuery(env, connectionGen)
e2e.SetupCDCFlowStateQuery(env, connectionGen)
/* inserting across multiple tables*/
_, err = s.pool.Exec(context.Background(), fmt.Sprintf(`
INSERT INTO %s (c1,c2) VALUES (1,'dummy_1');
Expand Down Expand Up @@ -819,7 +819,7 @@ func (s *PeerFlowE2ETestSuiteBQ) Test_Simple_Schema_Changes_BQ() {
// and then insert and mutate schema repeatedly.
go func() {
// insert first row.
e2e.SetupCDCFlowStatusQuery(env, connectionGen)
e2e.SetupCDCFlowStateQuery(env, connectionGen)
_, err = s.pool.Exec(context.Background(), fmt.Sprintf(`
INSERT INTO %s(c1) VALUES ($1)`, srcTableName), 1)
s.NoError(err)
Expand Down Expand Up @@ -923,7 +923,7 @@ func (s *PeerFlowE2ETestSuiteBQ) Test_Composite_PKey_BQ() {
// in a separate goroutine, wait for PeerFlowStatusQuery to finish setup
// and then insert, update and delete rows in the table.
go func() {
e2e.SetupCDCFlowStatusQuery(env, connectionGen)
e2e.SetupCDCFlowStateQuery(env, connectionGen)
// insert 10 rows into the source table
for i := 0; i < 10; i++ {
testValue := fmt.Sprintf("test_value_%d", i)
Expand Down Expand Up @@ -999,7 +999,7 @@ func (s *PeerFlowE2ETestSuiteBQ) Test_Composite_PKey_Toast_1_BQ() {
// in a separate goroutine, wait for PeerFlowStatusQuery to finish setup
// and then insert, update and delete rows in the table.
go func() {
e2e.SetupCDCFlowStatusQuery(env, connectionGen)
e2e.SetupCDCFlowStateQuery(env, connectionGen)
rowsTx, err := s.pool.Begin(context.Background())
s.NoError(err)

Expand Down Expand Up @@ -1078,7 +1078,7 @@ func (s *PeerFlowE2ETestSuiteBQ) Test_Composite_PKey_Toast_2_BQ() {
// in a separate goroutine, wait for PeerFlowStatusQuery to finish setup
// and then insert, update and delete rows in the table.
go func() {
e2e.SetupCDCFlowStatusQuery(env, connectionGen)
e2e.SetupCDCFlowStateQuery(env, connectionGen)

// insert 10 rows into the source table
for i := 0; i < 10; i++ {
Expand Down
10 changes: 5 additions & 5 deletions flow/e2e/postgres/peer_flow_pg_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,7 +52,7 @@ func (s *PeerFlowE2ETestSuitePG) Test_Simple_Flow_PG() {
// in a separate goroutine, wait for PeerFlowStatusQuery to finish setup
// and then insert 10 rows into the source table
go func() {
e2e.SetupCDCFlowStatusQuery(env, connectionGen)
e2e.SetupCDCFlowStateQuery(env, connectionGen)
// insert 10 rows into the source table
for i := 0; i < 10; i++ {
testKey := fmt.Sprintf("test_key_%d", i)
Expand Down Expand Up @@ -115,7 +115,7 @@ func (s *PeerFlowE2ETestSuitePG) Test_Simple_Schema_Changes_PG() {
// and then insert and mutate schema repeatedly.
go func() {
// insert first row.
e2e.SetupCDCFlowStatusQuery(env, connectionGen)
e2e.SetupCDCFlowStateQuery(env, connectionGen)
_, err = s.pool.Exec(context.Background(), fmt.Sprintf(`
INSERT INTO %s(c1) VALUES ($1)`, srcTableName), 1)
s.NoError(err)
Expand Down Expand Up @@ -278,7 +278,7 @@ func (s *PeerFlowE2ETestSuitePG) Test_Composite_PKey_PG() {
// in a separate goroutine, wait for PeerFlowStatusQuery to finish setup
// and then insert, update and delete rows in the table.
go func() {
e2e.SetupCDCFlowStatusQuery(env, connectionGen)
e2e.SetupCDCFlowStateQuery(env, connectionGen)
// insert 10 rows into the source table
for i := 0; i < 10; i++ {
testValue := fmt.Sprintf("test_value_%d", i)
Expand Down Expand Up @@ -357,7 +357,7 @@ func (s *PeerFlowE2ETestSuitePG) Test_Composite_PKey_Toast_1_PG() {
// in a separate goroutine, wait for PeerFlowStatusQuery to finish setup
// and then insert, update and delete rows in the table.
go func() {
e2e.SetupCDCFlowStatusQuery(env, connectionGen)
e2e.SetupCDCFlowStateQuery(env, connectionGen)
rowsTx, err := s.pool.Begin(context.Background())
s.NoError(err)

Expand Down Expand Up @@ -438,7 +438,7 @@ func (s *PeerFlowE2ETestSuitePG) Test_Composite_PKey_Toast_2_PG() {
// in a separate goroutine, wait for PeerFlowStatusQuery to finish setup
// and then insert, update and delete rows in the table.
go func() {
e2e.SetupCDCFlowStatusQuery(env, connectionGen)
e2e.SetupCDCFlowStateQuery(env, connectionGen)

// insert 10 rows into the source table
for i := 0; i < 10; i++ {
Expand Down
4 changes: 2 additions & 2 deletions flow/e2e/s3/cdc_s3_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,7 @@ func (s *PeerFlowE2ETestSuiteS3) Test_Complete_Simple_Flow_S3() {
}

go func() {
e2e.SetupCDCFlowStatusQuery(env, connectionGen)
e2e.SetupCDCFlowStateQuery(env, connectionGen)
s.NoError(err)
//insert 20 rows
for i := 1; i <= 20; i++ {
Expand Down Expand Up @@ -120,7 +120,7 @@ func (s *PeerFlowE2ETestSuiteS3) Test_Complete_Simple_Flow_GCS_Interop() {
}

go func() {
e2e.SetupCDCFlowStatusQuery(env, connectionGen)
e2e.SetupCDCFlowStateQuery(env, connectionGen)
s.NoError(err)
//insert 20 rows
for i := 1; i <= 20; i++ {
Expand Down
Loading
Loading