diff --git a/cmd/main.go b/cmd/main.go index 92ae164..1fd8de8 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -1,24 +1,17 @@ package main import ( - "context" - "os" - "os/signal" - "syscall" "task-processor/config" "task-processor/internal/app" "task-processor/internal/logging" ) func main() { - ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) - defer cancel() - c := config.NewConfig() logging.LogSetup(c.LogLevel) appl := app.New(c) - appl.Run(ctx) + appl.Run() } diff --git a/internal/app/app.go b/internal/app/app.go index 832e338..6353e7c 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -3,9 +3,11 @@ package app import ( "context" log "github.com/sirupsen/logrus" - "google.golang.org/grpc" "net" + "os" + "os/signal" "runtime" + "syscall" "task-processor/config" "task-processor/internal/appState" "task-processor/internal/parsers" @@ -26,8 +28,6 @@ type App struct { state *appState.State network *remote.Network numCPUs int - metricsSrv *router.Handler - taskApiSrv *grpc.Server } func New(c *config.Config) *App { @@ -38,14 +38,6 @@ func New(c *config.Config) *App { st := appState.NewState(numCPUs, c.CheckPeriod, c.TasksCfg.RetryCount, c.TasksCfg.RetryMinutes) - server := newServer(st) - - //metrics - mSrv := router.NewHandler(router.Deps{ - Addr: net.JoinHostPort(c.Metrics.Host, c.Metrics.Port), - GinMode: c.Metrics.GinMode, - }) - return &App{ config: c, checkPeriod: time.Duration(c.CheckPeriod), @@ -55,31 +47,33 @@ func New(c *config.Config) *App { state: st, network: remote.NewHandler(), numCPUs: numCPUs, - metricsSrv: mSrv, - taskApiSrv: server, } } -func (app *App) Run(ctx context.Context) { +func (app *App) Run() { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + log.Info("Application start") - - addr := net.JoinHostPort(app.config.GrpcCfg.ServerHost, app.config.GrpcCfg.ServerPort) - log.WithFields(log.Fields{ - "Service address": addr, + "Service address": app.config.GrpcCfg.ServerHost + ":" + app.config.GrpcCfg.ServerPort, "Number of CPUs": app.numCPUs, }).Debug("App settings") - errChan := make(chan error, 16) + //metrics + mSrv := router.NewHandler(router.Deps{ + Addr: net.JoinHostPort(app.config.Metrics.Host, app.config.Metrics.Port), + GinMode: app.config.Metrics.GinMode, + }) //main + server := newServer(app) apiClient := newApiClient(app.config.GrpcCfg.ApiClientHost + ":" + app.config.GrpcCfg.ApiClientPort) period := time.NewTicker(app.checkPeriod * time.Hour) defer period.Stop() sender := make(chan shared.TaskResult, app.numCPUs*10) - defer close(sender) // external scrapper surugayaScrapper := newSurugayaScrapperClient(app.config.GrpcCfg.SurugayaScrapperHost + ":" + app.config.GrpcCfg.SurugayaScrapperPort) @@ -88,11 +82,11 @@ func (app *App) Run(ctx context.Context) { handlers := make(map[string]parsers.TaskHandler) if app.config.OriginEnabled.Surugaya { - handlers[shared.OriginSurugaya] = parsers.NewSurugayaParser(surugayaScrapper) + handlers[shared.OriginSurugaya] = parsers.NewSurugayaParser(ctx, surugayaScrapper) } if app.config.OriginEnabled.Mandarake { - handlers[shared.OriginMandarake] = mandarake.NewParser(mandarake.Deps{ + handlers[shared.OriginMandarake] = mandarake.NewParser(mandarake.ParserDeps{ Enabled: app.config.OriginEnabled.Mandarake, ExternalBrowser: app.config.ExternalBrowser, GoroutinesNumber: app.numCPUs, @@ -114,7 +108,7 @@ func (app *App) Run(ctx context.Context) { receivedTasks := app.network.RequestTasks(ctx, apiClient) log.WithField("length", len(receivedTasks)).Debug("End receiving") - taskProcessor.StartWork(ctx, receivedTasks) + taskProcessor.StartWork(receivedTasks) } go func() { @@ -149,38 +143,37 @@ func (app *App) Run(ctx context.Context) { //start metrics server go func() { - if err := app.metricsSrv.Run(); err != nil { - errChan <- err + if err := mSrv.Run(); err != nil { + log.WithError(err).Error("Metrics server run failed") } }() //gRPC Server for status response go func() { - listener, err := net.Listen("tcp", addr) + listener, err := net.Listen("tcp", app.config.GrpcCfg.ServerHost+":"+app.config.GrpcCfg.ServerPort) if err != nil { - errChan <- err + log.Fatalf("failed to listen: %v", err) } - log.Infof("gRPC Server listening at %v", addr) - if err = app.taskApiSrv.Serve(listener); err != nil { - errChan <- err + log.Infof("gRPC Server listening at %v", app.config.GrpcCfg.ServerHost+":"+app.config.GrpcCfg.ServerPort) + if err := server.Serve(listener); err != nil { + log.Fatalf("failed to serve: %v", err) } }() - select { - case <-ctx.Done(): - app.shutdown(ctx) - case err := <-errChan: - log.WithError(err).Fatal("Application run error") - } -} - -func (app *App) shutdown(ctx context.Context) { - log.Info("Shutting down...") - - app.taskApiSrv.GracefulStop() - - if err := app.metricsSrv.Shutdown(ctx); err != nil { - log.WithError(err).Error("Failed to shutdown server") - } + go func() { + sigint := make(chan os.Signal, 1) + signal.Notify(sigint, os.Interrupt, syscall.SIGTERM) + <-sigint + log.Info("Shutting down...") + + period.Stop() + server.GracefulStop() + cancel() + if err := mSrv.Shutdown(ctx); err != nil { + log.WithError(err).Error("Failed to shutdown server") + } + }() + + <-ctx.Done() } diff --git a/internal/app/server.go b/internal/app/server.go index 9c9dadc..f837430 100644 --- a/internal/app/server.go +++ b/internal/app/server.go @@ -12,10 +12,10 @@ type Server struct { state *appState.State } -func newServer(state *appState.State) *grpc.Server { +func newServer(app *App) *grpc.Server { s := grpc.NewServer() srv := &Server{ - state: state, + state: app.state, } pb.RegisterTaskProcessorServer(s, srv) return s diff --git a/internal/parsers/interface.go b/internal/parsers/interface.go index 1a4c6fe..453d619 100644 --- a/internal/parsers/interface.go +++ b/internal/parsers/interface.go @@ -1,11 +1,10 @@ package parsers import ( - "context" "task-processor/internal/appState" "task-processor/internal/shared" ) type TaskHandler interface { - HandleTasks(ctx context.Context, tasks []shared.Task, sender chan shared.TaskResult, state *appState.State) + HandleTasks(tasks []shared.Task, sender chan shared.TaskResult, state *appState.State) } diff --git a/internal/parsers/mandarake/handleTasks.go b/internal/parsers/mandarake/handleTasks.go index 1b566e7..1920554 100644 --- a/internal/parsers/mandarake/handleTasks.go +++ b/internal/parsers/mandarake/handleTasks.go @@ -4,20 +4,15 @@ import ( "context" "github.com/chromedp/chromedp" log "github.com/sirupsen/logrus" - "runtime" "sync" "task-processor/internal/appState" "task-processor/internal/shared" ) -func (s *Parser) HandleTasks(ctx context.Context, tasks []shared.Task, sender chan shared.TaskResult, state *appState.State) { - log.Infof("%v %v handling tasks", logHeader, logWorker) +func (s *Parser) HandleTasks(tasks []shared.Task, sender chan shared.TaskResult, state *appState.State) { + log.Debug(logHeader + logWorker + "handling tasks") - allocCtx, allocCancel := chromedp.NewRemoteAllocator(ctx, s.externalBrowser) - defer allocCancel() - - sessionCtx, sessionCancel := chromedp.NewContext(allocCtx /* chromedp.WithLogf(log.Printf) */, chromedp.WithLogf(func(string, ...any) {})) - defer sessionCancel() + allocCtx, allocCancel := chromedp.NewRemoteAllocator(s.baseCtx, s.externalBrowser) receiver := make(chan shared.Task, len(tasks)) for _, task := range tasks { @@ -25,27 +20,41 @@ func (s *Parser) HandleTasks(ctx context.Context, tasks []shared.Task, sender ch } close(receiver) - log.Debugf("%v gorutines before wait group: %v", logHeader, runtime.NumGoroutine()) wg := sync.WaitGroup{} for i := 0; i < s.goroutinesNumber; i++ { wg.Add(1) go func() { defer wg.Done() - s.worker(sessionCtx, receiver, sender) + s.worker(allocCtx, receiver, sender, state) }() } wg.Wait() - - log.Debugf("%v gorutines after wait group: %v", logHeader, runtime.NumGoroutine()) - log.Infof(logHeader + logWorker + "finished handling tasks") + allocCancel() + log.Debug(logHeader + logWorker + "finished handling tasks") } -func (s *Parser) worker(ctx context.Context, receiver chan shared.Task, sender chan shared.TaskResult) { +func (s *Parser) worker(ctx context.Context, receiver chan shared.Task, sender chan shared.TaskResult, state *appState.State) { for task := range receiver { - log.WithField("task_uuid", task.MerchUuid).Infof("%v %v processing task", logHeader, logWorker) + log.WithField("task_uuid", task.MerchUuid).Debug(logHeader + logWorker + "processing task") + + //pageCtx, pageCancel := chromedp.NewContext(ctx, chromedp.WithLogf(func(string, ...any) {})) + // + //price, err := s.getPrice(pageCtx, task) + //pageCancel() + + //price, err := s.getMinimalPrice(task) + //if err != nil { + // log.WithField("task_uuid", task.MerchUuid).Warn(logHeader + logWorker + logTaskWarning + "failed to process, zero price") + // sender <- shared.TaskResult{ + // MerchUuid: task.MerchUuid, + // Origin: task.Origin, + // Price: zeroPrice, + // } + // continue + //} //price will be zeroPrice value in case of any error or if price not found - price := s.getMinimalPrice(ctx, task) + price := s.getMinimalPrice(task) sender <- shared.TaskResult{ MerchUuid: task.MerchUuid, Origin: task.Origin, diff --git a/internal/parsers/mandarake/handler.go b/internal/parsers/mandarake/handler.go index cd6b272..92e24dd 100644 --- a/internal/parsers/mandarake/handler.go +++ b/internal/parsers/mandarake/handler.go @@ -1,36 +1,39 @@ package mandarake import ( + "context" log "github.com/sirupsen/logrus" ) const ( zeroPrice int32 = 0 taxMultiplier float64 = 1.1 - logHeader = "Mandarake parser |" - logWorker = "worker:" - logTaskWarning = "task warning:" - logGetPrice = "get price:" + logHeader = "Mandarake parser | " + logWorker = "worker: " + logTaskWarning = "task warning: " + logGetPrice = "get price: " ) type Parser struct { + baseCtx context.Context externalBrowser string goroutinesNumber int } -type Deps struct { +type ParserDeps struct { Enabled bool ExternalBrowser string GoroutinesNumber int } -func NewParser(deps Deps) *Parser { +func NewParser(deps ParserDeps) *Parser { if !deps.Enabled { log.Info(logHeader + "disabled") return nil } return &Parser{ + baseCtx: context.Background(), externalBrowser: deps.ExternalBrowser, goroutinesNumber: deps.GoroutinesNumber, } diff --git a/internal/parsers/mandarake/service.go b/internal/parsers/mandarake/service.go index 1e47a3c..3147fa1 100644 --- a/internal/parsers/mandarake/service.go +++ b/internal/parsers/mandarake/service.go @@ -41,7 +41,14 @@ func (s *Parser) getPrice(ctx context.Context, task shared.Task) (int32, error) return minimal, nil } -func (s *Parser) getMinimalPrice(sessionCtx context.Context, task shared.Task) int32 { +func (s *Parser) getMinimalPrice(task shared.Task) int32 { + ctx := context.Background() + allocCtx, allocCancel := chromedp.NewRemoteAllocator(ctx, s.externalBrowser) + defer allocCancel() + + sessionCtx, sessionCancel := chromedp.NewContext(allocCtx, chromedp.WithLogf(log.Printf)) + defer sessionCancel() + var ( singlePrice string rangedPrice string diff --git a/internal/parsers/mandarake/service_test.go b/internal/parsers/mandarake/service_test.go index 736331a..c0f5a79 100644 --- a/internal/parsers/mandarake/service_test.go +++ b/internal/parsers/mandarake/service_test.go @@ -65,6 +65,7 @@ func TestParser_processPrices(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { s := &Parser{ + baseCtx: tt.fields.baseCtx, externalBrowser: tt.fields.externalBrowser, goroutinesNumber: tt.fields.goroutinesNumber, } diff --git a/internal/parsers/surugaya.go b/internal/parsers/surugaya.go index 1fecccf..9dc3c5f 100644 --- a/internal/parsers/surugaya.go +++ b/internal/parsers/surugaya.go @@ -11,19 +11,21 @@ import ( type SurugayaParser struct { scrapper sc.SurugayaScrapperClient + ctx context.Context } -func NewSurugayaParser(scrapper sc.SurugayaScrapperClient) *SurugayaParser { +func NewSurugayaParser(ctx context.Context, scrapper sc.SurugayaScrapperClient) *SurugayaParser { log.Debug("Surugaya parser init") return &SurugayaParser{ scrapper: scrapper, + ctx: ctx, } } -func (s *SurugayaParser) HandleTasks(ctx context.Context, tasks []shared.Task, sender chan shared.TaskResult, state *appState.State) { +func (s *SurugayaParser) HandleTasks(tasks []shared.Task, sender chan shared.TaskResult, state *appState.State) { log.WithField("count", len(tasks)).Debug("Handling Surugaya Tasks") - stream, err := s.scrapper.ProcessTasks(ctx) + stream, err := s.scrapper.ProcessTasks(s.ctx) if err != nil { log.WithField("err", err).Error("Error creating stream") return diff --git a/internal/processor/service.go b/internal/processor/service.go index 61e7fe0..4748566 100644 --- a/internal/processor/service.go +++ b/internal/processor/service.go @@ -1,14 +1,13 @@ package processor import ( - "context" log "github.com/sirupsen/logrus" "sync" "task-processor/internal/appState" "task-processor/internal/shared" ) -func (p *Processor) StartWork(ctx context.Context, receivedTasks []shared.TaskResponse) { +func (p *Processor) StartWork(receivedTasks []shared.TaskResponse) { log.Info("Starting work...") p.state.ResetCounters() @@ -28,7 +27,7 @@ func (p *Processor) StartWork(ctx context.Context, receivedTasks []shared.TaskRe go func(origin string, tasks []shared.Task) { defer wg.Done() log.Info("Running task handler for origin: ", origin) - p.handlers[origin].HandleTasks(ctx, tasks, p.out, p.state) + p.handlers[origin].HandleTasks(tasks, p.out, p.state) }(origin, tasks) } wg.Wait()