Flow API: Custom resume endpoint and blackboxed cdc status endpoint

[Amogh-Bharadwaj]

Endpoints:

Custom sync:

/v1/flows/cdc/sync

Status:

/v1/mirrors/{flow_job_name}?include_flow_info=true|false

Authentication

Authentication for both endpoints is the SHA256 hash of the peerdb instance password

API Endpoint to Resume Mirror With Specified Syncs

This PR introduces an endpoint in our Flow API to resume mirrors with a request of custom number of syncs.

• The number of syncs the client can request is, for now, a number between 1 and the MaxSyncsPerCDCFlow which is currently 32.
• Syncs in this case refers to one sync flow - normalise flow pair.
• This endpoint requires the mirror to be paused before hand.
• Request validation checks the above along with whether the mirror name exists, etc.

Implementation:

Extends CDCFlowConfigUpdate to include number of syncs information.

message CDCFlowConfigUpdate {
  repeated TableMapping additional_tables = 1;
  uint32 batch_size = 2;
  uint64 idle_timeout = 3;
  int32 number_of_syncs = 4;
}

• This is picked up in CDC flow after processing FlowConfigUpdate:

if state.FlowConfigUpdate != nil {
	err = processCDCFlowConfigUpdate(ctx, logger, cfg, state, mirrorNameSearch)
        if err != nil {
		return state, err
	}
	syncCountLimit = int(state.SyncFlowOptions.NumberOfSyncs)

• Limit check is done when listening for normalise flow finishing:

normDoneChan.AddToSelector(mainLoopSelector, func(x struct{}, _ bool) {
	if syncCount == syncCountLimit {
		logger.Info("sync count limit reached, pausing",
		slog.Int("limit", syncCountLimit),
		slog.Int("count", syncCount))
		state.ActiveSignal = model.PauseSignal
	}

Functional testing:

• Tested with a paused mirror with low sync interval via HTTP requests to this endpoint
• Resume mirror via the normal UI button route will stick to existing behaviour of MaxSyncsPerCDCFlow=32 sync flows.

Mirror Status Endpoint

Existing mirror status endpoint now has a flag in the request to include flow information. The UI sets this flag. If not set, response only contains flow name and running status, along with clean error messages.

This endpoint is also gated by authorization

Functionally tested

[CLAassistant]

All committers have signed the CLA.

[Amogh-Bharadwaj]

Description updated with new status endpoint

[Amogh-Bharadwaj]

Updated description and functionally tested post changes