-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add elastic documents purge job (#209)
* Add ElasticDocPurge job to scheduler with deletion logic This commit introduces a new `ElasticDocPurge` job type, enabling the purging of Elastic documents based on specific conditions. The corresponding logic for job validation, execution, and fact deletion has been implemented, including support for contextualized query construction and handling shard failures. Dependencies were updated to include the necessary SDK functionalities. * Add ElasticDocPurge job to scheduler with deletion logic This commit introduces a new `ElasticDocPurge` job type, enabling the purging of Elastic documents based on specific conditions. The corresponding logic for job validation, execution, and fact deletion has been implemented, including support for contextualized query construction and handling shard failures. Dependencies were updated to include the necessary SDK functionalities. * Update myrtea-sdk to stable version 5.1.7 * Update internals/scheduler/elastic_doc_purge_job.go Co-authored-by: Paul <[email protected]> * Update internals/scheduler/elastic_doc_purge_job.go Co-authored-by: Paul <[email protected]> * Update internals/scheduler/elastic_doc_purge_job.go Co-authored-by: Paul <[email protected]> * Update internals/scheduler/elastic_doc_purge_job.go Co-authored-by: Paul <[email protected]> * Add continue statement to skip non-delete facts --------- Co-authored-by: Paul <[email protected]>
- Loading branch information
1 parent
a06e638
commit f4cf050
Showing
5 changed files
with
154 additions
and
8 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,81 @@ | ||
| package scheduler | ||
|
|
||
| import ( | ||
| "errors" | ||
| "github.com/myrteametrics/myrtea-engine-api/v5/internals/fact" | ||
| "github.com/myrteametrics/myrtea-sdk/v5/engine" | ||
| "go.uber.org/zap" | ||
| "time" | ||
| ) | ||
|
|
||
| type ElasticDocPurgeJob struct { | ||
| FactIds []int64 `json:"factIds"` | ||
| ScheduleID int64 `json:"-"` | ||
| } | ||
|
|
||
| func (job ElasticDocPurgeJob) IsValid() (bool, error) { | ||
| if job.FactIds == nil || len(job.FactIds) <= 0 { | ||
| return false, errors.New("missing FactIds") | ||
| } | ||
| return true, nil | ||
| } | ||
|
|
||
| func (job ElasticDocPurgeJob) Run() { | ||
|
|
||
| if S().ExistingRunningJob(job.ScheduleID) { | ||
| zap.L().Info("Skipping Elastic document purge job because last execution is still running", zap.Int64s("ids", job.FactIds)) | ||
| return | ||
| } | ||
| S().AddRunningJob(job.ScheduleID) | ||
|
|
||
| zap.L().Info("Delete Elastic document job started", zap.Int64s("ids", job.FactIds)) | ||
|
|
||
| t := time.Now().Truncate(1 * time.Second).UTC() | ||
|
|
||
| PurgeElasticDocs(t, job.FactIds) | ||
|
|
||
| zap.L().Info("Elastic document purge job ended", zap.Int64("id Schedule", job.ScheduleID)) | ||
| S().RemoveRunningJob(job.ScheduleID) | ||
|
|
||
| } | ||
|
|
||
| func PurgeElasticDocs(t time.Time, factIds []int64) { | ||
| zap.L().Info("Starting Elastic document purge", zap.Time("timestamp", t), zap.Int("number_of_facts", len(factIds))) | ||
|
|
||
| for _, factId := range factIds { | ||
| zap.L().Debug("Processing fact", zap.Int64("factId", factId)) | ||
|
|
||
| // Retrieve the Fact | ||
| f, found, err := fact.R().Get(factId) | ||
| if err != nil { | ||
| zap.L().Error("Error retrieving the fact; skipping deletion", | ||
| zap.Int64("factId", factId), | ||
| zap.Error(err)) | ||
| continue | ||
| } | ||
|
|
||
| if !found { | ||
| zap.L().Warn("Fact does not exist; skipping deletion", zap.Int64("factId", factId)) | ||
| continue | ||
| } | ||
|
|
||
| if f.Intent.Operator != engine.Delete { | ||
| zap.L().Warn("Fact is not a delete fact; skipping deletion", zap.Int64("factId", factId)) | ||
| continue | ||
| } | ||
|
|
||
| // Execute deletion | ||
| _, err = fact.ExecuteFactDeleteQuery(t, f) | ||
| if err != nil { | ||
| zap.L().Error("Error during fact deletion", | ||
| zap.Int64("factId", f.ID), | ||
| zap.Any("fact", f), | ||
| zap.Error(err)) | ||
| continue | ||
| } | ||
|
|
||
| zap.L().Info("Fact successfully deleted from Elastic", zap.Int64("factId", f.ID)) | ||
| } | ||
|
|
||
| zap.L().Info("Elastic document purge completed", zap.Time("timestamp", t), zap.Int("processed_facts", len(factIds))) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters