From 17afa5a383b775da87da7429d68a5d1fac208da0 Mon Sep 17 00:00:00 2001 From: hltav Date: Wed, 19 Aug 2026 18:58:49 -0300 Subject: [PATCH] feat(scraper): configure global concurrency and runtime limits Add centralized SCRAPER_MAX_CONCURRENCY configuration with a safe default of 12. Reject invalid environment values during startup instead of using a silent fallback. Cap manual and scheduled executions before cache-key and pipeline processing. Configure Go runtime, CPU, and memory limits for the scraper container. Add structured logs, defensive validation, tests, and operational documentation. Refs: PAV-118 --- .env.example | 3 + SCRAPER.md | 35 ++++++++++ docker-compose.yml | 5 ++ scraper-go/cmd/server/handlers.go | 40 +++++++----- scraper-go/cmd/server/handlers_test.go | 56 ++++++++++++++++ scraper-go/cmd/server/main.go | 49 +++++++++++++- scraper-go/cmd/server/runtime_log.go | 17 +++++ scraper-go/cmd/server/server.go | 6 +- scraper-go/internal/config/config.go | 64 +++++++++++++++++++ scraper-go/internal/config/config_test.go | 58 +++++++++++++++++ scraper-go/internal/cronjob/cronjob.go | 40 +++++++----- scraper-go/internal/cronjob/cronjob_test.go | 33 ++++++++++ .../internal/pipeline/cache_key_test.go | 17 ++++- scraper-go/internal/pipeline/pipeline.go | 8 +-- scraper-go/internal/pipeline/scrape.go | 5 +- .../internal/pipeline/source_schedule_test.go | 28 +++++++- 16 files changed, 421 insertions(+), 43 deletions(-) create mode 100644 scraper-go/cmd/server/handlers_test.go create mode 100644 scraper-go/cmd/server/runtime_log.go create mode 100644 scraper-go/internal/config/config.go create mode 100644 scraper-go/internal/config/config_test.go create mode 100644 scraper-go/internal/cronjob/cronjob_test.go diff --git a/.env.example b/.env.example index c2cc788..1223345 100644 --- a/.env.example +++ b/.env.example @@ -30,6 +30,9 @@ TIME_FILTER=r604800 SEARCH_KEYWORDS=UX Designer,UI Designer,Product Manager,Product Owner # Scraping behavior +SCRAPER_MAX_CONCURRENCY=12 +GOMAXPROCS=2 +GOMEMLIMIT=1500MiB WAIT_BETWEEN_SEARCHES_MS=5000 PAGE_TIMEOUT_MS=10000 MAX_PAGES_PER_KEYWORD=5 diff --git a/SCRAPER.md b/SCRAPER.md index 75025cb..2378208 100644 --- a/SCRAPER.md +++ b/SCRAPER.md @@ -195,6 +195,26 @@ Docker: há um `Dockerfile` em `scraper-go/`. No Docker Compose, configure `VALK No Compose da raiz, o serviço escuta em . +### Limites globais de execução + +O scraper possui um orçamento global de concorrência por execução controlado por `SCRAPER_MAX_CONCURRENCY`. + +- Padrão interno: `12`, usado quando a variável não está definida. +- Valor válido: inteiro positivo. +- Valores inválidos explícitos (`""`, `0`, negativo ou não numérico) fazem a aplicação falhar no startup, sem fallback silencioso. +- Cron e `POST /admin/scrape` usam o limite global configurado. +- `POST /scrape` preserva o contrato atual: quando `maxConcurrency` não é informado, ou vem como `0`/negativo, usa o limite global; quando vem positivo abaixo do teto, usa o valor solicitado; quando vem acima do teto, usa o teto global. +- A concorrência efetiva é calculada antes da chave de cache e é o mesmo valor usado pelo pipeline, logs e semáforo. + +O semáforo global atual é criado uma vez por chamada do pipeline. Portanto, o limite é por execução: duas execuções simultâneas ainda podem possuir dois semáforos independentes com a mesma capacidade. Lock entre cron/manual, prevenção de simultaneidade e limites por provider pertencem às próximas sub-issues. + +No Docker Compose de produção, o serviço `scraper-go` também define: + +- `GOMAXPROCS=2`: limita a quantidade de threads do Go executando código simultaneamente. +- `GOMEMLIMIT=1500MiB`: define uma meta de memória para o runtime e influencia o garbage collector. +- `mem_limit: 2g`: limite externo do container. `GOMEMLIMIT` não substitui esse limite; ele fica abaixo de `2g` para preservar margem operacional. +- `cpus: 1.5`: limita o container a 1,5 CPU. + ## Endpoints HTTP O serviço expõe endpoints HTTP (implementação em `cmd/server` e arquivos associados). Principais rotas: @@ -271,6 +291,7 @@ Fluxo principal: Concorrência e resiliência: - Semáforos por adaptador (ex.: LinkedIn usa um semáforo de 5 simultâneos para proteção). +- Orçamento global por execução via `SCRAPER_MAX_CONCURRENCY`, aplicado antes do cache e do pipeline. - Tratamento de status 429 com backoff; aborta apenas a keyword afetada em caso de falhas persistentes. - Uso de `inflight` para evitar que múltiplas requisições idênticas disparem scrapes simultâneos. - Slots rotativos reduzem o número de keywords/queries por rodada em fontes caras, preservando cobertura progressiva em execuções futuras. @@ -319,10 +340,24 @@ Boas práticas nos adaptadores: - Há testes e fixtures (ex.: `internal/keywords/keywords.test.json`) para validar normalização. - Recomenda-se executar `go test ./...` dentro de `scraper-go`. +- Para validar a configuração final do Compose, execute na raiz: + +```bash +docker compose \ + -f docker-compose.infra.yml \ + -f docker-compose.yml \ + -f docker-compose.migrate.yml \ + config +``` + +Confirme no serviço `scraper-go` os equivalentes de `SCRAPER_MAX_CONCURRENCY=12`, `GOMAXPROCS=2`, `GOMEMLIMIT=1500MiB`, `cpus: 1.5` e `mem_limit: 2g`. ## Variáveis de ambiente importantes - `VALKEY_URL` — conexão Redis/Valkey. Em Docker Compose, use `redis://valkey:6379/0`; em execução local fora do Docker, use uma URL acessível pelo host, por exemplo `redis://localhost:6379/0`. +- `SCRAPER_MAX_CONCURRENCY` — teto global de concorrência por execução. Padrão: `12`. Configuração explícita inválida impede a inicialização. +- `GOMAXPROCS` — limite efetivo de threads executando código Go simultaneamente. Valor inicial no Compose: `2`. +- `GOMEMLIMIT` — meta de memória do runtime/GC. Valor inicial no Compose: `1500MiB`; não substitui `mem_limit` do container. - `JOOBLE_API_KEY` — Jooble integration. - `ADZUNA_APP_ID` / `ADZUNA_APP_KEY` — Adzuna API. - `LINKEDIN_KEYWORD_SLOT_SIZE` — quantidade máxima de keywords do LinkedIn por execução quando a busca vier com uma lista grande. Padrão: `30`. diff --git a/docker-compose.yml b/docker-compose.yml index b602607..f658b5e 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -4,6 +4,8 @@ services: context: ./scraper-go dockerfile: Dockerfile container_name: vagas-scraper-go + cpus: 1.5 + mem_limit: 2g env_file: - ./.env volumes: @@ -11,6 +13,9 @@ services: environment: - GO_SCRAPER_ADDR=:8081 - VALKEY_URL=redis://valkey:6379/0 + - SCRAPER_MAX_CONCURRENCY=${SCRAPER_MAX_CONCURRENCY-12} + - GOMAXPROCS=${GOMAXPROCS:-2} + - GOMEMLIMIT=${GOMEMLIMIT:-1500MiB} - GUPY_ENABLED=${GUPY_ENABLED:-true} - GUPY_RAW_DISCOVERY_ENABLED=${GUPY_RAW_DISCOVERY_ENABLED:-true} - GUPY_FULL_SWEEP_ENABLED=${GUPY_FULL_SWEEP_ENABLED:-true} diff --git a/scraper-go/cmd/server/handlers.go b/scraper-go/cmd/server/handlers.go index 37adf66..32f86f7 100644 --- a/scraper-go/cmd/server/handlers.go +++ b/scraper-go/cmd/server/handlers.go @@ -9,6 +9,7 @@ import ( "github.com/redis/go-redis/v9" "github.com/Benevanio/Jobs_Scraper_Global/scraper-go/internal/cache" + "github.com/Benevanio/Jobs_Scraper_Global/scraper-go/internal/config" "github.com/Benevanio/Jobs_Scraper_Global/scraper-go/internal/domain" "github.com/Benevanio/Jobs_Scraper_Global/scraper-go/internal/keywords" "github.com/Benevanio/Jobs_Scraper_Global/scraper-go/internal/pipeline" @@ -20,7 +21,7 @@ const ( scrapeTimeout = 15 * time.Minute ) -func handleScrape(adapterList []ports.JobSource, kwStore *keywords.Store, c cache.Cache, rdb *redis.Client) http.HandlerFunc { +func handleScrape(adapterList []ports.JobSource, kwStore *keywords.Store, c cache.Cache, rdb *redis.Client, runtimeCfg config.RuntimeConfig) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { var req domain.ScrapeRequest if err := json.NewDecoder(r.Body).Decode(&req); err != nil { @@ -40,25 +41,12 @@ func handleScrape(adapterList []ports.JobSource, kwStore *keywords.Store, c cach ctx, cancel := context.WithTimeout(r.Context(), scrapeTimeout) defer cancel() - config := pipeline.SearchConfig{ - Keywords: req.Keywords, - SearchLocation: req.SearchLocation, - SearchGeoID: req.SearchGeoID, - SearchLanguage: req.SearchLanguage, - JobTypes: req.JobTypes, - TimeFilter: req.TimeFilter, - RemoteOnly: req.RemoteOnly, - Sources: req.Sources, - ResultsPerPage: req.ResultsPerPage, - MaxPagesPerKeyword: req.MaxPagesPerKeyword, - WaitBetweenSearchesMs: req.WaitBetweenSearchesMs, - PageTimeoutMs: req.PageTimeoutMs, - MaxConcurrency: req.MaxConcurrency, - } + searchConfig := searchConfigFromRequest(req, runtimeCfg.MaxConcurrency) + slogScrapeStart("public_endpoint", runtimeCfg.MaxConcurrency, req.MaxConcurrency, searchConfig.MaxConcurrency, len(searchConfig.Keywords), len(adapterList)) start := time.Now() - result, err := pipeline.SearchJobs(ctx, c, config, adapterList, scrapeTTL, rdb) + result, err := pipeline.SearchJobs(ctx, c, searchConfig, adapterList, scrapeTTL, rdb) if err != nil { http.Error(w, "Erro ao buscar vagas.", http.StatusInternalServerError) return @@ -76,6 +64,24 @@ func handleScrape(adapterList []ports.JobSource, kwStore *keywords.Store, c cach } } +func searchConfigFromRequest(req domain.ScrapeRequest, globalMaxConcurrency int) pipeline.SearchConfig { + return pipeline.SearchConfig{ + Keywords: req.Keywords, + SearchLocation: req.SearchLocation, + SearchGeoID: req.SearchGeoID, + SearchLanguage: req.SearchLanguage, + JobTypes: req.JobTypes, + TimeFilter: req.TimeFilter, + RemoteOnly: req.RemoteOnly, + Sources: req.Sources, + ResultsPerPage: req.ResultsPerPage, + MaxPagesPerKeyword: req.MaxPagesPerKeyword, + WaitBetweenSearchesMs: req.WaitBetweenSearchesMs, + PageTimeoutMs: req.PageTimeoutMs, + MaxConcurrency: config.ResolveEffectiveConcurrency(req.MaxConcurrency, globalMaxConcurrency), + } +} + func handleHealth(c cache.Cache) http.HandlerFunc { return func(w http.ResponseWriter, _ *http.Request) { w.Header().Set("Content-Type", "application/json") diff --git a/scraper-go/cmd/server/handlers_test.go b/scraper-go/cmd/server/handlers_test.go new file mode 100644 index 0000000..22ba7b6 --- /dev/null +++ b/scraper-go/cmd/server/handlers_test.go @@ -0,0 +1,56 @@ +package main + +import ( + "testing" + + "github.com/Benevanio/Jobs_Scraper_Global/scraper-go/internal/domain" + "github.com/Benevanio/Jobs_Scraper_Global/scraper-go/internal/pipeline" + "github.com/stretchr/testify/assert" +) + +func TestSearchConfigFromRequestUsesGlobalLimitWhenRequestMissing(t *testing.T) { + cfg := searchConfigFromRequest(domain.ScrapeRequest{}, 12) + + assert.Equal(t, 12, cfg.MaxConcurrency) +} + +func TestSearchConfigFromRequestUsesGlobalLimitWhenRequestIsNotPositive(t *testing.T) { + assert.Equal(t, 12, searchConfigFromRequest(domain.ScrapeRequest{MaxConcurrency: 0}, 12).MaxConcurrency) + assert.Equal(t, 12, searchConfigFromRequest(domain.ScrapeRequest{MaxConcurrency: -1}, 12).MaxConcurrency) +} + +func TestSearchConfigFromRequestPreservesRequestBelowLimit(t *testing.T) { + cfg := searchConfigFromRequest(domain.ScrapeRequest{MaxConcurrency: 8}, 12) + + assert.Equal(t, 8, cfg.MaxConcurrency) +} + +func TestSearchConfigFromRequestCapsRequestAboveLimitBeforeCacheKey(t *testing.T) { + req := domain.ScrapeRequest{ + Keywords: []string{"go"}, + SearchLocation: "Brasil", + MaxConcurrency: 40, + } + + cfg := searchConfigFromRequest(req, 12) + + assert.Equal(t, 12, cfg.MaxConcurrency) + assert.Contains(t, pipeline.BuildCacheKey(cfg), ":12") +} + +func TestSearchConfigFromRequestUsesSameCacheKeyForRequestsAboveLimit(t *testing.T) { + base := domain.ScrapeRequest{ + Keywords: []string{"go"}, + SearchLocation: "Brasil", + } + + req40 := base + req40.MaxConcurrency = 40 + req100 := base + req100.MaxConcurrency = 100 + + key40 := pipeline.BuildCacheKey(searchConfigFromRequest(req40, 12)) + key100 := pipeline.BuildCacheKey(searchConfigFromRequest(req100, 12)) + + assert.Equal(t, key40, key100) +} diff --git a/scraper-go/cmd/server/main.go b/scraper-go/cmd/server/main.go index 57692f5..eeefc5c 100644 --- a/scraper-go/cmd/server/main.go +++ b/scraper-go/cmd/server/main.go @@ -3,6 +3,11 @@ package main import ( "log/slog" "os" + "runtime" + "runtime/debug" + "strconv" + + "github.com/Benevanio/Jobs_Scraper_Global/scraper-go/internal/config" ) func main() { @@ -11,6 +16,13 @@ func main() { loadEnv() + runtimeCfg, err := config.LoadRuntimeConfig() + if err != nil { + slog.Error("configuração inválida do scraper", "error", err) + os.Exit(1) + } + logRuntimeConfig(runtimeCfg) + // newRedisClient() está em server.go — usa ParseURL corretamente // e valida a conexão com Ping antes de retornar. rdb, err := newRedisClient() @@ -25,5 +37,40 @@ func main() { adapterList := buildAdapters(rdb) slog.Info("servidor inicializado", "adapters_total", len(adapterList)) - run(adapterList) + run(adapterList, runtimeCfg) +} + +func logRuntimeConfig(cfg config.RuntimeConfig) { + _, gomaxprocsSet := os.LookupEnv("GOMAXPROCS") + _, gomemlimitSet := os.LookupEnv("GOMEMLIMIT") + + memLimit := debug.SetMemoryLimit(-1) + + slog.Info("scraper runtime configurado", + "max_concurrency", cfg.MaxConcurrency, + "max_concurrency_source", cfg.MaxConcurrencySource, + "gomaxprocs_effective", runtime.GOMAXPROCS(0), + "gomaxprocs_source", envSource(gomaxprocsSet), + "gomemlimit_effective_bytes", memLimit, + "gomemlimit_effective", formatBytes(memLimit), + "gomemlimit_source", envSource(gomemlimitSet), + ) +} + +func envSource(set bool) string { + if set { + return "environment" + } + return "go_runtime_default" +} + +func formatBytes(value int64) string { + if value < 0 { + return "unlimited" + } + const mib = 1024 * 1024 + if value%mib == 0 { + return strconv.FormatInt(value/mib, 10) + "MiB" + } + return strconv.FormatInt(value, 10) + "B" } diff --git a/scraper-go/cmd/server/runtime_log.go b/scraper-go/cmd/server/runtime_log.go new file mode 100644 index 0000000..28f17e7 --- /dev/null +++ b/scraper-go/cmd/server/runtime_log.go @@ -0,0 +1,17 @@ +package main + +import "log/slog" + +func slogScrapeStart(origin string, configured, requested, effective, keywords, adapters int) { + attrs := []any{ + "origin", origin, + "max_concurrency_configured", configured, + "max_concurrency_effective", effective, + "keywords", keywords, + "adapters", adapters, + } + if requested != 0 { + attrs = append(attrs, "max_concurrency_requested", requested) + } + slog.Info("scraper execução iniciada", attrs...) +} diff --git a/scraper-go/cmd/server/server.go b/scraper-go/cmd/server/server.go index fabd9e4..9fe15a8 100644 --- a/scraper-go/cmd/server/server.go +++ b/scraper-go/cmd/server/server.go @@ -11,6 +11,7 @@ import ( "time" "github.com/Benevanio/Jobs_Scraper_Global/scraper-go/internal/cache" + "github.com/Benevanio/Jobs_Scraper_Global/scraper-go/internal/config" "github.com/Benevanio/Jobs_Scraper_Global/scraper-go/internal/cronjob" "github.com/Benevanio/Jobs_Scraper_Global/scraper-go/internal/jobstore" "github.com/Benevanio/Jobs_Scraper_Global/scraper-go/internal/keywords" @@ -19,7 +20,7 @@ import ( "github.com/redis/go-redis/v9" ) -func run(adapterList []ports.JobSource) { +func run(adapterList []ports.JobSource, runtimeCfg config.RuntimeConfig) { addr := os.Getenv("GO_SCRAPER_ADDR") if addr == "" { addr = ":8081" @@ -44,6 +45,7 @@ func run(adapterList []ports.JobSource) { // ── Scheduler (cronjob) ── schedulerCfg := cronjob.DefaultConfig() + schedulerCfg.MaxConcurrency = runtimeCfg.MaxConcurrency scheduler := cronjob.New(schedulerCfg, kwStore, jobStore, adapterList, rdb) scheduler.OnComplete = func(kws []string, scraped, saved int, duration time.Duration) { @@ -54,7 +56,7 @@ func run(adapterList []ports.JobSource) { mux := http.NewServeMux() // Públicas - mux.Handle("POST /scrape", handleScrape(adapterList, kwStore, c, rdb)) + mux.Handle("POST /scrape", handleScrape(adapterList, kwStore, c, rdb, runtimeCfg)) mux.Handle("GET /health", handleHealth(c)) mux.Handle("GET /metrics", promhttp.Handler()) mux.Handle("GET /api/keywords", handleGetKeywords(kwStore)) diff --git a/scraper-go/internal/config/config.go b/scraper-go/internal/config/config.go new file mode 100644 index 0000000..f9d2b26 --- /dev/null +++ b/scraper-go/internal/config/config.go @@ -0,0 +1,64 @@ +package config + +import ( + "fmt" + "os" + "strconv" + "strings" +) + +const ( + DefaultMaxConcurrency = 12 + + SourceEnvironment = "environment" + SourceInternalDefault = "internal_default" + + ScraperMaxConcurrencyEnv = "SCRAPER_MAX_CONCURRENCY" +) + +type RuntimeConfig struct { + MaxConcurrency int + MaxConcurrencySource string +} + +func LoadRuntimeConfig() (RuntimeConfig, error) { + return LoadRuntimeConfigFromLookup(os.LookupEnv) +} + +func LoadRuntimeConfigFromLookup(lookup func(string) (string, bool)) (RuntimeConfig, error) { + value, ok := lookup(ScraperMaxConcurrencyEnv) + if !ok { + return RuntimeConfig{ + MaxConcurrency: DefaultMaxConcurrency, + MaxConcurrencySource: SourceInternalDefault, + }, nil + } + + trimmed := strings.TrimSpace(value) + if trimmed == "" { + return RuntimeConfig{}, fmt.Errorf("%s must be a positive integer", ScraperMaxConcurrencyEnv) + } + + parsed, err := strconv.Atoi(trimmed) + if err != nil { + return RuntimeConfig{}, fmt.Errorf("%s must be a positive integer: %w", ScraperMaxConcurrencyEnv, err) + } + if parsed <= 0 { + return RuntimeConfig{}, fmt.Errorf("%s must be greater than zero", ScraperMaxConcurrencyEnv) + } + + return RuntimeConfig{ + MaxConcurrency: parsed, + MaxConcurrencySource: SourceEnvironment, + }, nil +} + +func ResolveEffectiveConcurrency(requested, globalMax int) int { + if requested <= 0 { + return globalMax + } + if requested > globalMax { + return globalMax + } + return requested +} diff --git a/scraper-go/internal/config/config_test.go b/scraper-go/internal/config/config_test.go new file mode 100644 index 0000000..86bf76e --- /dev/null +++ b/scraper-go/internal/config/config_test.go @@ -0,0 +1,58 @@ +package config + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestLoadRuntimeConfigUsesDefaultWhenEnvMissing(t *testing.T) { + cfg, err := LoadRuntimeConfigFromLookup(func(string) (string, bool) { + return "", false + }) + + require.NoError(t, err) + assert.Equal(t, 12, cfg.MaxConcurrency) + assert.Equal(t, SourceInternalDefault, cfg.MaxConcurrencySource) +} + +func TestLoadRuntimeConfigUsesEnvValue(t *testing.T) { + cfg, err := LoadRuntimeConfigFromLookup(func(string) (string, bool) { + return "8", true + }) + + require.NoError(t, err) + assert.Equal(t, 8, cfg.MaxConcurrency) + assert.Equal(t, SourceEnvironment, cfg.MaxConcurrencySource) +} + +func TestLoadRuntimeConfigRejectsInvalidEnvValues(t *testing.T) { + cases := []struct { + name string + value string + }{ + {name: "empty", value: ""}, + {name: "blank", value: " "}, + {name: "zero", value: "0"}, + {name: "negative", value: "-1"}, + {name: "not numeric", value: "abc"}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + _, err := LoadRuntimeConfigFromLookup(func(string) (string, bool) { + return tc.value, true + }) + + require.Error(t, err) + }) + } +} + +func TestResolveEffectiveConcurrency(t *testing.T) { + assert.Equal(t, 12, ResolveEffectiveConcurrency(0, 12)) + assert.Equal(t, 12, ResolveEffectiveConcurrency(-1, 12)) + assert.Equal(t, 8, ResolveEffectiveConcurrency(8, 12)) + assert.Equal(t, 12, ResolveEffectiveConcurrency(40, 12)) +} diff --git a/scraper-go/internal/cronjob/cronjob.go b/scraper-go/internal/cronjob/cronjob.go index a6ef559..dc04ef6 100644 --- a/scraper-go/internal/cronjob/cronjob.go +++ b/scraper-go/internal/cronjob/cronjob.go @@ -8,6 +8,7 @@ import ( "github.com/redis/go-redis/v9" + "github.com/Benevanio/Jobs_Scraper_Global/scraper-go/internal/config" "github.com/Benevanio/Jobs_Scraper_Global/scraper-go/internal/jobstore" "github.com/Benevanio/Jobs_Scraper_Global/scraper-go/internal/keywords" "github.com/Benevanio/Jobs_Scraper_Global/scraper-go/internal/pipeline" @@ -32,7 +33,7 @@ func DefaultConfig() Config { JobTypes: "C,F", TimeFilter: "r604800", RemoteOnly: false, - MaxConcurrency: 40, + MaxConcurrency: config.DefaultMaxConcurrency, } } @@ -66,7 +67,7 @@ func (s *Scheduler) Start(ctx context.Context) { slog.Info("cronjob: scheduler iniciado", "interval", s.cfg.Interval) go func() { - s.run(ctx) + s.run(ctx, "cron") ticker := time.NewTicker(s.cfg.Interval) defer ticker.Stop() @@ -74,7 +75,7 @@ func (s *Scheduler) Start(ctx context.Context) { for { select { case <-ticker.C: - s.run(ctx) + s.run(ctx, "cron") case <-s.stop: slog.Info("cronjob: scheduler encerrado") return @@ -98,7 +99,7 @@ func (s *Scheduler) RunNow(ctx context.Context) error { } s.mu.Unlock() - go s.run(ctx) + go s.run(ctx, "admin_manual") return nil } @@ -114,7 +115,7 @@ func (s *Scheduler) Snapshot() (running bool, lastRunAt time.Time, jobsCollected return s.running, s.lastRunAt, s.lastJobs } -func (s *Scheduler) run(ctx context.Context) { +func (s *Scheduler) run(ctx context.Context, origin string) { s.mu.Lock() if s.running { s.mu.Unlock() @@ -131,8 +132,6 @@ func (s *Scheduler) run(ctx context.Context) { }() start := time.Now() - slog.Info("cronjob: iniciando execução") - scrapeCtx, cancel := context.WithTimeout(ctx, s.cfg.ScrapeTimeout) defer cancel() @@ -142,14 +141,14 @@ func (s *Scheduler) run(ctx context.Context) { return } - config := pipeline.SearchConfig{ - Keywords: kws, - SearchLocation: s.cfg.SearchLocation, - JobTypes: s.cfg.JobTypes, - TimeFilter: s.cfg.TimeFilter, - RemoteOnly: s.cfg.RemoteOnly, - MaxConcurrency: s.cfg.MaxConcurrency, - } + config := s.searchConfig(kws) + slog.Info("scraper execução iniciada", + "origin", origin, + "max_concurrency_configured", s.cfg.MaxConcurrency, + "max_concurrency_effective", config.MaxConcurrency, + "keywords", len(kws), + "adapters", len(s.adapterList), + ) jobs, err := pipeline.ScrapeAllSources(scrapeCtx, config, s.adapterList, s.rdb) if err != nil { @@ -185,6 +184,17 @@ func (s *Scheduler) run(ctx context.Context) { } } +func (s *Scheduler) searchConfig(kws []string) pipeline.SearchConfig { + return pipeline.SearchConfig{ + Keywords: kws, + SearchLocation: s.cfg.SearchLocation, + JobTypes: s.cfg.JobTypes, + TimeFilter: s.cfg.TimeFilter, + RemoteOnly: s.cfg.RemoteOnly, + MaxConcurrency: s.cfg.MaxConcurrency, + } +} + type alreadyRunningError struct{} func (e alreadyRunningError) Error() string { diff --git a/scraper-go/internal/cronjob/cronjob_test.go b/scraper-go/internal/cronjob/cronjob_test.go new file mode 100644 index 0000000..be0701d --- /dev/null +++ b/scraper-go/internal/cronjob/cronjob_test.go @@ -0,0 +1,33 @@ +package cronjob + +import ( + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestDefaultConfigUsesSafeMaxConcurrency(t *testing.T) { + cfg := DefaultConfig() + + assert.Equal(t, 12, cfg.MaxConcurrency) +} + +func TestSchedulerSearchConfigReceivesGlobalMaxConcurrency(t *testing.T) { + cfg := DefaultConfig() + cfg.MaxConcurrency = 9 + + scheduler := New(cfg, nil, nil, nil, nil) + searchConfig := scheduler.searchConfig([]string{"go"}) + + assert.Equal(t, 9, searchConfig.MaxConcurrency) + assert.Equal(t, []string{"go"}, searchConfig.Keywords) +} + +func TestAdminManualSharesSchedulerConfig(t *testing.T) { + cfg := DefaultConfig() + cfg.MaxConcurrency = 7 + + scheduler := New(cfg, nil, nil, nil, nil) + + assert.Equal(t, 7, scheduler.cfg.MaxConcurrency) +} diff --git a/scraper-go/internal/pipeline/cache_key_test.go b/scraper-go/internal/pipeline/cache_key_test.go index e01a6f9..0c33094 100644 --- a/scraper-go/internal/pipeline/cache_key_test.go +++ b/scraper-go/internal/pipeline/cache_key_test.go @@ -1,6 +1,9 @@ package pipeline -import "testing" +import ( + "strings" + "testing" +) func TestBuildCacheKeyNormalizesAndDeduplicatesKeywords(t *testing.T) { a := BuildCacheKey(SearchConfig{ @@ -77,3 +80,15 @@ func TestBuildCacheKeyNormalizesSourcesOrder(t *testing.T) { t.Fatalf("expected equivalent source sets to share cache key, got %q and %q", a, b) } } + +func TestBuildCacheKeyUsesEffectiveMaxConcurrency(t *testing.T) { + key := BuildCacheKey(SearchConfig{ + Keywords: []string{"go"}, + SearchLocation: "Brasil", + MaxConcurrency: 12, + }) + + if !strings.HasSuffix(key, ":12") { + t.Fatalf("expected cache key to contain effective max concurrency, got %q", key) + } +} diff --git a/scraper-go/internal/pipeline/pipeline.go b/scraper-go/internal/pipeline/pipeline.go index 87cff9e..690c64d 100644 --- a/scraper-go/internal/pipeline/pipeline.go +++ b/scraper-go/internal/pipeline/pipeline.go @@ -21,7 +21,7 @@ import ( "golang.org/x/text/unicode/norm" ) -const defaultMaxConcurrency = 40 +var errInvalidMaxConcurrency = fmt.Errorf("pipeline: max concurrency must be greater than zero") type result struct { jobs []domain.Job @@ -34,12 +34,12 @@ type adapterTask struct { batch bool } -func Run(ctx context.Context, adapterList []ports.JobSource, req domain.ScrapeRequest) []domain.Job { +func Run(ctx context.Context, adapterList []ports.JobSource, req domain.ScrapeRequest) ([]domain.Job, error) { pipelineStart := time.Now() maxConcurrency := req.MaxConcurrency if maxConcurrency <= 0 { - maxConcurrency = defaultMaxConcurrency + return nil, errInvalidMaxConcurrency } tasks := make([]adapterTask, 0, len(adapterList)*len(req.Keywords)) @@ -116,7 +116,7 @@ func Run(ctx context.Context, adapterList []ports.JobSource, req domain.ScrapeRe metrics.PipelineRunDuration.Observe(time.Since(pipelineStart).Seconds()) metrics.PipelineJobsTotal.Observe(float64(len(classified))) - return classified + return classified, nil } func runAdapterTask(ctx context.Context, t adapterTask, req domain.ScrapeRequest) ([]domain.Job, error) { diff --git a/scraper-go/internal/pipeline/scrape.go b/scraper-go/internal/pipeline/scrape.go index 0665fb7..fbe3b25 100644 --- a/scraper-go/internal/pipeline/scrape.go +++ b/scraper-go/internal/pipeline/scrape.go @@ -58,7 +58,10 @@ func ScrapeAllSources( MaxConcurrency: config.MaxConcurrency, } - jobs := Run(ctx, adapterList, req) + jobs, err := Run(ctx, adapterList, req) + if err != nil { + return nil, err + } slog.Info("scrape finished", "total_jobs", len(jobs), diff --git a/scraper-go/internal/pipeline/source_schedule_test.go b/scraper-go/internal/pipeline/source_schedule_test.go index 13374de..79623e7 100644 --- a/scraper-go/internal/pipeline/source_schedule_test.go +++ b/scraper-go/internal/pipeline/source_schedule_test.go @@ -89,11 +89,35 @@ func TestShouldRunJoobleAllowsWhenRedisUnavailable(t *testing.T) { func TestRunUsesBatchAdapterOnceForAllKeywords(t *testing.T) { adapter := &batchRunTestAdapter{} - Run(context.Background(), []adapters.Adapter{adapter}, domain.ScrapeRequest{ - Keywords: []string{"go", "java", "python"}, + _, err := Run(context.Background(), []adapters.Adapter{adapter}, domain.ScrapeRequest{ + Keywords: []string{"go", "java", "python"}, + MaxConcurrency: 1, }) + require.NoError(t, err) assert.Equal(t, 0, adapter.searchCalls) assert.Equal(t, 1, adapter.batchCalls) assert.Equal(t, []string{"go", "java", "python"}, adapter.keywords) } + +func TestRunRejectsInvalidMaxConcurrency(t *testing.T) { + _, err := Run(context.Background(), []adapters.Adapter{&batchRunTestAdapter{}}, domain.ScrapeRequest{ + Keywords: []string{"go"}, + }) + + require.Error(t, err) + assert.ErrorContains(t, err, "max concurrency") +} + +func TestRunExecutesWithValidMaxConcurrency(t *testing.T) { + adapter := &batchRunTestAdapter{} + + jobs, err := Run(context.Background(), []adapters.Adapter{adapter}, domain.ScrapeRequest{ + Keywords: []string{"go"}, + MaxConcurrency: 1, + }) + + require.NoError(t, err) + assert.Empty(t, jobs) + assert.Equal(t, 1, adapter.batchCalls) +}