Flow API: Custom resume endpoint, polished mirror status endpoint, auth to all endpoints

[Amogh-Bharadwaj]

Endpoints:

Custom sync:

/v1/flows/cdc/sync

Status:

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

Authentication

Flow API now has authentication on all routes (except for health). It uses Bearer Authentication with token being:
base64Encode(bcryptHash(PEERDB_PASSWORD))

It can be generated on Unix based systems via:
htpasswd -nbB -C 10 "" <PEERDB_PASSWORD_HERE> | tr -d ":" | base64

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

[serprex]

testing on my cloud setup, drop mirror not working

[iamKunalGupta]

Why do we need to look at live password changes?
Why not send the hash directly?

[iamKunalGupta]

Can you check dropmirror flow once @heavycrystal ? @serprex said that it was not working

[iamKunalGupta]

remove token from logging now since it is plaintext now.

[Amogh-Bharadwaj]

Closing, this is replaced by #1823