From a4c028c993070edcc132aa8e621d321d9264bb7a Mon Sep 17 00:00:00 2001 From: Amogh Bharadwaj Date: Wed, 29 Nov 2023 01:02:12 +0530 Subject: [PATCH] Support for mirror name filter (#731) Fixes #696 We can now filter workflows by mirror name in Temporal UI Screenshot 2023-11-28 at 11 42 12 PM --- docker-compose-dev.yml | 3 +++ docker-compose.yml | 3 +++ flow/cmd/handler.go | 9 +++++++++ flow/shared/constants.go | 2 ++ flow/workflows/cdc_flow.go | 13 ++++++++++++- flow/workflows/qrep_flow.go | 3 +++ scripts/mirror-name-search.sh | 7 +++++++ 7 files changed, 39 insertions(+), 1 deletion(-) create mode 100644 scripts/mirror-name-search.sh diff --git a/docker-compose-dev.yml b/docker-compose-dev.yml index c34083c819..6dbb170740 100644 --- a/docker-compose-dev.yml +++ b/docker-compose-dev.yml @@ -86,11 +86,14 @@ services: image: temporalio/admin-tools:1.22 stdin_open: true tty: true + entrypoint: ["bash", "/etc/temporal/entrypoint.sh"] healthcheck: test: ["CMD", "tctl", "workflow", "list"] interval: 1s timeout: 5s retries: 30 + volumes: + - ./scripts/mirror-name-search.sh:/etc/temporal/entrypoint.sh temporal-ui: container_name: temporal-ui diff --git a/docker-compose.yml b/docker-compose.yml index 4b445e29c3..440f06545d 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -73,11 +73,14 @@ services: image: temporalio/admin-tools:1.22 stdin_open: true tty: true + entrypoint: ["bash", "/etc/temporal/entrypoint.sh"] healthcheck: test: ["CMD", "tctl", "workflow", "list"] interval: 1s timeout: 5s retries: 30 + volumes: + - ./scripts/mirror-name-search.sh:/etc/temporal/entrypoint.sh temporal-ui: container_name: temporal-ui diff --git a/flow/cmd/handler.go b/flow/cmd/handler.go index 8cd4951787..128bde255c 100644 --- a/flow/cmd/handler.go +++ b/flow/cmd/handler.go @@ -129,6 +129,9 @@ func (h *FlowRequestHandler) CreateCDCFlow( workflowOptions := client.StartWorkflowOptions{ ID: workflowID, TaskQueue: h.peerflowTaskQueueID, + SearchAttributes: map[string]interface{}{ + shared.MirrorNameSearchAttribute: cfg.FlowJobName, + }, } maxBatchSize := int(cfg.MaxBatchSize) @@ -229,6 +232,9 @@ func (h *FlowRequestHandler) CreateQRepFlow( workflowOptions := client.StartWorkflowOptions{ ID: workflowID, TaskQueue: h.peerflowTaskQueueID, + SearchAttributes: map[string]interface{}{ + shared.MirrorNameSearchAttribute: cfg.FlowJobName, + }, } if req.CreateCatalogEntry { err := h.createQrepJobEntry(ctx, req, workflowID) @@ -311,6 +317,9 @@ func (h *FlowRequestHandler) ShutdownFlow( workflowOptions := client.StartWorkflowOptions{ ID: workflowID, TaskQueue: h.peerflowTaskQueueID, + SearchAttributes: map[string]interface{}{ + shared.MirrorNameSearchAttribute: req.FlowJobName, + }, } dropFlowHandle, err := h.temporalClient.ExecuteWorkflow( ctx, // context diff --git a/flow/shared/constants.go b/flow/shared/constants.go index d1dfbdd6e1..48a6f8ae2f 100644 --- a/flow/shared/constants.go +++ b/flow/shared/constants.go @@ -11,6 +11,8 @@ const ( CDCFlowSignalName = "peer-flow-signal" ) +const MirrorNameSearchAttribute = "MirrorName" + type CDCFlowSignal int64 type ContextKey string diff --git a/flow/workflows/cdc_flow.go b/flow/workflows/cdc_flow.go index bbf134bf09..2dfe99ff25 100644 --- a/flow/workflows/cdc_flow.go +++ b/flow/workflows/cdc_flow.go @@ -189,6 +189,10 @@ func CDCFlowWorkflowWithConfig( } } + mirrorNameSearch := map[string]interface{}{ + shared.MirrorNameSearchAttribute: cfg.FlowJobName, + } + // start the SetupFlow workflow as a child workflow, and wait for it to complete // it should return the table schema for the source peer setupFlowID, err := GetChildWorkflowID(ctx, "setup-flow", cfg.FlowJobName) @@ -201,6 +205,7 @@ func CDCFlowWorkflowWithConfig( RetryPolicy: &temporal.RetryPolicy{ MaximumAttempts: 20, }, + SearchAttributes: mirrorNameSearch, } setupFlowCtx := workflow.WithChildOptions(ctx, childSetupFlowOpts) setupFlowFuture := workflow.ExecuteChildWorkflow(setupFlowCtx, SetupFlowWorkflow, cfg) @@ -226,7 +231,8 @@ func CDCFlowWorkflowWithConfig( RetryPolicy: &temporal.RetryPolicy{ MaximumAttempts: 20, }, - TaskQueue: taskQueue, + TaskQueue: taskQueue, + SearchAttributes: mirrorNameSearch, } snapshotFlowCtx := workflow.WithChildOptions(ctx, childSnapshotFlowOpts) snapshotFlowFuture := workflow.ExecuteChildWorkflow(snapshotFlowCtx, SnapshotFlowWorkflow, cfg) @@ -323,6 +329,9 @@ func CDCFlowWorkflowWithConfig( return state, err } + mirrorNameSearch := map[string]interface{}{ + shared.MirrorNameSearchAttribute: cfg.FlowJobName, + } // execute the sync flow as a child workflow childSyncFlowOpts := workflow.ChildWorkflowOptions{ WorkflowID: syncFlowID, @@ -330,6 +339,7 @@ func CDCFlowWorkflowWithConfig( RetryPolicy: &temporal.RetryPolicy{ MaximumAttempts: 20, }, + SearchAttributes: mirrorNameSearch, } ctx = workflow.WithChildOptions(ctx, childSyncFlowOpts) syncFlowOptions.RelationMessageMapping = *state.RelationMessageMapping @@ -362,6 +372,7 @@ func CDCFlowWorkflowWithConfig( RetryPolicy: &temporal.RetryPolicy{ MaximumAttempts: 20, }, + SearchAttributes: mirrorNameSearch, } ctx = workflow.WithChildOptions(ctx, childNormalizeFlowOpts) diff --git a/flow/workflows/qrep_flow.go b/flow/workflows/qrep_flow.go index 3b8e77a686..d56404af4a 100644 --- a/flow/workflows/qrep_flow.go +++ b/flow/workflows/qrep_flow.go @@ -206,6 +206,9 @@ func (q *QRepFlowExecution) startChildWorkflow( RetryPolicy: &temporal.RetryPolicy{ MaximumAttempts: 20, }, + SearchAttributes: map[string]interface{}{ + shared.MirrorNameSearchAttribute: q.config.FlowJobName, + }, }) future := workflow.ExecuteChildWorkflow( diff --git a/scripts/mirror-name-search.sh b/scripts/mirror-name-search.sh new file mode 100644 index 0000000000..c4f738a67a --- /dev/null +++ b/scripts/mirror-name-search.sh @@ -0,0 +1,7 @@ +# Check if MirrorName attribute exists +if ! temporal operator search-attribute list | grep -w MirrorName >/dev/null 2>&1; then + # If not, create MirrorName attribute + temporal operator search-attribute create --name MirrorName --type Text +fi + +tini -- sleep infinity