diff --git a/configs/application.yml b/configs/application.yml index b43b5be..4ba726e 100644 --- a/configs/application.yml +++ b/configs/application.yml @@ -86,3 +86,6 @@ mjolnir: houston: service.url: HOUSTON_SERVICE_URL realm.id: HOUSTON_REALM_ID + +jobScheduler: + retentionInDays: 7 diff --git a/configs/config.go b/configs/config.go index b4b88af..7a7fb2d 100644 --- a/configs/config.go +++ b/configs/config.go @@ -9,20 +9,21 @@ import ( ) type AppConfig struct { - name string - env string - port int - metricsPort int - prometheus *Prometheus - postgres Postgres - timezone string - httpConfig *HttpConfig - clientConfigs *ClientConfigs - awsConfig *AwsConfig - KafkaConfig *KafkaConfig - ElasticConfig *ElasticConfig - mjolnir *MjolnirClientConfig - houston *HoustonClientConfig + name string + env string + port int + metricsPort int + prometheus *Prometheus + postgres Postgres + timezone string + httpConfig *HttpConfig + clientConfigs *ClientConfigs + awsConfig *AwsConfig + KafkaConfig *KafkaConfig + ElasticConfig *ElasticConfig + mjolnir *MjolnirClientConfig + houston *HoustonClientConfig + JobSchedulerConfig *JobSchedulerConfig } type MigConfig struct { @@ -37,20 +38,21 @@ func LoadConfig() { readConfig() appConfig = AppConfig{ - name: getString("name", true), - env: getString("env", true), - port: getInt("port", true), - metricsPort: getInt("metrics.port", true), - prometheus: GetPrometheusConfig(), - postgres: getPostgresConfig(), - timezone: getString("timezone", true), - clientConfigs: loadClientConfigs(), - httpConfig: NewHttpConfig(), - awsConfig: NewAWSConfig(), - KafkaConfig: NewKafkaConfig(), - ElasticConfig: NewElasticConfig(), - mjolnir: NewMjolnirConfig(), - houston: NewHoustonConfig(), + name: getString("name", true), + env: getString("env", true), + port: getInt("port", true), + metricsPort: getInt("metrics.port", true), + prometheus: GetPrometheusConfig(), + postgres: getPostgresConfig(), + timezone: getString("timezone", true), + clientConfigs: loadClientConfigs(), + httpConfig: NewHttpConfig(), + awsConfig: NewAWSConfig(), + KafkaConfig: NewKafkaConfig(), + ElasticConfig: NewElasticConfig(), + mjolnir: NewMjolnirConfig(), + houston: NewHoustonConfig(), + JobSchedulerConfig: NewJobSchedulerConfig(), } } @@ -124,3 +126,7 @@ func GetMjolnirConfig() *MjolnirClientConfig { func GetHoustonConfig() *HoustonClientConfig { return appConfig.houston } + +func GetJobSchedulerConfig() *JobSchedulerConfig { + return appConfig.JobSchedulerConfig +} diff --git a/configs/jobScheduler.go b/configs/jobScheduler.go new file mode 100644 index 0000000..bc9fda7 --- /dev/null +++ b/configs/jobScheduler.go @@ -0,0 +1,15 @@ +package configs + +type JobSchedulerConfig struct { + MaxRetentionDays int +} + +func NewJobSchedulerConfig() *JobSchedulerConfig { + return &JobSchedulerConfig{ + MaxRetentionDays: getInt("jobScheduler.retentionInDays", true), + } +} + +func (p *JobSchedulerConfig) GetMaxRetentionDays() int { + return p.MaxRetentionDays +} diff --git a/go.mod b/go.mod index 7c904a8..35281b8 100644 --- a/go.mod +++ b/go.mod @@ -8,10 +8,11 @@ require ( github.com/confluentinc/confluent-kafka-go/v2 v2.5.0 github.com/elastic/go-elasticsearch/v8 v8.14.0 github.com/gin-contrib/cors v1.7.2 - github.com/gin-contrib/zap v0.2.0 github.com/gin-gonic/gin v1.9.1 + github.com/go-co-op/gocron/v2 v2.12.4 github.com/golang-migrate/migrate/v4 v4.17.1 github.com/google/uuid v1.6.0 + github.com/lib/pq v1.10.9 github.com/prometheus/client_golang v1.19.1 github.com/spf13/cobra v1.8.0 github.com/spf13/viper v1.17.0 @@ -50,7 +51,6 @@ 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 @@ -72,7 +72,6 @@ require ( github.com/klauspost/compress v1.17.8 // indirect github.com/klauspost/cpuid/v2 v2.2.7 // indirect github.com/leodido/go-urn v1.4.0 // indirect - github.com/lib/pq v1.10.9 // indirect github.com/magiconair/properties v1.8.7 // indirect github.com/mattn/go-isatty v0.0.20 // indirect github.com/mitchellh/mapstructure v1.5.0 // indirect diff --git a/internal/client/elastic/elastic.go b/internal/client/elastic/elastic.go index 8ff62f1..98db07e 100644 --- a/internal/client/elastic/elastic.go +++ b/internal/client/elastic/elastic.go @@ -49,6 +49,7 @@ 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()) + return } log.Printf("successfully deleted documents: %s", searchRequest) } diff --git a/internal/dependencies/dependencies.go b/internal/dependencies/dependencies.go index 8c0cd81..8a176e6 100644 --- a/internal/dependencies/dependencies.go +++ b/internal/dependencies/dependencies.go @@ -78,11 +78,11 @@ 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) + jobScheduler := jobs.NewJobScheduler(logger, elasticSearch, configs.GetJobSchedulerConfig()) + jobScheduler.ScheduleEsCleanUpJob() + jobScheduler.Start() return &Dependencies{ Service: services, diff --git a/pkg/jobs/jobScheduler.go b/pkg/jobs/jobScheduler.go index 4e999cc..f930501 100644 --- a/pkg/jobs/jobScheduler.go +++ b/pkg/jobs/jobScheduler.go @@ -1,6 +1,7 @@ package jobs import ( + "cybertron/configs" "cybertron/internal/client/elastic" "cybertron/pkg/log" "cybertron/pkg/utils" @@ -13,9 +14,10 @@ type JobScheduler struct { logger *log.Logger scheduler gocron.Scheduler elasticSearchClient *elastic.ElasticSearchClient + config configs.JobSchedulerConfig } -func NewJobScheduler(logger *log.Logger, elasticSearchClient *elastic.ElasticSearchClient) *JobScheduler { +func NewJobScheduler(logger *log.Logger, elasticSearchClient *elastic.ElasticSearchClient, jonSchedulerConfig *configs.JobSchedulerConfig) *JobScheduler { s, err := gocron.NewScheduler() if err != nil { logger.Error("Failed to start scheduler", zap.Error(err)) @@ -25,17 +27,23 @@ func NewJobScheduler(logger *log.Logger, elasticSearchClient *elastic.ElasticSea logger: logger, scheduler: s, elasticSearchClient: elasticSearchClient, + config: *jonSchedulerConfig, } } func (s *JobScheduler) ScheduleEsCleanUpJob() { - s.scheduler.NewJob(gocron.DurationJob(2*time.Hour), gocron.NewTask(esCleanUpJob, s)) + s.scheduler.NewJob(gocron.DurationJob(2*time.Second), 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) + retentionDays := s.config.GetMaxRetentionDays() + + if retentionDays <= 0 { + retentionDays = 7 + } + sevenDaysAgo := now - (int64(retentionDays) * 24 * 60 * 60) rangeQuery := utils.CreateRangeQueryForLteString("created_at", sevenDaysAgo) search_query := utils.CreateSearchQuery(rangeQuery) es_query := utils.CreateEsQuery(search_query)