chore: refactoring

common/interfaces.go

@@ -14,6 +14,10 @@ import (
 type Pool interface {
 	// Workers returns worker list associated with the pool.
 	Workers() (workers []*worker.Process)
+	// RemoveWorker removes worker from the pool.
+	RemoveWorker(ctx context.Context) error
+	// AddWorker adds worker to the pool.
+	AddWorker() error
 	// Exec payload
 	Exec(ctx context.Context, p *payload.Payload, stopCh chan struct{}) (chan *staticPool.PExec, error)
 	// Reset kill all workers inside the watcher and replaces with new

go.mod

@@ -14,7 +14,7 @@ require (
 	github.com/roadrunner-server/endure/v2 v2.4.2
 	github.com/roadrunner-server/errors v1.3.0
 	github.com/roadrunner-server/goridge/v3 v3.8.1
-	github.com/roadrunner-server/sdk/v4 v4.4.0
+	github.com/roadrunner-server/sdk/v4 v4.5.0-beta.1
 	github.com/stretchr/testify v1.8.4
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.44.0
 	go.opentelemetry.io/contrib/propagators/jaeger v1.19.0
@@ -42,7 +42,7 @@ require (
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/prometheus/client_model v0.4.0 // indirect
 	github.com/prometheus/common v0.44.0 // indirect
-	github.com/prometheus/procfs v0.11.1 // indirect
+	github.com/prometheus/procfs v0.12.0 // indirect
 	github.com/roadrunner-server/tcplisten v1.4.0 // indirect
 	github.com/shirou/gopsutil v3.21.11+incompatible // indirect
 	github.com/tklauser/go-sysconf v0.3.12 // indirect

go.sum

@@ -1,5 +1,3 @@
-github.com/benbjohnson/clock v1.3.0 h1:ip6w0uFQkncKQ979AypyG0ER7mqUSBdKLOgAle/AT8A=
-github.com/benbjohnson/clock v1.3.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA=
 github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
 github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
 github.com/caddyserver/certmagic v0.19.2 h1:HZd1AKLx4592MalEGQS39DKs2ZOAJCEM/xYPMQ2/ui0=
@@ -52,8 +50,8 @@ github.com/prometheus/client_model v0.4.0 h1:5lQXD3cAg1OXBf4Wq03gTrXHeaV0TQvGfUo
 github.com/prometheus/client_model v0.4.0/go.mod h1:oMQmHW1/JoDwqLtg57MGgP/Fb1CJEYF2imWWhWtMkYU=
 github.com/prometheus/common v0.44.0 h1:+5BrQJwiBB9xsMygAB3TNvpQKOwlkc25LbISbrdOOfY=
 github.com/prometheus/common v0.44.0/go.mod h1:ofAIvZbQ1e/nugmZGz4/qCb9Ap1VoSTIO7x0VV9VvuY=
-github.com/prometheus/procfs v0.11.1 h1:xRC8Iq1yyca5ypa9n1EZnWZkt7dwcoRPQwX/5gwaUuI=
-github.com/prometheus/procfs v0.11.1/go.mod h1:eesXgaPo1q7lBpVMoMy0ZOFTth9hBn4W/y0/p/ScXhY=
+github.com/prometheus/procfs v0.12.0 h1:jluTpSng7V9hY0O2R9DzzJHYb2xULk9VTR1V1R/k6Bo=
+github.com/prometheus/procfs v0.12.0/go.mod h1:pcuDEFsWDnvcgNzo4EEweacyhjeA9Zk3cnaOZAZEfOo=
 github.com/roadrunner-server/api/v4 v4.8.0 h1:wsANwein0dD3q9OQi8L6XGjgEkEn/23mAJ4ZphavfLM=
 github.com/roadrunner-server/api/v4 v4.8.0/go.mod h1:FhCdSyHWBJfyZ0yuggVE72WYKcmAOsGuJbC3dvzaMOg=
 github.com/roadrunner-server/endure/v2 v2.4.2 h1:aFnPc321l5HDzE2mN5wwfksJ40lgXwfU3RSqdS1LyUQ=
@@ -62,8 +60,8 @@ github.com/roadrunner-server/errors v1.3.0 h1:kLVXpXne0jMReN7pj8KIhyYyjqKjsPC5DR
 github.com/roadrunner-server/errors v1.3.0/go.mod h1:XYVuhXvxi3yQaP/zCLB6QRZ0JvQIRaBa0SKFHL4WLKg=
 github.com/roadrunner-server/goridge/v3 v3.8.1 h1:mdS5lDKQwPuVJ2jwW7l5cngJNJiie7xEGwpgw7a6CuQ=
 github.com/roadrunner-server/goridge/v3 v3.8.1/go.mod h1:L5UkNzD8aKLz6TzpqmmiHOJ6EnsadsWEYNoqK/4qoK0=
-github.com/roadrunner-server/sdk/v4 v4.4.0 h1:WPrx5+5k43SGAsS/9EKYReU17jnONY9eRmW1pQbu/+w=
-github.com/roadrunner-server/sdk/v4 v4.4.0/go.mod h1:9TPfWYRm4jtlA//QAUBWAKC2c7Mt33FRAA2c9Be9yc8=
+github.com/roadrunner-server/sdk/v4 v4.5.0-beta.1 h1:4YGYsf2GmP4d7eQRaHWakJ37I02CgtOG5A2TkL9dU3c=
+github.com/roadrunner-server/sdk/v4 v4.5.0-beta.1/go.mod h1:BQFbO2nGRFnwP5JMN4t/wQcTH7asOZ32L+3zpzipJeY=
 github.com/roadrunner-server/tcplisten v1.4.0 h1:yWo09zktv/CSV6VywLfw4pwNcUchgTiIrW4uIICtO5M=
 github.com/roadrunner-server/tcplisten v1.4.0/go.mod h1:A6+VSnW2ETGnN/e/CMdP63ZXqQDaC0UDMU6QmyuB0yM=
 github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ=
@@ -98,8 +96,6 @@ go.uber.org/goleak v1.2.0 h1:xqgm/S+aQvhWFTtR0XK3Jvg7z8kGV8P4X14IzwN3Eqk=
 go.uber.org/goleak v1.2.0/go.mod h1:XJYK+MuIchqpmGmUSAzotztawfKvYLUIgg7guXrwVUo=
 go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0=
 go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y=
-go.uber.org/zap v1.25.0 h1:4Hvk6GtkucQ790dqmj7l1eEnRdKm3k3ZUrUMS2d5+5c=
-go.uber.org/zap v1.25.0/go.mod h1:JIAUzQIH94IC4fOJQm7gMmBJP5k7wQfdcnYdPoEXJYk=
 go.uber.org/zap v1.26.0 h1:sI7k6L95XOKS281NhVKOFCUNIvv9e0w4BF8N3u+tCRo=
 go.uber.org/zap v1.26.0/go.mod h1:dtElttAiwGvoJ/vj4IwHBS/gXsEu/pZ50mUIRWuG0so=
 golang.org/x/crypto v0.13.0 h1:mvySKfSWJ+UKUii46M40LOvyWfN0s2U+46/jDd0e6Ck=

handler/request.go

@@ -206,7 +206,7 @@ func (r *Request) contentType() int {
 		return contentNone
 	}
 
-	ct := r.Header.Get("content-type")
+	ct := r.Header.Get("Content-Type")
 	if strings.Contains(ct, "application/x-www-form-urlencoded") {
 		return contentURLEncoded
 	}

init.go

@@ -0,0 +1,75 @@
+package http
+
+import (
+	"github.com/roadrunner-server/http/v4/common"
+	bundledMw "github.com/roadrunner-server/http/v4/middleware"
+	"github.com/roadrunner-server/http/v4/servers/fcgi"
+	httpServer "github.com/roadrunner-server/http/v4/servers/http"
+	httpsServer "github.com/roadrunner-server/http/v4/servers/https"
+)
+
+// ------- PRIVATE ---------
+
+func (p *Plugin) initServers() error {
+	if p.cfg.EnableHTTP() {
+		p.servers = append(p.servers, httpServer.NewHTTPServer(p, p.cfg, p.stdLog, p.log))
+	}
+
+	if p.cfg.EnableTLS() {
+		https, err := httpsServer.NewHTTPSServer(p, p.cfg.SSLConfig, p.cfg.HTTP2Config, p.stdLog, p.log)
+		if err != nil {
+			return err
+		}
+
+		p.servers = append(p.servers, https)
+	}
+
+	if p.cfg.EnableFCGI() {
+		p.servers = append(p.servers, fcgi.NewFCGIServer(p, p.cfg.FCGIConfig, p.log, p.stdLog))
+	}
+
+	return nil
+}
+
+func (p *Plugin) applyBundledMiddleware() {
+	// apply max_req_size and logger middleware
+	for i := 0; i < len(p.servers); i++ {
+		serv := p.servers[i].Server()
+		serv.Handler = bundledMw.MaxRequestSize(serv.Handler, p.cfg.MaxRequestSize*MB)
+		serv.Handler = bundledMw.NewLogMiddleware(serv.Handler, p.cfg.AccessLogs, p.log)
+	}
+}
+
+func (p *Plugin) unmarshal(cfg common.Configurer) error {
+	// unmarshal general section
+	err := cfg.UnmarshalKey(PluginName, &p.cfg)
+	if err != nil {
+		return err
+	}
+
+	// unmarshal HTTPS section
+	err = cfg.UnmarshalKey(sectionHTTPS, &p.cfg.SSLConfig)
+	if err != nil {
+		return err
+	}
+
+	// unmarshal H2C section
+	err = cfg.UnmarshalKey(sectionHTTP2, &p.cfg.HTTP2Config)
+	if err != nil {
+		return err
+	}
+
+	// unmarshal uploads section
+	err = cfg.UnmarshalKey(sectionUploads, &p.cfg.Uploads)
+	if err != nil {
+		return err
+	}
+
+	// unmarshal fcgi section
+	err = cfg.UnmarshalKey(sectionFCGI, &p.cfg.FCGIConfig)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}

plugin.go

@@ -9,18 +9,12 @@ import (
 	"github.com/roadrunner-server/endure/v2/dep"
 	"github.com/roadrunner-server/http/v4/common"
 
-	"github.com/roadrunner-server/http/v4/servers/fcgi"
-	httpServer "github.com/roadrunner-server/http/v4/servers/http"
-	httpsServer "github.com/roadrunner-server/http/v4/servers/https"
-
 	"github.com/roadrunner-server/errors"
 	"github.com/roadrunner-server/http/v4/config"
 	"github.com/roadrunner-server/http/v4/handler"
-	bundledMw "github.com/roadrunner-server/http/v4/middleware"
 	"github.com/roadrunner-server/sdk/v4/metrics"
 	"github.com/roadrunner-server/sdk/v4/state/process"
 	"github.com/roadrunner-server/sdk/v4/utils"
-	"github.com/roadrunner-server/sdk/v4/worker"
 	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
 	jprop "go.opentelemetry.io/contrib/propagators/jaeger"
 	"go.opentelemetry.io/otel/propagation"
@@ -47,8 +41,8 @@ const (
 
 // internal interface to start-stop http servers
 type internalServer interface {
-	Start(map[string]common.Middleware, []string) error
-	GetServer() *http.Server
+	Serve(map[string]common.Middleware, []string) error
+	Server() *http.Server
 	Stop()
 }
 
@@ -89,32 +83,7 @@ func (p *Plugin) Init(cfg common.Configurer, rrLogger common.Logger, srv common.
 		return errors.E(op, errors.Disabled)
 	}
 
-	// unmarshal general section
-	err := cfg.UnmarshalKey(PluginName, &p.cfg)
-	if err != nil {
-		return errors.E(op, err)
-	}
-
-	// unmarshal HTTPS section
-	err = cfg.UnmarshalKey(sectionHTTPS, &p.cfg.SSLConfig)
-	if err != nil {
-		return errors.E(op, err)
-	}
-
-	// unmarshal H2C section
-	err = cfg.UnmarshalKey(sectionHTTP2, &p.cfg.HTTP2Config)
-	if err != nil {
-		return errors.E(op, err)
-	}
-
-	// unmarshal uploads section
-	err = cfg.UnmarshalKey(sectionUploads, &p.cfg.Uploads)
-	if err != nil {
-		return errors.E(op, err)
-	}
-
-	// unmarshal fcgi section
-	err = cfg.UnmarshalKey(sectionFCGI, &p.cfg.FCGIConfig)
+	err := p.unmarshal(cfg)
 	if err != nil {
 		return errors.E(op, err)
 	}
@@ -151,19 +120,17 @@ func (p *Plugin) Init(cfg common.Configurer, rrLogger common.Logger, srv common.
 // Serve serves the svc.
 func (p *Plugin) Serve() chan error {
 	errCh := make(chan error, 2)
+
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
 	var err error
 	p.pool, err = p.server.NewPool(context.Background(), p.cfg.Pool, map[string]string{RrMode: RrModeHTTP}, p.log)
 	if err != nil {
 		errCh <- err
 		return errCh
 	}
 
-	// just to be safe :)
-	if p.pool == nil {
-		errCh <- errors.Str("pool should be initialized")
-		return errCh
-	}
-
 	p.handler, err = handler.NewHandler(
 		p.cfg,
 		p.pool,
@@ -187,7 +154,7 @@ func (p *Plugin) Serve() chan error {
 	// start all servers
 	for i := 0; i < len(p.servers); i++ {
 		go func(idx int) {
-			errSt := p.servers[idx].Start(p.mdwr, p.cfg.Middleware)
+			errSt := p.servers[idx].Serve(p.mdwr, p.cfg.Middleware)
 			if errSt != nil {
 				errCh <- errSt
 				return
@@ -249,11 +216,12 @@ func (p *Plugin) Workers() []*process.State {
 	p.mu.RLock()
 	defer p.mu.RUnlock()
 
-	workers := p.workers()
-	if workers == nil {
+	if p.pool == nil {
 		return nil
 	}
 
+	workers := p.pool.Workers()
+
 	ps := make([]*process.State, 0, len(workers))
 	for i := 0; i < len(workers); i++ {
 		state, err := process.WorkerProcessState(workers[i])
@@ -266,14 +234,6 @@ func (p *Plugin) Workers() []*process.State {
 	return ps
 }
 
-// internal
-func (p *Plugin) workers() []*worker.Process {
-	if p == nil || p.pool == nil {
-		return nil
-	}
-	return p.pool.Workers()
-}
-
 // Name returns endure.Named interface implementation
 func (p *Plugin) Name() string {
 	return PluginName
@@ -314,35 +274,3 @@ func (p *Plugin) Collects() []*dep.In {
 		}, (*common.Middleware)(nil)),
 	}
 }
-
-// ------- PRIVATE ---------
-
-func (p *Plugin) initServers() error {
-	if p.cfg.EnableHTTP() {
-		p.servers = append(p.servers, httpServer.NewHTTPServer(p, p.cfg, p.stdLog, p.log))
-	}
-
-	if p.cfg.EnableTLS() {
-		https, err := httpsServer.NewHTTPSServer(p, p.cfg.SSLConfig, p.cfg.HTTP2Config, p.stdLog, p.log)
-		if err != nil {
-			return err
-		}
-
-		p.servers = append(p.servers, https)
-	}
-
-	if p.cfg.EnableFCGI() {
-		p.servers = append(p.servers, fcgi.NewFCGIServer(p, p.cfg.FCGIConfig, p.log, p.stdLog))
-	}
-
-	return nil
-}
-
-func (p *Plugin) applyBundledMiddleware() {
-	// apply max_req_size and logger middleware
-	for i := 0; i < len(p.servers); i++ {
-		serv := p.servers[i].GetServer()
-		serv.Handler = bundledMw.MaxRequestSize(serv.Handler, p.cfg.MaxRequestSize*MB)
-		serv.Handler = bundledMw.NewLogMiddleware(serv.Handler, p.cfg.AccessLogs, p.log)
-	}
-}

servers/fcgi/fcgi.go

@@ -33,7 +33,7 @@ func NewFCGIServer(handler http.Handler, cfg *FCGI, log *zap.Logger, errLog *log
 	}
 }
 
-func (s *Server) Start(mdwr map[string]common.Middleware, order []string) error {
+func (s *Server) Serve(mdwr map[string]common.Middleware, order []string) error {
 	const op = errors.Op("serve_fcgi")
 
 	if len(mdwr) > 0 {
@@ -53,7 +53,7 @@ func (s *Server) Start(mdwr map[string]common.Middleware, order []string) error
 	return nil
 }
 
-func (s *Server) GetServer() *http.Server {
+func (s *Server) Server() *http.Server {
 	return s.fcgi
 }
 

servers/http/http.go

@@ -66,8 +66,8 @@ func NewHTTPServer(handler http.Handler, cfg *config.Config, errLog *log.Logger,
 	}
 }
 
-// Start is a blocking function
-func (s *Server) Start(mdwr map[string]common.Middleware, order []string) error {
+// Serve is a blocking function
+func (s *Server) Serve(mdwr map[string]common.Middleware, order []string) error {
 	const op = errors.Op("serveHTTP")
 
 	if len(mdwr) > 0 {
@@ -93,7 +93,7 @@ func (s *Server) Start(mdwr map[string]common.Middleware, order []string) error
 	return nil
 }
 
-func (s *Server) GetServer() *http.Server {
+func (s *Server) Server() *http.Server {
 	return s.http
 }
 

servers/https/https.go

@@ -90,7 +90,7 @@ func NewHTTPSServer(handler http.Handler, cfg *SSL, cfgHTTP2 *HTTP2, errLog *log
 	}, nil
 }
 
-func (s *Server) Start(mdwr map[string]common.Middleware, order []string) error {
+func (s *Server) Serve(mdwr map[string]common.Middleware, order []string) error {
 	const op = errors.Op("serveHTTPS")
 	if len(mdwr) > 0 {
 		applyMiddleware(s.https, mdwr, order, s.log)
@@ -132,7 +132,7 @@ func (s *Server) Start(mdwr map[string]common.Middleware, order []string) error
 	return nil
 }
 
-func (s *Server) GetServer() *http.Server {
+func (s *Server) Server() *http.Server {
 	return s.https
 }
 

status.go

@@ -12,7 +12,8 @@ func (p *Plugin) Status() (*status.Status, error) {
 	p.mu.RLock()
 	defer p.mu.RUnlock()
 
-	workers := p.workers()
+	workers := p.pool.Workers()
+
 	for i := 0; i < len(workers); i++ {
 		if workers[i].State().IsActive() {
 			return &status.Status{
@@ -31,7 +32,8 @@ func (p *Plugin) Ready() (*status.Status, error) {
 	p.mu.RLock()
 	defer p.mu.RUnlock()
 
-	workers := p.workers()
+	workers := p.pool.Workers()
+
 	for i := 0; i < len(workers); i++ {
 		// If state of the worker is ready (at least 1)
 		// we assume, that plugin's worker pool is ready