From c955615fb1b896412eac47bfac272342c62547c2 Mon Sep 17 00:00:00 2001 From: nquidox Date: Sat, 28 Feb 2026 10:53:02 +0300 Subject: [PATCH 01/10] context fix --- cmd/main.go | 9 ++++- internal/parsers/interface.go | 3 +- internal/parsers/mandarake/handleTasks.go | 41 +++++++++------------- internal/parsers/mandarake/handler.go | 15 ++++---- internal/parsers/mandarake/service.go | 9 +---- internal/parsers/mandarake/service_test.go | 1 - internal/parsers/surugaya.go | 8 ++--- internal/processor/service.go | 5 +-- 8 files changed, 39 insertions(+), 52 deletions(-) diff --git a/cmd/main.go b/cmd/main.go index 1fd8de8..92ae164 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -1,17 +1,24 @@ 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() + appl.Run(ctx) } diff --git a/internal/parsers/interface.go b/internal/parsers/interface.go index 453d619..1a4c6fe 100644 --- a/internal/parsers/interface.go +++ b/internal/parsers/interface.go @@ -1,10 +1,11 @@ package parsers import ( + "context" "task-processor/internal/appState" "task-processor/internal/shared" ) type TaskHandler interface { - HandleTasks(tasks []shared.Task, sender chan shared.TaskResult, state *appState.State) + HandleTasks(ctx context.Context, 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 1920554..1b566e7 100644 --- a/internal/parsers/mandarake/handleTasks.go +++ b/internal/parsers/mandarake/handleTasks.go @@ -4,15 +4,20 @@ 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(tasks []shared.Task, sender chan shared.TaskResult, state *appState.State) { - log.Debug(logHeader + logWorker + "handling tasks") +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) - allocCtx, allocCancel := chromedp.NewRemoteAllocator(s.baseCtx, s.externalBrowser) + 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() receiver := make(chan shared.Task, len(tasks)) for _, task := range tasks { @@ -20,41 +25,27 @@ func (s *Parser) HandleTasks(tasks []shared.Task, sender chan shared.TaskResult, } 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(allocCtx, receiver, sender, state) + s.worker(sessionCtx, receiver, sender) }() } wg.Wait() - allocCancel() - log.Debug(logHeader + logWorker + "finished handling tasks") + + log.Debugf("%v gorutines after wait group: %v", logHeader, runtime.NumGoroutine()) + log.Infof(logHeader + logWorker + "finished handling tasks") } -func (s *Parser) worker(ctx context.Context, receiver chan shared.Task, sender chan shared.TaskResult, state *appState.State) { +func (s *Parser) worker(ctx context.Context, receiver chan shared.Task, sender chan shared.TaskResult) { for task := range receiver { - 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 - //} + log.WithField("task_uuid", task.MerchUuid).Infof("%v %v processing task", logHeader, logWorker) //price will be zeroPrice value in case of any error or if price not found - price := s.getMinimalPrice(task) + price := s.getMinimalPrice(ctx, 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 92e24dd..cd6b272 100644 --- a/internal/parsers/mandarake/handler.go +++ b/internal/parsers/mandarake/handler.go @@ -1,39 +1,36 @@ 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 ParserDeps struct { +type Deps struct { Enabled bool ExternalBrowser string GoroutinesNumber int } -func NewParser(deps ParserDeps) *Parser { +func NewParser(deps Deps) *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 3147fa1..1e47a3c 100644 --- a/internal/parsers/mandarake/service.go +++ b/internal/parsers/mandarake/service.go @@ -41,14 +41,7 @@ func (s *Parser) getPrice(ctx context.Context, task shared.Task) (int32, error) return minimal, nil } -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() - +func (s *Parser) getMinimalPrice(sessionCtx context.Context, task shared.Task) int32 { var ( singlePrice string rangedPrice string diff --git a/internal/parsers/mandarake/service_test.go b/internal/parsers/mandarake/service_test.go index c0f5a79..736331a 100644 --- a/internal/parsers/mandarake/service_test.go +++ b/internal/parsers/mandarake/service_test.go @@ -65,7 +65,6 @@ 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 9dc3c5f..1fecccf 100644 --- a/internal/parsers/surugaya.go +++ b/internal/parsers/surugaya.go @@ -11,21 +11,19 @@ import ( type SurugayaParser struct { scrapper sc.SurugayaScrapperClient - ctx context.Context } -func NewSurugayaParser(ctx context.Context, scrapper sc.SurugayaScrapperClient) *SurugayaParser { +func NewSurugayaParser(scrapper sc.SurugayaScrapperClient) *SurugayaParser { log.Debug("Surugaya parser init") return &SurugayaParser{ scrapper: scrapper, - ctx: ctx, } } -func (s *SurugayaParser) HandleTasks(tasks []shared.Task, sender chan shared.TaskResult, state *appState.State) { +func (s *SurugayaParser) HandleTasks(ctx context.Context, tasks []shared.Task, sender chan shared.TaskResult, state *appState.State) { log.WithField("count", len(tasks)).Debug("Handling Surugaya Tasks") - stream, err := s.scrapper.ProcessTasks(s.ctx) + stream, err := s.scrapper.ProcessTasks(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 4748566..61e7fe0 100644 --- a/internal/processor/service.go +++ b/internal/processor/service.go @@ -1,13 +1,14 @@ package processor import ( + "context" log "github.com/sirupsen/logrus" "sync" "task-processor/internal/appState" "task-processor/internal/shared" ) -func (p *Processor) StartWork(receivedTasks []shared.TaskResponse) { +func (p *Processor) StartWork(ctx context.Context, receivedTasks []shared.TaskResponse) { log.Info("Starting work...") p.state.ResetCounters() @@ -27,7 +28,7 @@ func (p *Processor) StartWork(receivedTasks []shared.TaskResponse) { go func(origin string, tasks []shared.Task) { defer wg.Done() log.Info("Running task handler for origin: ", origin) - p.handlers[origin].HandleTasks(tasks, p.out, p.state) + p.handlers[origin].HandleTasks(ctx, tasks, p.out, p.state) }(origin, tasks) } wg.Wait() From 13ebb273359f09639e8d73fab8f006fe7dd082b5 Mon Sep 17 00:00:00 2001 From: nquidox Date: Sat, 28 Feb 2026 10:53:33 +0300 Subject: [PATCH 02/10] small refactor --- internal/app/app.go | 85 +++++++++++++++++++++++------------------- internal/app/server.go | 4 +- 2 files changed, 48 insertions(+), 41 deletions(-) diff --git a/internal/app/app.go b/internal/app/app.go index 6353e7c..832e338 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -3,11 +3,9 @@ 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" @@ -28,6 +26,8 @@ type App struct { state *appState.State network *remote.Network numCPUs int + metricsSrv *router.Handler + taskApiSrv *grpc.Server } func New(c *config.Config) *App { @@ -38,6 +38,14 @@ 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), @@ -47,33 +55,31 @@ func New(c *config.Config) *App { state: st, network: remote.NewHandler(), numCPUs: numCPUs, + metricsSrv: mSrv, + taskApiSrv: server, } } -func (app *App) Run() { - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - +func (app *App) Run(ctx context.Context) { log.Info("Application start") + + addr := net.JoinHostPort(app.config.GrpcCfg.ServerHost, app.config.GrpcCfg.ServerPort) + log.WithFields(log.Fields{ - "Service address": app.config.GrpcCfg.ServerHost + ":" + app.config.GrpcCfg.ServerPort, + "Service address": addr, "Number of CPUs": app.numCPUs, }).Debug("App settings") - //metrics - mSrv := router.NewHandler(router.Deps{ - Addr: net.JoinHostPort(app.config.Metrics.Host, app.config.Metrics.Port), - GinMode: app.config.Metrics.GinMode, - }) + errChan := make(chan error, 16) //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) @@ -82,11 +88,11 @@ func (app *App) Run() { handlers := make(map[string]parsers.TaskHandler) if app.config.OriginEnabled.Surugaya { - handlers[shared.OriginSurugaya] = parsers.NewSurugayaParser(ctx, surugayaScrapper) + handlers[shared.OriginSurugaya] = parsers.NewSurugayaParser(surugayaScrapper) } if app.config.OriginEnabled.Mandarake { - handlers[shared.OriginMandarake] = mandarake.NewParser(mandarake.ParserDeps{ + handlers[shared.OriginMandarake] = mandarake.NewParser(mandarake.Deps{ Enabled: app.config.OriginEnabled.Mandarake, ExternalBrowser: app.config.ExternalBrowser, GoroutinesNumber: app.numCPUs, @@ -108,7 +114,7 @@ func (app *App) Run() { receivedTasks := app.network.RequestTasks(ctx, apiClient) log.WithField("length", len(receivedTasks)).Debug("End receiving") - taskProcessor.StartWork(receivedTasks) + taskProcessor.StartWork(ctx, receivedTasks) } go func() { @@ -143,37 +149,38 @@ func (app *App) Run() { //start metrics server go func() { - if err := mSrv.Run(); err != nil { - log.WithError(err).Error("Metrics server run failed") + if err := app.metricsSrv.Run(); err != nil { + errChan <- err } }() //gRPC Server for status response go func() { - listener, err := net.Listen("tcp", app.config.GrpcCfg.ServerHost+":"+app.config.GrpcCfg.ServerPort) + listener, err := net.Listen("tcp", addr) if err != nil { - log.Fatalf("failed to listen: %v", err) + 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) + log.Infof("gRPC Server listening at %v", addr) + if err = app.taskApiSrv.Serve(listener); err != nil { + errChan <- err } }() - 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() + 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") + } } diff --git a/internal/app/server.go b/internal/app/server.go index f837430..9c9dadc 100644 --- a/internal/app/server.go +++ b/internal/app/server.go @@ -12,10 +12,10 @@ type Server struct { state *appState.State } -func newServer(app *App) *grpc.Server { +func newServer(state *appState.State) *grpc.Server { s := grpc.NewServer() srv := &Server{ - state: app.state, + state: state, } pb.RegisterTaskProcessorServer(s, srv) return s From 56309cafb914254b42e741b10bd588e8f2d6f9ba Mon Sep 17 00:00:00 2001 From: nquidox Date: Sat, 28 Feb 2026 23:32:47 +0300 Subject: [PATCH 03/10] env added --- config.env | 1 + config/config.go | 2 ++ 2 files changed, 3 insertions(+) diff --git a/config.env b/config.env index 7633de5..b22e762 100644 --- a/config.env +++ b/config.env @@ -14,6 +14,7 @@ GRPC_SURUGAYA_SCRAPPER_PORT=9070 TASK_RETRY_COUNT=3 TASK_RETRY_MINUTES=5 +TASK_TIMEOUT_MINUTES=5 ORIGIN_SURUGAYA_ENABLED=false ORIGIN_MANDARAKE_ENABLED=false diff --git a/config/config.go b/config/config.go index b19ffe0..f2ebaa4 100644 --- a/config/config.go +++ b/config/config.go @@ -29,6 +29,7 @@ type GrpcConfig struct { type TasksConfig struct { RetryCount int RetryMinutes int + TaskTimeout int } type OriginEnabled struct { @@ -62,6 +63,7 @@ func NewConfig() *Config { TasksCfg: TasksConfig{ RetryCount: getEnvInt("TASK_RETRY_COUNT", 3), RetryMinutes: getEnvInt("TASK_RETRY_MINUTES", 5), + TaskTimeout: getEnvInt("TASK_TIMEOUT_MINUTES", 5), }, OriginEnabled: OriginEnabled{ From f4665665056ba1f984998735a5abb4253a10caa2 Mon Sep 17 00:00:00 2001 From: nquidox Date: Sat, 28 Feb 2026 23:33:22 +0300 Subject: [PATCH 04/10] page timeout added --- internal/app/app.go | 7 ++++--- internal/parsers/mandarake/handleTasks.go | 13 ++++++++----- internal/parsers/mandarake/handler.go | 6 +++++- 3 files changed, 17 insertions(+), 9 deletions(-) diff --git a/internal/app/app.go b/internal/app/app.go index 832e338..c7359e1 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -75,9 +75,6 @@ func (app *App) Run(ctx context.Context) { //main 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) @@ -96,6 +93,7 @@ func (app *App) Run(ctx context.Context) { Enabled: app.config.OriginEnabled.Mandarake, ExternalBrowser: app.config.ExternalBrowser, GoroutinesNumber: app.numCPUs, + TaskTimeout: app.config.TasksCfg.TaskTimeout, }) } @@ -117,6 +115,9 @@ func (app *App) Run(ctx context.Context) { taskProcessor.StartWork(ctx, receivedTasks) } + period := time.NewTicker(app.checkPeriod * time.Hour) + defer period.Stop() + go func() { process() //immediate start for range period.C { diff --git a/internal/parsers/mandarake/handleTasks.go b/internal/parsers/mandarake/handleTasks.go index 1b566e7..1ddd10a 100644 --- a/internal/parsers/mandarake/handleTasks.go +++ b/internal/parsers/mandarake/handleTasks.go @@ -16,9 +16,6 @@ func (s *Parser) HandleTasks(ctx context.Context, tasks []shared.Task, sender ch 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() - receiver := make(chan shared.Task, len(tasks)) for _, task := range tasks { receiver <- task @@ -31,7 +28,7 @@ func (s *Parser) HandleTasks(ctx context.Context, tasks []shared.Task, sender ch wg.Add(1) go func() { defer wg.Done() - s.worker(sessionCtx, receiver, sender) + s.worker(allocCtx, receiver, sender) }() } wg.Wait() @@ -42,14 +39,20 @@ func (s *Parser) HandleTasks(ctx context.Context, tasks []shared.Task, sender ch func (s *Parser) worker(ctx context.Context, receiver chan shared.Task, sender chan shared.TaskResult) { for task := range receiver { + taskCtx, taskCancel := chromedp.NewContext(ctx /* chromedp.WithLogf(log.Printf) */, chromedp.WithLogf(func(string, ...any) {})) + timeoutCtx, timeoutCancel := context.WithTimeout(taskCtx, s.taskTimeout) + log.WithField("task_uuid", task.MerchUuid).Infof("%v %v processing task", logHeader, logWorker) //price will be zeroPrice value in case of any error or if price not found - price := s.getMinimalPrice(ctx, task) + price := s.getMinimalPrice(timeoutCtx, task) sender <- shared.TaskResult{ MerchUuid: task.MerchUuid, Origin: task.Origin, Price: price, } + + timeoutCancel() + taskCancel() } } diff --git a/internal/parsers/mandarake/handler.go b/internal/parsers/mandarake/handler.go index cd6b272..0955206 100644 --- a/internal/parsers/mandarake/handler.go +++ b/internal/parsers/mandarake/handler.go @@ -2,6 +2,7 @@ package mandarake import ( log "github.com/sirupsen/logrus" + "time" ) const ( @@ -16,22 +17,25 @@ const ( type Parser struct { externalBrowser string goroutinesNumber int + taskTimeout time.Duration } type Deps struct { Enabled bool ExternalBrowser string GoroutinesNumber int + TaskTimeout int } func NewParser(deps Deps) *Parser { if !deps.Enabled { - log.Info(logHeader + "disabled") + log.Infof("%v disabled", logHeader) return nil } return &Parser{ externalBrowser: deps.ExternalBrowser, goroutinesNumber: deps.GoroutinesNumber, + taskTimeout: time.Minute * time.Duration(deps.TaskTimeout), } } From 8395cf71b4a7d751987569c1e78c1eee97b1561c Mon Sep 17 00:00:00 2001 From: nquidox Date: Sat, 28 Feb 2026 23:33:37 +0300 Subject: [PATCH 05/10] pprof --- cmd/main.go | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/cmd/main.go b/cmd/main.go index 92ae164..409b141 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -2,6 +2,9 @@ package main import ( "context" + log "github.com/sirupsen/logrus" + "net/http" + _ "net/http/pprof" "os" "os/signal" "syscall" @@ -18,6 +21,12 @@ func main() { logging.LogSetup(c.LogLevel) + if c.Metrics.GinMode != "release" { + go func() { + log.Println(http.ListenAndServe("localhost:6060", nil)) + }() + } + appl := app.New(c) appl.Run(ctx) From d90682e183e45e9754127dac9fa470d38948ed3a Mon Sep 17 00:00:00 2001 From: nquidox Date: Sun, 1 Mar 2026 00:05:55 +0300 Subject: [PATCH 06/10] pprof env --- cmd/main.go | 2 +- config.env | 1 + config/config.go | 2 ++ 3 files changed, 4 insertions(+), 1 deletion(-) diff --git a/cmd/main.go b/cmd/main.go index 409b141..e244516 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -21,7 +21,7 @@ func main() { logging.LogSetup(c.LogLevel) - if c.Metrics.GinMode != "release" { + if c.PprofEnabled { go func() { log.Println(http.ListenAndServe("localhost:6060", nil)) }() diff --git a/config.env b/config.env index b22e762..757bc25 100644 --- a/config.env +++ b/config.env @@ -2,6 +2,7 @@ APP_LOG_LEVEL=error APP_NUMCPUS=-1 APP_CHECK_PERIOD=6 EXTERNAL_BROWSER= +PPROF_ENABLED=false GRPC_SERVER_HOST=0.0.0.0 GRPC_SERVER_PORT=9060 diff --git a/config/config.go b/config/config.go index f2ebaa4..6b1ec59 100644 --- a/config/config.go +++ b/config/config.go @@ -7,6 +7,7 @@ import ( ) type Config struct { + PprofEnabled bool LogLevel string NumCPUs int CheckPeriod int @@ -46,6 +47,7 @@ type MetricsConfig struct { func NewConfig() *Config { return &Config{ + PprofEnabled: getEnvBool("PPROF_ENABLED", true), LogLevel: getEnv("APP_LOG_LEVEL", "debug"), NumCPUs: getEnvInt("APP_NUMCPUS", -1), CheckPeriod: getEnvInt("APP_CHECK_PERIOD", 6), From 77c626a7c3027153758f3508a5ca81bcd04d0d3c Mon Sep 17 00:00:00 2001 From: nquidox Date: Sun, 1 Mar 2026 00:06:19 +0300 Subject: [PATCH 07/10] exclude metrics req from log --- pkg/router/handler.go | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/pkg/router/handler.go b/pkg/router/handler.go index ddd1b91..5ede9de 100644 --- a/pkg/router/handler.go +++ b/pkg/router/handler.go @@ -37,6 +37,12 @@ func NewHandler(deps Deps) *Handler { p := ginprometheus.NewPrometheus("gin") p.Use(engine) + engine.Use(gin.LoggerWithConfig(gin.LoggerConfig{ + Skip: func(c *gin.Context) bool { + return c.Request.URL.Path == "/metrics" + }, + })) + srv := http.Server{ Addr: deps.Addr, Handler: engine, From a32b1ef69ba2a67b5f0d0b0448b7594e68d1d1d7 Mon Sep 17 00:00:00 2001 From: nquidox Date: Sun, 1 Mar 2026 01:01:06 +0300 Subject: [PATCH 08/10] CloseAndRecv in send --- internal/app/app.go | 2 +- internal/remote/interface.go | 2 +- internal/remote/send.go | 17 +++++++++++++---- 3 files changed, 15 insertions(+), 6 deletions(-) diff --git a/internal/app/app.go b/internal/app/app.go index c7359e1..4f27a7d 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -141,7 +141,7 @@ func (app *App) Run(ctx context.Context) { l := len(sendData) if l > 0 { log.WithField("length", l).Debug("Sending parsed data") - app.network.SendResult(apiClient, sendData) + app.network.SendResult(ctx, apiClient, sendData) sendData = sendData[:0] } } diff --git a/internal/remote/interface.go b/internal/remote/interface.go index b34de33..210d214 100644 --- a/internal/remote/interface.go +++ b/internal/remote/interface.go @@ -8,5 +8,5 @@ import ( type Handler interface { RequestTasks(ctx context.Context, client pb.TaskProcessorClient) []shared.TaskResponse - SendResult(client pb.TaskProcessorClient, tasksDone []shared.TaskResult) + SendResult(ctx context.Context, client pb.TaskProcessorClient, tasksDone []shared.TaskResult) } diff --git a/internal/remote/send.go b/internal/remote/send.go index 6e9c148..046fadf 100644 --- a/internal/remote/send.go +++ b/internal/remote/send.go @@ -5,10 +5,14 @@ import ( log "github.com/sirupsen/logrus" "task-processor/internal/shared" pb "task-processor/proto/taskProcessor" + "time" ) -func (n *Network) SendResult(client pb.TaskProcessorClient, tasksDone []shared.TaskResult) { - stream, err := client.SendResult(context.Background()) +func (n *Network) SendResult(ctx context.Context, client pb.TaskProcessorClient, tasksDone []shared.TaskResult) { + sendCtx, cancel := context.WithTimeout(ctx, time.Second*60) + defer cancel() + + stream, err := client.SendResult(sendCtx) if err != nil { log.Fatalf("Error calling PostMerch: %v", err) } @@ -29,7 +33,12 @@ func (n *Network) SendResult(client pb.TaskProcessorClient, tasksDone []shared.T } } - if err = stream.CloseSend(); err != nil { - log.Fatalf("Error closing stream: %v", err) + //if err = stream.CloseSend(); err != nil { + // log.Fatalf("Error closing stream: %v", err) + //} + + _, err = stream.CloseAndRecv() + if err != nil { + log.Fatalf("Error receiving response: %v", err) } } From 532386222bcc3dce1da70c53e99911223babb049 Mon Sep 17 00:00:00 2001 From: nquidox Date: Wed, 18 Mar 2026 20:34:22 +0300 Subject: [PATCH 09/10] setup browser --- internal/parsers/mandarake/handleTasks.go | 33 ++++++++++++++++++++++- 1 file changed, 32 insertions(+), 1 deletion(-) diff --git a/internal/parsers/mandarake/handleTasks.go b/internal/parsers/mandarake/handleTasks.go index 1ddd10a..8596747 100644 --- a/internal/parsers/mandarake/handleTasks.go +++ b/internal/parsers/mandarake/handleTasks.go @@ -10,8 +10,39 @@ import ( "task-processor/internal/shared" ) +func (s *Parser) setupBrowser(ctx context.Context) (string, error) { + allocCtx, allocCancel := chromedp.NewRemoteAllocator(ctx, s.externalBrowser) + defer allocCancel() + + pageCtx, pageCancel := chromedp.NewContext(allocCtx, chromedp.WithLogf(func(string, ...any) {})) + defer pageCancel() + + copyright := "No copyright div found." + + if err := chromedp.Run(pageCtx, + chromedp.Navigate("https://www.mandarake.co.jp/"), + chromedp.WaitReady("body", chromedp.ByQuery), + chromedp.Text(`div.copyright`, ©right, chromedp.ByQuery, chromedp.AtLeast(0)), + chromedp.Navigate("https://www.mandarake.co.jp/index2.html"), + chromedp.WaitReady("body", chromedp.ByQuery), + ); err != nil { + log.WithError(err).Error(logHeader + logGetPrice + "failed to get single price tag") + return copyright, err + } + + return copyright, nil +} + 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) + log.Infof("%v Start handling tasks", logHeader) + log.Infof("%v Setting up browser", logHeader) + cr, err := s.setupBrowser(ctx) + if err != nil { + log.WithError(err).Error(logHeader + logGetPrice + "failed to setup browser") + } + log.WithField("Copyright message", cr).Infof("%v Finished setting up browser.", logHeader) + + log.Infof("%v %v processing tasks...", logHeader, logWorker) allocCtx, allocCancel := chromedp.NewRemoteAllocator(ctx, s.externalBrowser) defer allocCancel() From 1e331cada590fea93f47e1fb1d13852478a902ac Mon Sep 17 00:00:00 2001 From: nquidox Date: Wed, 18 Mar 2026 20:34:33 +0300 Subject: [PATCH 10/10] gin logger --- pkg/router/handler.go | 11 +++-------- 1 file changed, 3 insertions(+), 8 deletions(-) diff --git a/pkg/router/handler.go b/pkg/router/handler.go index 5ede9de..0f924cc 100644 --- a/pkg/router/handler.go +++ b/pkg/router/handler.go @@ -21,7 +21,7 @@ type Deps struct { const pkgLogHeader string = "Router |" func NewHandler(deps Deps) *Handler { - engine := gin.Default() + engine := gin.New() if deps.GinMode == "release" { gin.SetMode(gin.ReleaseMode) @@ -32,17 +32,12 @@ func NewHandler(deps Deps) *Handler { } } - engine.GET("/", func(c *gin.Context) { c.JSON(200, gin.H{"msg": "v2"}) }) + logGroup := engine.Group("") + logGroup.GET("/", func(c *gin.Context) { c.JSON(200, gin.H{"msg": "v2"}) }) p := ginprometheus.NewPrometheus("gin") p.Use(engine) - engine.Use(gin.LoggerWithConfig(gin.LoggerConfig{ - Skip: func(c *gin.Context) bool { - return c.Request.URL.Path == "/metrics" - }, - })) - srv := http.Server{ Addr: deps.Addr, Handler: engine,