Skip to content

Conversation

@vishalm0509
Copy link
Collaborator

Description

This PR implements the Temporal workflow and activity logic for executing the Clear Destination process.

Type of change

  • New feature (non-breaking change which adds functionality)
  • This change requires a documentation update

How Has This Been Tested?

  • Run clear destination manually
  • Update job and check if configured streams are getting dropped from the destination

chore: staging to master (helm chart version v0.0.1)
fix: add write permissions for gh token (#11)
feat: add release workflows (#17)
- Add nfsServer.external.storageClass and size parameters in values.yaml
- Modify shared-storage-pvc.yaml to always create PVC with conditional storage class
- Simplify _helpers.tpl sharedStoragePVC function to use consistent naming
- Add validation error when storageClass is empty and NFS is disabled
- Support RWX storage classes like EFS, Azure Files, GCP Filestore, etc.
- Remove PodAntiAffinity that prevented sync pods from running on same node
- Allow multiple sync operations to run concurrently for better resource utilization
- Keep JobID-based node affinity for resource mapping
- Fix tautological condition in affinity assignment
- Add ConfigMapWatcher using client-go informers for live updates
- Integrate watcher into K8sPodManager with graceful shutdown
- Update pod scheduling to use live mapping instead of static config
- Support debouncing (2s) for rapid ConfigMap changes
- Maintain backward compatibility with existing Helm charts
Comment on lines +123 to +124
workflowID := fmt.Sprintf("sync-%s-%d", req.ProjectID, req.JobID)
scheduleID := fmt.Sprintf("schedule-%s", workflowID)
Copy link

Choose a reason for hiding this comment

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

insted of constructing can we get this from ctx or as function param

Copy link
Collaborator Author

Choose a reason for hiding this comment

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

workflowID from context -> sync-123-37-2025-11-05T18:32:05Z
workflowID that we want -> sync-123-37

even in the param we are passing this sync-123-37-2025-11-05T18:32:05Z since we are creating workdir based on this. So we need to construct it manually only


// Register search attributes
_, err := t.GetClient().OperatorService().AddSearchAttributes(context.Background(), &operatorservice.AddSearchAttributesRequest{
SearchAttributes: map[string]enums.IndexedValueType{"OperationType": enums.INDEXED_VALUE_TYPE_KEYWORD},
Copy link

Choose a reason for hiding this comment

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

"OperationType" make it a constant

Copy link
Collaborator Author

Choose a reason for hiding this comment

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

Done

Base automatically changed from refactor/olake-worker to staging November 6, 2025 14:09
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

5 participants