Define domain specific workers in php_server and php blocks

caddy/caddy.go

@@ -4,6 +4,8 @@
 package caddy
 
 import (
+	"crypto/sha256"
+	"encoding/binary"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -30,6 +32,12 @@ const defaultDocumentRoot = "public"
 
 var iniError = errors.New("'php_ini' must be in the format: php_ini \"<key>\" \"<value>\"")
 
+// sharedState is a package-level variable to store information that can be accessed by both FrankenPHPModule and FrankenPHPApp
+var sharedState struct {
+	ModuleIDs []uint64
+	Workers   []workerConfig
+}
+
 func init() {
 	caddy.RegisterModule(FrankenPHPApp{})
 	caddy.RegisterModule(FrankenPHPModule{})
@@ -55,6 +63,8 @@ type workerConfig struct {
 	Env map[string]string `json:"env,omitempty"`
 	// Directories to watch for file changes
 	Watch []string `json:"watch,omitempty"`
+	// ModuleID identifies which module created this worker
+	ModuleID uint64 `json:"module_id,omitempty"`
 }
 
 type FrankenPHPApp struct {
@@ -108,8 +118,14 @@ func (f *FrankenPHPApp) Start() error {
 		frankenphp.WithPhpIni(f.PhpIni),
 		frankenphp.WithMaxWaitTime(f.MaxWaitTime),
 	}
+	// Add workers from FrankenPHPApp configuration
 	for _, w := range f.Workers {
-		opts = append(opts, frankenphp.WithWorkers(w.Name, repl.ReplaceKnown(w.FileName, ""), w.Num, w.Env, w.Watch))
+		opts = append(opts, frankenphp.WithWorkers(w.Name, repl.ReplaceKnown(w.FileName, ""), w.Num, w.Env, w.Watch, w.ModuleID))
+	}
+
+	// Add workers from FrankenPHPModule configurations
+	for _, w := range sharedState.Workers {
+		opts = append(opts, frankenphp.WithWorkers(w.Name, repl.ReplaceKnown(w.FileName, ""), w.Num, w.Env, w.Watch, w.ModuleID))
 	}
 
 	frankenphp.Shutdown()
@@ -130,11 +146,14 @@ func (f *FrankenPHPApp) Stop() error {
 		frankenphp.DrainWorkers()
 	}
 
-	// reset configuration so it doesn't bleed into later tests
+	// reset the configuration so it doesn't bleed into later tests
 	f.Workers = nil
 	f.NumThreads = 0
 	f.MaxWaitTime = 0
 
+	// reset moduleWorkers
+	sharedState.Workers = nil
+
 	return nil
 }
 
@@ -341,6 +360,10 @@ type FrankenPHPModule struct {
 	ResolveRootSymlink *bool `json:"resolve_root_symlink,omitempty"`
 	// Env sets an extra environment variable to the given value. Can be specified more than once for multiple environment variables.
 	Env map[string]string `json:"env,omitempty"`
+	// ModuleID is the module ID that created this request.
+	ModuleID uint64 `json:"-"`
+	// Workers configures the worker scripts to start.
+	Workers []workerConfig `json:"workers,omitempty"`
 
 	resolvedDocumentRoot        string
 	preparedEnv                 frankenphp.PreparedEnv
@@ -412,6 +435,21 @@ func (f *FrankenPHPModule) Provision(ctx caddy.Context) error {
 		}
 	}
 
+	if len(f.Workers) > 0 {
+		envString := ""
+		for k, v := range f.Env {
+			envString += k + "=" + v + ","
+		}
+		data := []byte(f.Root + envString)
+		hash := sha256.Sum256(data)
+		f.ModuleID = binary.LittleEndian.Uint64(hash[:8])
+
+		for i := range f.Workers {
+			f.Workers[i].ModuleID = f.ModuleID
+		}
+		sharedState.Workers = append(sharedState.Workers, f.Workers...)
+	}
+
 	return nil
 }
 
@@ -422,7 +460,7 @@ func needReplacement(s string) bool {
 
 // ServeHTTP implements caddyhttp.MiddlewareHandler.
 // TODO: Expose TLS versions as env vars, as Apache's mod_ssl: https://github.com/caddyserver/caddy/blob/master/modules/caddyhttp/reverseproxy/fastcgi/fastcgi.go#L298
-func (f FrankenPHPModule) ServeHTTP(w http.ResponseWriter, r *http.Request, _ caddyhttp.Handler) error {
+func (f *FrankenPHPModule) ServeHTTP(w http.ResponseWriter, r *http.Request, _ caddyhttp.Handler) error {
 	origReq := r.Context().Value(caddyhttp.OriginalRequestCtxKey).(http.Request)
 	repl := r.Context().Value(caddy.ReplacerCtxKey).(*caddy.Replacer)
 
@@ -447,6 +485,7 @@ func (f FrankenPHPModule) ServeHTTP(w http.ResponseWriter, r *http.Request, _ ca
 		frankenphp.WithRequestSplitPath(f.SplitPath),
 		frankenphp.WithRequestPreparedEnv(env),
 		frankenphp.WithOriginalRequest(&origReq),
+		frankenphp.WithModuleID(f.ModuleID),
 	)
 
 	if err != nil {
@@ -462,7 +501,7 @@ func (f FrankenPHPModule) ServeHTTP(w http.ResponseWriter, r *http.Request, _ ca
 
 // UnmarshalCaddyfile implements caddyfile.Unmarshaler.
 func (f *FrankenPHPModule) UnmarshalCaddyfile(d *caddyfile.Dispenser) error {
-	// when adding a new directive, also update the allowedDirectives error message
+	// First pass: Parse all directives except "worker"
 	for d.Next() {
 		for d.NextBlock(0) {
 			switch d.Val() {
@@ -494,29 +533,121 @@ func (f *FrankenPHPModule) UnmarshalCaddyfile(d *caddyfile.Dispenser) error {
 				if !d.NextArg() {
 					continue
 				}
-
 				v, err := strconv.ParseBool(d.Val())
 				if err != nil {
 					return err
 				}
 				if d.NextArg() {
 					return d.ArgErr()
 				}
-
 				f.ResolveRootSymlink = &v
+
+			case "worker":
+				for d.NextBlock(1) {
+				}
+				for d.NextArg() {
+				}
+				// Skip "worker" blocks in the first pass
+				continue
+
 			default:
-				allowedDirectives := "root, split, env, resolve_root_symlink"
+				allowedDirectives := "root, split, env, resolve_root_symlink, worker"
 				return wrongSubDirectiveError("php or php_server", allowedDirectives, d.Val())
 			}
 		}
 	}
 
+	// Second pass: Parse only "worker" blocks
+	d.Reset()
+	for d.Next() {
+		for d.NextBlock(0) {
+			if d.Val() == "worker" {
+				wc := workerConfig{}
+				if d.NextArg() {
+					wc.FileName = d.Val()
+				}
+
+				if d.NextArg() {
+					v, err := strconv.Atoi(d.Val())
+					if err != nil {
+						return err
+					}
+					wc.Num = v
+				}
+
+				for d.NextBlock(1) {
+					switch d.Val() {
+					case "name":
+						if !d.NextArg() {
+							return d.ArgErr()
+						}
+						wc.Name = d.Val()
+					case "file":
+						if !d.NextArg() {
+							return d.ArgErr()
+						}
+						wc.FileName = d.Val()
+					case "num":
+						if !d.NextArg() {
+							return d.ArgErr()
+						}
+						v, err := strconv.Atoi(d.Val())
+						if err != nil {
+							return err
+						}
+						wc.Num = v
+					case "env":
+						args := d.RemainingArgs()
+						if len(args) != 2 {
+							return d.ArgErr()
+						}
+						if wc.Env == nil {
+							wc.Env = make(map[string]string)
+						}
+						wc.Env[args[0]] = args[1]
+					case "watch":
+						if !d.NextArg() {
+							wc.Watch = append(wc.Watch, "./**/*.{php,yaml,yml,twig,env}")
+						} else {
+							wc.Watch = append(wc.Watch, d.Val())
+						}
+					default:
+						return fmt.Errorf("unknown worker subdirective: %s", d.Val())
+					}
+				}
+
+				if wc.FileName == "" {
+					return errors.New(`the "file" argument must be specified`)
+				}
+
+				// Inherit environment variables from the parent php_server directive
+				if !filepath.IsAbs(wc.FileName) && f.Root != "" {
+					wc.FileName = filepath.Join(f.Root, wc.FileName)
+				}
+
+				if f.Env != nil {
+					if wc.Env == nil {
+						wc.Env = make(map[string]string)
+					}
+					for k, v := range f.Env {
+						// Only set if not already defined in the worker
+						if _, exists := wc.Env[k]; !exists {
+							wc.Env[k] = v
+						}
+					}
+				}
+
+				f.Workers = append(f.Workers, wc)
+			}
+		}
+	}
+
 	return nil
 }
 
 // parseCaddyfile unmarshals tokens from h into a new Middleware.
 func parseCaddyfile(h httpcaddyfile.Helper) (caddyhttp.MiddlewareHandler, error) {
-	m := FrankenPHPModule{}
+	m := &FrankenPHPModule{}
 	err := m.UnmarshalCaddyfile(h.Dispenser)
 
 	return m, err
@@ -753,6 +884,7 @@ func parsePhpServer(h httpcaddyfile.Helper) ([]httpcaddyfile.ConfigValue, error)
 	// using the php directive syntax
 	dispenser.Next() // consume the directive name
 	err = phpsrv.UnmarshalCaddyfile(dispenser)
+
 	if err != nil {
 		return nil, err
 	}

context.go

@@ -31,6 +31,9 @@ type frankenPHPContext struct {
 
 	done      chan interface{}
 	startedAt time.Time
+
+	// The module ID that created this request
+	moduleID uint64
 }
 
 // fromContext extracts the frankenPHPContext from a context.

frankenphp.go

@@ -405,12 +405,17 @@ func ServeHTTP(responseWriter http.ResponseWriter, request *http.Request) error
 	}
 
 	// Detect if a worker is available to handle this request
-	if worker, ok := workers[fc.scriptFilename]; ok {
-		worker.handleRequest(fc)
-		return nil
+	// Look for a worker with matching moduleID or a global worker (moduleID == 0)
+	if workersList, ok := workers[fc.scriptFilename]; ok {
+		for _, worker := range workersList {
+			if worker.moduleID == 0 || worker.moduleID == fc.moduleID {
+				worker.handleRequest(fc)
+				return nil
+			}
+		}
 	}
 
-	// If no worker was availabe send the request to non-worker threads
+	// If no worker was available, send the request to non-worker threads
 	handleRequestWithRegularPHPThreads(fc)
 
 	return nil

frankenphp_test.go

@@ -65,7 +65,7 @@ func runTest(t *testing.T, test func(func(http.ResponseWriter, *http.Request), *
 
 	initOpts := []frankenphp.Option{frankenphp.WithLogger(opts.logger)}
 	if opts.workerScript != "" {
-		initOpts = append(initOpts, frankenphp.WithWorkers("workerName", testDataDir+opts.workerScript, opts.nbWorkers, opts.env, opts.watch))
+		initOpts = append(initOpts, frankenphp.WithWorkers("workerName", testDataDir+opts.workerScript, opts.nbWorkers, opts.env, opts.watch, 0))
 	}
 	initOpts = append(initOpts, opts.initOpts...)
 	if opts.phpIni != nil {

options.go

@@ -28,6 +28,7 @@ type workerOpt struct {
 	num      int
 	env      PreparedEnv
 	watch    []string
+	moduleID uint64
 }
 
 // WithNumThreads configures the number of PHP threads to start.
@@ -55,10 +56,10 @@ func WithMetrics(m Metrics) Option {
 	}
 }
 
-// WithWorkers configures the PHP workers to start.
-func WithWorkers(name string, fileName string, num int, env map[string]string, watch []string) Option {
+// WithWorkers configures the PHP workers to start, moduleID is used to identify the worker for a specific domain
+func WithWorkers(name string, fileName string, num int, env map[string]string, watch []string, moduleID uint64) Option {
 	return func(o *opt) error {
-		o.workers = append(o.workers, workerOpt{name, fileName, num, PrepareEnv(env), watch})
+		o.workers = append(o.workers, workerOpt{name, fileName, num, PrepareEnv(env), watch, moduleID})
 
 		return nil
 	}

phpmainthread_test.go

@@ -94,8 +94,8 @@ func TestTransitionThreadsWhileDoingRequests(t *testing.T) {
 
 	assert.NoError(t, Init(
 		WithNumThreads(numThreads),
-		WithWorkers(worker1Name, worker1Path, 1, map[string]string{"ENV1": "foo"}, []string{}),
-		WithWorkers(worker2Name, worker2Path, 1, map[string]string{"ENV1": "foo"}, []string{}),
+		WithWorkers(worker1Name, worker1Path, 1, map[string]string{"ENV1": "foo"}, []string{}, 0),
+		WithWorkers(worker2Name, worker2Path, 1, map[string]string{"ENV1": "foo"}, []string{}, 0),
 		WithLogger(zap.NewNop()),
 	))
 
@@ -179,7 +179,7 @@ func TestFinishBootingAWorkerScript(t *testing.T) {
 
 func getDummyWorker(fileName string) *worker {
 	if workers == nil {
-		workers = make(map[string]*worker)
+		workers = make(map[string][]*worker)
 	}
 	worker, _ := newWorker(workerOpt{
 		fileName: testDataPath + "/" + fileName,
@@ -211,9 +211,9 @@ func allPossibleTransitions(worker1Path string, worker2Path string) []func(*phpT
 				thread.boot()
 			}
 		},
-		func(thread *phpThread) { convertToWorkerThread(thread, workers[worker1Path]) },
+		func(thread *phpThread) { convertToWorkerThread(thread, workers[worker1Path][0]) },
 		convertToInactiveThread,
-		func(thread *phpThread) { convertToWorkerThread(thread, workers[worker2Path]) },
+		func(thread *phpThread) { convertToWorkerThread(thread, workers[worker2Path][0]) },
 		convertToInactiveThread,
 	}
 }

request_options.go

@@ -123,3 +123,12 @@ func WithRequestLogger(logger *zap.Logger) RequestOption {
 		return nil
 	}
 }
+
+// WithModuleID sets the module ID associated with the current request
+func WithModuleID(moduleID uint64) RequestOption {
+	return func(o *frankenPHPContext) error {
+		o.moduleID = moduleID
+
+		return nil
+	}
+}

scaling.go

@@ -130,6 +130,19 @@ func scaleRegularThread() {
 	autoScaledThreads = append(autoScaledThreads, thread)
 }
 
+func getWorker(fc *frankenPHPContext) *worker {
+	// if the request has been stalled long enough, scale
+	if workersList, ok := workers[fc.scriptFilename]; ok {
+		// Look for a worker with matching moduleID or a global worker (moduleID == 0)
+		for _, worker := range workersList {
+			if worker.moduleID == 0 || worker.moduleID == fc.moduleID {
+				return worker
+			}
+		}
+	}
+	return nil
+}
+
 func startUpscalingThreads(maxScaledThreads int, scale chan *frankenPHPContext, done chan struct{}) {
 	for {
 		scalingMu.Lock()
@@ -160,7 +173,7 @@ func startUpscalingThreads(maxScaledThreads int, scale chan *frankenPHPContext,
 			}
 
 			// if the request has been stalled long enough, scale
-			if worker, ok := workers[fc.scriptFilename]; ok {
+			if worker := getWorker(fc); worker != nil {
 				scaleWorkerThread(worker)
 			} else {
 				scaleRegularThread()