feat: add maintenance mode for upgrades #2211

iamKunalGupta · 2024-11-01T03:47:06Z

Introduces Maintenance mode (status is available via dynamic config: PEERDB_MAINTENANCE_MODE_ENABLED)
Maintenance mode consists of 2 Workflows:
- StartMaintenance - for pre-upgrade, responsible for
  - Waiting for running snapshots
  - Updating dynamic config to true
  - Pausing and backing up currently running mirrors
- EndMaintenance - for post-upgrade, responsible for
  - Resuming backed up mirrors
  - Updating dynamic config to false
During the upgrade (between Start and End), mirrors cannot be mutated/created in any way,
There is also an instance info API which returns Ready/Maintenance which can be used for UI changes later.

There are 2 ways to trigger these 2 workflows:

API call to flow-api
Running the new maintenance entrypoint with the respective args

A new task queue is added so that the maintenance tasks can be spun up even during pre-upgrade hooks (from version earlier than ones containing this PR) and this also ensures that always the latest version of the maintenance flows run irrespective of the old version.

flow/activities/maintenance_activity.go

flow/workflows/maintenance_flow.go

flow/activities/maintenance_activity.go

flow/peerdbenv/dynamicconf.go

serprex · 2024-11-01T14:38:30Z

flow/peerdbenv/dynamicconf.go

+}
+
+func UpdatePeerDBMaintenanceModeEnabled(ctx context.Context, pool *pgxpool.Pool, enabled bool) error {
+	return UpdateDynamicSetting(ctx, pool, "PEERDB_MAINTENANCE_MODE_ENABLED", ptr.String(strconv.FormatBool(enabled)))


If this is controlled by system I'm not sure we want to expose it to user where they can toggle it, in which case it shouldn't be in dynconf

This can be helpful for manual intervention or maintenance. We can hide it later from dynamic config if needed

I've been thinking it'd be nice to be able to auth into ui with peerdb/clickhouse email & have access to extended functionality. Then this could be in catalog & have ui for us

Yes, maybe via oauth where different roles can map to different permissions like superadmin, admin or readonly

flow/workflows/maintenance_flow.go

nexus/catalog/migrations/V40__maintenance_flows.sql

flow/activities/maintenance_activity.go

heavycrystal · 2024-11-01T15:14:57Z

flow/activities/maintenance_activity.go

+	logEvery time.Duration,
+	alertEvery time.Duration,
+) (protos.FlowStatus, error) {
+	// In case a mirror was just kicked off, it shows up in the running state, we wait for a bit before checking for snapshot


this really shouldn't happen, seems like a bug

protos/route.proto

flow/activities/maintenance_activity.go

flow/cmd/maintenance.go

docker-bake.hcl

flow/cmd/handler.go

- also shift to tickers for waiting

serprex · 2024-11-06T16:54:03Z

flow/activities/maintenance_activity.go

+
+	flowStatus, err := RunEveryIntervalUntilFinish(ctx, func() (bool, protos.FlowStatus, error) {
+		activity.RecordHeartbeat(ctx, fmt.Sprintf("Waiting for mirror %s to finish snapshot", mirror.MirrorName))
+		mirrorStatus, err = a.getMirrorStatus(ctx, mirror)


Suggested change

mirrorStatus, err = a.getMirrorStatus(ctx, mirror)

mirrorStatus, err := a.getMirrorStatus(ctx, mirror)

serprex · 2024-11-06T16:54:41Z

flow/activities/maintenance_activity.go

+	mirrorStatus, err := a.getMirrorStatus(ctx, mirror)
+	if err != nil {
+		return mirrorStatus, err
+	}
+
+	slog.Info("Checking and waiting if mirror is snapshot", "mirror", mirror.MirrorName, "workflowId", mirror.WorkflowId, "status",
+		mirrorStatus.String())
+	if mirrorStatus != protos.FlowStatus_STATUS_SNAPSHOT && mirrorStatus != protos.FlowStatus_STATUS_SETUP {
+		return mirrorStatus, nil
+	}


Suggested change

mirrorStatus, err := a.getMirrorStatus(ctx, mirror)

if err != nil {

return mirrorStatus, err

}

slog.Info("Checking and waiting if mirror is snapshot", "mirror", mirror.MirrorName, "workflowId", mirror.WorkflowId, "status",

mirrorStatus.String())

if mirrorStatus != protos.FlowStatus_STATUS_SNAPSHOT && mirrorStatus != protos.FlowStatus_STATUS_SETUP {

return mirrorStatus, nil

}

serprex · 2024-11-06T16:55:12Z

flow/activities/maintenance_activity.go

+	err = model.FlowSignal.SignalClientWorkflow(ctx, a.TemporalClient, mirror.WorkflowId, "", model.PauseSignal)
+	if err != nil {


Suggested change

err = model.FlowSignal.SignalClientWorkflow(ctx, a.TemporalClient, mirror.WorkflowId, "", model.PauseSignal)

if err != nil {

if err := model.FlowSignal.SignalClientWorkflow(ctx, a.TemporalClient, mirror.WorkflowId, "", model.PauseSignal); err != nil {

elsewhere too

serprex · 2024-11-06T16:58:54Z

flow/cmd/maintenance.go

+	row := pool.QueryRow(ctx, `
+	select cli_version, api_version, skipped, skipped_reason
+	from maintenance.start_maintenance_outputs
+	order by created_at desc
+	limit 1
+	`)
+	var result StartMaintenanceResult
+	err = row.Scan(&result.CLIVersion, &result.APIVersion, &result.Skipped, &result.SkippedReason)
+	if err != nil {


Suggested change

row := pool.QueryRow(ctx, `

select cli_version, api_version, skipped, skipped_reason

from maintenance.start_maintenance_outputs

order by created_at desc

limit 1

`)

var result StartMaintenanceResult

err = row.Scan(&result.CLIVersion, &result.APIVersion, &result.Skipped, &result.SkippedReason)

if err != nil {

var result StartMaintenanceResult

if err := pool.QueryRow(ctx, `

select cli_version, api_version, skipped, skipped_reason

from maintenance.start_maintenance_outputs

order by created_at desc

limit 1

`).Scan(&result.CLIVersion, &result.APIVersion, &result.Skipped, &result.SkippedReason); err != nil {

serprex · 2024-11-06T17:01:10Z

flow/shared/worklow.go

+	var state protos.FlowStatus
+	err = res.Get(&state)
+	if err != nil {
+		slog.Error(fmt.Sprintf("failed to get status in workflow with ID %s: %s", workflowID, err.Error()))


Suggested change

slog.Error(fmt.Sprintf("failed to get status in workflow with ID %s: %s", workflowID, err.Error()))

slog.Error("failed to get status in workflow with ID "+workflowID, slog.Any("error", err)))

serprex · 2024-11-06T17:01:40Z

flow/shared/worklow.go

+	err = res.Get(&state)
+	if err != nil {


Suggested change

err = res.Get(&state)

if err != nil {

if err := res.Get(&state); err != nil {

serprex · 2024-11-06T17:02:39Z

flow/shared/worklow.go

+		slog.Error(fmt.Sprintf("failed to get status in workflow with ID %s: %s", workflowID, err.Error()))
+		return protos.FlowStatus_STATUS_UNKNOWN,
+			fmt.Errorf("failed to get status in workflow with ID %s: %w", workflowID, err)


Suggested change

slog.Error(fmt.Sprintf("failed to get status in workflow with ID %s: %s", workflowID, err.Error()))

return protos.FlowStatus_STATUS_UNKNOWN,

fmt.Errorf("failed to get status in workflow with ID %s: %w", workflowID, err)

slog.Error(fmt.Sprintf("failed to query status in workflow with ID %s: %s", workflowID, err.Error()))

return protos.FlowStatus_STATUS_UNKNOWN,

fmt.Errorf("failed to query status in workflow with ID %s: %w", workflowID, err)

makes error messages distinct so that when looking up error messages in code we know which line was hit

iamKunalGupta force-pushed the feat/maintenance-mode branch from fb17402 to ec62418 Compare November 1, 2024 03:50

iamKunalGupta marked this pull request as ready for review November 1, 2024 04:16

iamKunalGupta requested a review from a team November 1, 2024 04:24

iamKunalGupta changed the title ~~feat: add maintenance mode~~ feat: add maintenance mode for upgrades Nov 1, 2024