diff --git a/go.mod b/go.mod index fc86b5f..7c904a8 100644 --- a/go.mod +++ b/go.mod @@ -50,6 +50,7 @@ require ( github.com/fsnotify/fsnotify v1.6.0 // indirect github.com/gabriel-vasile/mimetype v1.4.3 // indirect github.com/gin-contrib/sse v0.1.0 // indirect + github.com/go-co-op/gocron/v2 v2.12.4 // indirect github.com/go-logr/logr v1.4.1 // indirect github.com/go-logr/stdr v1.2.2 // indirect github.com/go-playground/locales v0.14.1 // indirect @@ -66,6 +67,7 @@ require ( github.com/jackc/puddle/v2 v2.2.1 // indirect github.com/jinzhu/inflection v1.0.0 // indirect github.com/jinzhu/now v1.1.5 // indirect + github.com/jonboulle/clockwork v0.4.0 // indirect github.com/json-iterator/go v1.1.12 // indirect github.com/klauspost/compress v1.17.8 // indirect github.com/klauspost/cpuid/v2 v2.2.7 // indirect @@ -81,6 +83,7 @@ require ( github.com/prometheus/client_model v0.5.0 // indirect github.com/prometheus/common v0.48.0 // indirect github.com/prometheus/procfs v0.12.0 // indirect + github.com/robfig/cron/v3 v3.0.1 // indirect github.com/rogpeppe/go-internal v1.11.0 // indirect github.com/sagikazarmark/locafero v0.3.0 // indirect github.com/sagikazarmark/slog-shim v0.1.0 // indirect @@ -100,7 +103,7 @@ require ( go.uber.org/multierr v1.11.0 // indirect golang.org/x/arch v0.7.0 // indirect golang.org/x/crypto v0.22.0 // indirect - golang.org/x/exp v0.0.0-20240112132812-db7319d0e0e3 // indirect + golang.org/x/exp v0.0.0-20240613232115-7f521ea00fb8 // indirect golang.org/x/net v0.24.0 // indirect golang.org/x/sync v0.7.0 // indirect golang.org/x/sys v0.20.0 // indirect diff --git a/internal/client/elastic/elastic.go b/internal/client/elastic/elastic.go index 808550a..8ff62f1 100644 --- a/internal/client/elastic/elastic.go +++ b/internal/client/elastic/elastic.go @@ -45,6 +45,13 @@ func (el *ElasticSearchClient) IndexDocument(document interface{}) { } } +func (el *ElasticSearchClient) DeleteDocuments(searchRequest string) { + _, err := el.client.DeleteByQuery(el.Config.Index).Raw(strings.NewReader(searchRequest)).Do(context.TODO()) + if err != nil { + log.Printf("unable to delete documents: %s", err.Error()) + } + log.Printf("successfully deleted documents: %s", searchRequest) +} func (el *ElasticSearchClient) SearchDocuments(searchRequest string, fields []string) ([]map[string]interface{}, map[string]interface{}, int64, error) { res, err := el.client.Search(). Index(el.Config.Index).Raw(strings.NewReader(searchRequest)). diff --git a/internal/dependencies/dependencies.go b/internal/dependencies/dependencies.go index 20b5e8d..8c0cd81 100644 --- a/internal/dependencies/dependencies.go +++ b/internal/dependencies/dependencies.go @@ -11,6 +11,7 @@ import ( "cybertron/pkg/db" "cybertron/pkg/houstonClient" httpclient "cybertron/pkg/httpClient" + "cybertron/pkg/jobs" "cybertron/pkg/kafka/producer" "cybertron/pkg/log" "cybertron/pkg/mjolnirClient" @@ -77,7 +78,9 @@ func InitDependencies() *Dependencies { searchServiceClient := service.NewSearchService(logger, elasticSearch) authService := service.NewAuthService(mjolnirClient) houstonService := service.NewHoustonService(logger, dbClient, kafkaProducer, houstonClient) - + jobScheduler := jobs.NewJobScheduler(logger, elasticSearch) + jobScheduler.ScheduleEsCleanUpJob() + jobScheduler.Start() services := initServices(documentServiceClient, projectServiceClient, sourceMapServiceClient, releaseServiceClient, exceptionServiceClient, searchServiceClient, authService) handlers := initHandlers(projectServiceClient, sourceMapServiceClient, releaseServiceClient, exceptionServiceClient, searchServiceClient, houstonService) diff --git a/pkg/jobs/jobScheduler.go b/pkg/jobs/jobScheduler.go new file mode 100644 index 0000000..4e999cc --- /dev/null +++ b/pkg/jobs/jobScheduler.go @@ -0,0 +1,49 @@ +package jobs + +import ( + "cybertron/internal/client/elastic" + "cybertron/pkg/log" + "cybertron/pkg/utils" + "github.com/go-co-op/gocron/v2" + "go.uber.org/zap" + "time" +) + +type JobScheduler struct { + logger *log.Logger + scheduler gocron.Scheduler + elasticSearchClient *elastic.ElasticSearchClient +} + +func NewJobScheduler(logger *log.Logger, elasticSearchClient *elastic.ElasticSearchClient) *JobScheduler { + s, err := gocron.NewScheduler() + if err != nil { + logger.Error("Failed to start scheduler", zap.Error(err)) + } + + return &JobScheduler{ + logger: logger, + scheduler: s, + elasticSearchClient: elasticSearchClient, + } +} + +func (s *JobScheduler) ScheduleEsCleanUpJob() { + s.scheduler.NewJob(gocron.DurationJob(2*time.Hour), gocron.NewTask(esCleanUpJob, s)) +} +func esCleanUpJob(s *JobScheduler) { + println("running es clean up job ") + now := time.Now().Unix() + // Subtract 7 days (7 days * 24 hours * 60 minutes * 60 seconds) + sevenDaysAgo := now - (7 * 24 * 60 * 60) + rangeQuery := utils.CreateRangeQueryForLteString("created_at", sevenDaysAgo) + search_query := utils.CreateSearchQuery(rangeQuery) + es_query := utils.CreateEsQuery(search_query) + println("final query %v", es_query) + s.elasticSearchClient.DeleteDocuments(es_query) + +} +func (s *JobScheduler) Start() { + s.logger.Info("Starting job scheduler") + s.scheduler.Start() +} diff --git a/pkg/utils/elastic_query_util.go b/pkg/utils/elastic_query_util.go index 2513d56..ace1dea 100644 --- a/pkg/utils/elastic_query_util.go +++ b/pkg/utils/elastic_query_util.go @@ -11,6 +11,7 @@ const ( TermQueryForBool = `{ "term": { "%s": %t } }` RangeQuery = `{ "range": { "%s": { "gte": %d, "lte": %d } } }` RangeQueryGteString = `{ "range": { "%s": { "gte": "%s" } } }` + RangeQueryLteString = `{ "range": { "%s": { "lte": "%d" } } }` MustQuery = `"must": [ %s ] ` MustNotQuery = `"must_not": [ %s ]` ShouldQuery = `"should": [ %s ] ` @@ -107,6 +108,10 @@ func CreateRangeQueryForGteString(key string, greaterThan string) string { return fmt.Sprintf(RangeQueryGteString, key, greaterThan) } +func CreateRangeQueryForLteString(key string, lessThan int64) string { + return fmt.Sprintf(RangeQueryLteString, key, lessThan) +} + func CreateMustQuery(filters ...string) string { return fmt.Sprintf(MustQuery, strings.Join(filters, ",")) } diff --git a/service/searchService.go b/service/searchService.go index dfc492c..2d690d0 100644 --- a/service/searchService.go +++ b/service/searchService.go @@ -47,24 +47,6 @@ func (s *SearchService) GetErrorDetails(c *gin.Context) { after_query := utils.CreateFromQuery(fromInNumber) es_query := utils.CreateEsQuery(search_query, size_query, sort_query, after_query) - // searchRequest := ` - //{ - // "size": 1, - // "query": { - // "term": { - // "error_hash": { - // "value": "%s" - // } - // } - // }, - // "sort": [ - // { "created_at": { "order": "asc" } } - // ], - // "search_after": ["1724732743"] - //} - // - // ` - searchRequestformatted := es_query fields := []string{"error", "significant_stack", "title"} @@ -125,51 +107,6 @@ func (s *SearchService) GetErrorList(c *gin.Context) { final_query := utils.CreateEsQuery(search_query, size_query, compositeAggsQuery) println("%s", final_query) - // searchRequest := ` - //{ - // "size": 0, - // "query": { - // "term": { - // "project_id": { - // "value": "%s" - // } - // } - // }, - // "aggs": { - // "errors_by_hash": { - // "composite": { - // "size": 3, - // "sources": [ - // { - // "error_hash": { - // "terms": { - // "field": "error_hash.keyword" - // } - // } - // } - // ] - // }, - // "aggs": { - // "unique_errors": { - // "top_hits": { - // "_source": { - // "includes": ["error", "error_hash"] - // } - // } - // }, - // "error_count": { - // "value_count": { - // "field": "error.keyword" - // } - // } - // } - // } - // - // } - //} - // - // ` - fields := []string{"error", "significant_stack", "title"} var _, aggs, total, err = s.elasticSearchClient.SearchDocuments(final_query, fields)